From e566e565436a2e7d5e2ea4a7824619f239418bfe Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E3=81=82=E3=83=BC=E3=82=8B?= Date: Mon, 2 Nov 2026 13:38:52 +0900 Subject: [PATCH] =?UTF-8?q?=E3=82=AD=E3=83=A5=E3=83=BC=E3=83=96=E3=82=92?= =?UTF-8?q?=E3=81=A4=E3=81=AA=E3=81=84=E3=81=A0=E3=83=96=E3=83=A9=E3=82=A6?= =?UTF-8?q?=E3=82=B6=E3=81=AE=E3=82=BF=E3=82=A4=E3=83=9E=E3=83=BC=E7=94=BB?= =?UTF-8?q?=E9=9D=A2=E3=82=92=E3=80=81=E3=81=BB=E3=81=8B=E3=81=AE=E3=83=96?= =?UTF-8?q?=E3=83=A9=E3=82=A6=E3=82=B6=EF=BC=88/live=EF=BC=89=E3=81=AB?= =?UTF-8?q?=E6=98=A0=E3=81=99?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit メイン（キューブをつないだタイマー画面）が回転・状態・タイマーを POST /api/live/events で送り、 見る側は GET /api/live/stream（SSE）で受け取って、自分用のキューブのストアとタイマーの表示に映す。 状態はサーバーのメモリにだけ持つ。 Co-Authored-By: Claude Opus 5.5 --- README.md | 1 + cmd/cubalyst/main.go | 5 +- docs/openapi.yaml | 88 ++++++++++ internal/live/hub.go | 228 +++++++++++++++++++++++++ internal/live/hub_test.go | 170 ++++++++++++++++++ internal/server/live.go | 81 +++++++++ internal/server/live_test.go | 84 +++++++++ internal/server/server.go | 13 +- web/src/App.tsx | 12 ++ web/src/components/ScrambleDisplay.tsx | 74 ++++---- web/src/components/TimerDisplay.tsx | 9 +- web/src/index.css | 15 ++ web/src/live/feed.ts | 116 +++++++++++++ web/src/live/hooks.ts | 37 ++++ web/src/live/live.test.ts | 168 ++++++++++++++++++ web/src/live/protocol.ts | 128 ++++++++++++++ web/src/live/publisher.ts | 159 +++++++++++++++++ web/src/pages/LivePage.tsx | 56 ++++++ web/src/pages/TimerPage.tsx | 3 + 19 files changed, 1406 insertions(+), 41 deletions(-) create mode 100644 internal/live/hub.go create mode 100644 internal/live/hub_test.go create mode 100644 internal/server/live.go create mode 100644 internal/server/live_test.go create mode 100644 web/src/live/feed.ts create mode 100644 web/src/live/hooks.ts create mode 100644 web/src/live/live.test.ts create mode 100644 web/src/live/protocol.ts create mode 100644 web/src/live/publisher.ts create mode 100644 web/src/pages/LivePage.tsx diff --git a/README.md b/README.md index c439e3e..91df42a 100644 --- a/README.md +++ b/README.md @@ -50,6 +50,7 @@ Cubalyst は記録を [MCP](https://modelcontextprotocol.io/) のツールとし - **持ち方に合わせた表記**：解法・再生・解析の記号を、自分の持ち方（下の色・前の色）に合わせて読み替えます - **持ち替えの記録**：ジャイロのあるキューブ（GAN i4 などの Gen4）なら、ソルブ中の持ち替え（y・x・z）も記録し、解法をその時点の持ち方で表示します。設定でオンにし、キューブ側のジャイロは GAN のアプリの省電力の設定で切り替えます（ジャイロは電池を多く使います） +- **中継**：キューブをつないだブラウザのタイマー画面（スクランブル・タイマー・キューブの状態・直前の段階）を、ほかのブラウザ（スマホなど）の「中継」（`/live`）に映します。見るだけで、記録も操作もしません。状態はサーバーのメモリにだけ持ちます - **クロス色**を設定でき、白以外のクロスでも段階を正しく分けます（設定は右上の ⚙ から開きます） - 記録は手元の SQLite ファイルに保存し、外部のサービスには送りません diff --git a/cmd/cubalyst/main.go b/cmd/cubalyst/main.go index 80656ed..f273825 100644 --- a/cmd/cubalyst/main.go +++ b/cmd/cubalyst/main.go @@ -54,11 +54,14 @@ func run() error { } log.Printf("db: %s (schema version %d)", cfg.DBPath, v) + app := server.New(database, cfg.StaticDir) srv := &http.Server{ Addr: cfg.Addr, - Handler: server.New(database, cfg.StaticDir).Handler(), + Handler: app.Handler(), ReadHeaderTimeout: 10 * time.Second, } + // 止めるとき、タイマー画面の中継（SSE）の接続を切る。切らないと Shutdown が終わるのを待ち続ける + srv.RegisterOnShutdown(app.Close) errCh := make(chan error, 1) go func() { log.Printf("listening on http://%s", cfg.Addr) diff --git a/docs/openapi.yaml b/docs/openapi.yaml index a4c27d2..ef31cce 100644 --- a/docs/openapi.yaml +++ b/docs/openapi.yaml @@ -28,6 +28,8 @@ tags: - name: 入出力 - name: 集計 description: 期間の練習の集計。文面にするのは呼ぶ側（通知の本文を作る kawaraban など） + - name: 中継 + description: キューブをつないだブラウザ（メイン）のタイマー画面を、ほかのブラウザ（`/live`）に映す。状態はサーバーのメモリにだけ持つ - name: その他 paths: @@ -467,6 +469,46 @@ paths: schema: { $ref: "#/components/schemas/PracticeStats" } "400": { $ref: "#/components/responses/BadRequest" } + /api/live/events: + post: + tags: [中継] + summary: メインのタイマー画面の出来事を送る + description: | + キューブをつないだブラウザ（メイン）が、タイマー画面の出来事を順に送る。見る側（`GET /api/live/stream`）にそのまま流す。 + 何もなくても15秒ごとに送り（`events` は空でよい）、40秒届かなければメインはいなくなったとみなす。 + メインは1つの前提で、2つのブラウザから送ると混ざる。 + operationId: publishLiveEvents + requestBody: + required: true + content: + application/json: + schema: + type: object + required: [events] + properties: + events: + type: array + items: { $ref: "#/components/schemas/LiveEvent" } + responses: + "204": { description: 受け取った } + "400": { $ref: "#/components/responses/BadRequest" } + + /api/live/stream: + get: + tags: [中継] + summary: メインのタイマー画面の状態を受け取り続ける（SSE） + description: | + Server-Sent Events。各出来事は `data:` の1行の JSON（LiveMessage）。 + 最初に今の状態（`snapshot`）を送り、そのあとメインから届いた出来事と、15秒ごとの `ping` を送る。 + 読むのが遅いと切るので、つなぎ直す（EventSource は自動でつなぎ直し、また `snapshot` から受け取る）。 + operationId: liveStream + responses: + "200": + description: 出来事の流れ + content: + text/event-stream: + schema: { $ref: "#/components/schemas/LiveMessage" } + components: parameters: period: @@ -517,6 +559,52 @@ components: schema: { $ref: "#/components/schemas/Error" } schemas: + LiveEvent: + type: object + description: | + メインから届く出来事。 + - main：キューブの接続の状態（cube） + - move：1手の回転（move）と、回したあとの状態（facelets） + - facelets：状態の置き換え（実機に合わせ直したときなど） + - session：タイマーの状態（LiveSession） + - bye：メインが画面を閉じた + required: [type] + properties: + type: { type: string, enum: [main, move, facelets, session, bye] } + cube: { type: string, enum: [connected, connecting, disconnected] } + move: { type: string, example: "R'" } + facelets: + type: string + description: キューブの状態。URFDLB の54文字（Kociemba の並び） + example: UUUUUUUUURRRRRRRRRFFFFFFFFFDDDDDDDDDLLLLLLLLLBBBBBBBBB + session: { $ref: "#/components/schemas/LiveSession" } + LiveMessage: + type: object + description: | + 見る側に送る出来事。LiveEvent の main・move・facelets・session をそのまま送るほか、 + - snapshot：見始めたときの今の状態（online・cube・facelets・session・age_ms） + - ping：メインが生きているか（online） + - main：bye のときは online=false で送る + required: [type] + properties: + type: { type: string, enum: [snapshot, ping, main, move, facelets, session] } + online: { type: boolean, description: メインがいるか } + cube: { type: string, enum: [connected, connecting, disconnected] } + move: { type: string } + facelets: { type: string } + session: { $ref: "#/components/schemas/LiveSession" } + age_ms: { type: integer, description: session を受け取ってからの時間（snapshot のときだけ）。経過時間をずらすのに使う } + LiveSession: + type: object + description: | + タイマーの状態。中身は画面（web/src/live/protocol.ts）が決め、サーバーは確かめずにそのまま流す。 + 時刻は端末ごとに基準が違うので、`*_ago_ms`（送った時点で、計測やインスペクションが始まってから何 ms たったか）で送る。 + additionalProperties: true + properties: + phase: { type: string, enum: [generating, scrambling, ready, running] } + scramble: { type: [string, "null"] } + started_ago_ms: { type: [integer, "null"] } + ready_ago_ms: { type: [integer, "null"] } PracticeStats: type: object required: [period, count, full, stages, days, practice] diff --git a/internal/live/hub.go b/internal/live/hub.go new file mode 100644 index 0000000..aa256ae --- /dev/null +++ b/internal/live/hub.go @@ -0,0 +1,228 @@ +// Package live は、キューブをつないだブラウザ（メイン）のタイマー画面の状態を、ほかのブラウザに中継する。 +// +// メインは出来事（回転・状態・タイマー）を POST で送り、見る側は SSE で受け取る。 +// 状態はメモリにだけ持ち、DB には書かない（サーバーを再起動すると消える）。 +package live + +import ( + "bytes" + "encoding/json" + "errors" + "fmt" + "strings" + "sync" + "time" +) + +// StaleAfter は、メインから何も届かないとき、メインがいなくなったとみなすまでの時間。 +// メインは何もなくても15秒ごとに送ってくる。 +const StaleAfter = 40 * time.Second + +// subscriberBuffer は、見る側1つに溜めておける出来事の数。溢れたら切り、つなぎ直してもらう（最初の状態から送り直す）。 +const subscriberBuffer = 256 + +// maxSession は、タイマーの状態（JSON）の大きさの上限。 +const maxSession = 64 << 10 + +// Event はメインから届く出来事。 +// +// - main：メインのキューブの接続の状態（cube）。何もなくても送り、生きていることを知らせる +// - move：1手の回転（move）と、回したあとの状態（facelets） +// - facelets：状態の置き換え（実機に合わせ直したときなど） +// - session：タイマーの状態（中身はアプリが決める JSON のオブジェクト） +// - bye：メインが画面を閉じた +type Event struct { + Type string `json:"type"` + Cube string `json:"cube,omitempty"` + Move string `json:"move,omitempty"` + Facelets string `json:"facelets,omitempty"` + Session json.RawMessage `json:"session,omitempty"` +} + +// Message は見る側に送る出来事。snapshot は見始めたときの今の状態、ping は生きているかの知らせ。 +type Message struct { + Type string `json:"type"` + Online *bool `json:"online,omitempty"` + Cube string `json:"cube,omitempty"` + Move string `json:"move,omitempty"` + Facelets string `json:"facelets,omitempty"` + Session json.RawMessage `json:"session,omitempty"` + // AgeMS は session を受け取ってからの時間（snapshot のときだけ）。タイマーの経過時間をずらすのに使う。 + AgeMS int64 `json:"age_ms,omitempty"` +} + +// Hub はメインの最新の状態と、見る側の一覧を持つ。 +type Hub struct { + now func() time.Time + + mu sync.Mutex + lastSeen time.Time + bye bool + cube string + facelets string + session json.RawMessage + sessionAt time.Time + subs map[chan []byte]struct{} + closed bool +} + +// NewHub は Hub を作る。 +func NewHub() *Hub { + return &Hub{now: time.Now, subs: map[chan []byte]struct{}{}} +} + +// ErrInvalid は、メインから届いた出来事が正しくないときのエラー。 +var ErrInvalid = errors.New("invalid live event") + +func invalid(format string, a ...any) error { + return fmt.Errorf("%w: %s", ErrInvalid, fmt.Sprintf(format, a...)) +} + +// Validate は出来事の中身を確かめる。 +func (e Event) Validate() error { + switch e.Type { + case "main": + switch e.Cube { + case "connected", "connecting", "disconnected": + default: + return invalid("cube must be connected, connecting or disconnected") + } + case "move": + if e.Move == "" || len(e.Move) > 4 { + return invalid("move is required") + } + if err := validFacelets(e.Facelets); err != nil { + return err + } + case "facelets": + if err := validFacelets(e.Facelets); err != nil { + return err + } + case "session": + if len(e.Session) > maxSession { + return invalid("session is too large") + } + if t := bytes.TrimSpace(e.Session); len(t) == 0 || t[0] != '{' || !json.Valid(t) { + return invalid("session must be a JSON object") + } + case "bye": + default: + return invalid("unknown type %q", e.Type) + } + return nil +} + +func validFacelets(f string) error { + if len(f) != 54 || strings.Trim(f, "URFDLB") != "" { + return invalid("facelets must be 54 letters of URFDLB") + } + return nil +} + +// Publish はメインから届いた出来事を、状態に反映して見る側に送る。先に Validate で確かめておく。 +func (h *Hub) Publish(events []Event) { + h.mu.Lock() + defer h.mu.Unlock() + now := h.now() + h.lastSeen = now + h.bye = false + for _, e := range events { + msg := Message{Type: e.Type} + switch e.Type { + case "main": + h.cube = e.Cube + msg.Online, msg.Cube = boolPtr(true), e.Cube + case "move", "facelets": + h.facelets = e.Facelets + msg.Move, msg.Facelets = e.Move, e.Facelets + case "session": + h.session, h.sessionAt = compact(e.Session), now + msg.Session = h.session + case "bye": + h.bye = true + msg = Message{Type: "main", Online: boolPtr(false), Cube: h.cube} + } + h.broadcast(msg) + } +} + +// Subscribe は見る側を足し、最初に送る今の状態と、続く出来事を受け取るチャネルを返す。 +// チャネルが閉じたら（溢れたとき・Close のとき）接続を終える。終えるときは unsubscribe を呼ぶ。 +func (h *Hub) Subscribe() (snapshot []byte, ch <-chan []byte, unsubscribe func()) { + h.mu.Lock() + defer h.mu.Unlock() + c := make(chan []byte, subscriberBuffer) + if h.closed { + close(c) + return nil, c, func() {} + } + h.subs[c] = struct{}{} + now := h.now() + snap := Message{Type: "snapshot", Online: boolPtr(h.online(now)), Cube: h.cube, Facelets: h.facelets, Session: h.session} + if h.session != nil { + snap.AgeMS = max(1, now.Sub(h.sessionAt).Milliseconds()) + } + return marshal(snap), c, func() { + h.mu.Lock() + defer h.mu.Unlock() + if _, ok := h.subs[c]; ok { + delete(h.subs, c) + close(c) + } + } +} + +// Ping は、メインが生きているかの知らせ（見る側に定期的に送る）。 +func (h *Hub) Ping() []byte { + h.mu.Lock() + defer h.mu.Unlock() + return marshal(Message{Type: "ping", Online: boolPtr(h.online(h.now()))}) +} + +// Close は見る側をすべて切る（サーバーを止めるとき）。 +func (h *Hub) Close() { + h.mu.Lock() + defer h.mu.Unlock() + h.closed = true + for c := range h.subs { + delete(h.subs, c) + close(c) + } +} + +func (h *Hub) online(now time.Time) bool { + return !h.lastSeen.IsZero() && !h.bye && now.Sub(h.lastSeen) < StaleAfter +} + +// broadcast は h.mu を持ったまま呼ぶ。 +func (h *Hub) broadcast(msg Message) { + b := marshal(msg) + for c := range h.subs { + select { + case c <- b: + default: + // 読むのが遅い見る側は切る（つなぎ直すと、今の状態から送り直す） + delete(h.subs, c) + close(c) + } + } +} + +func marshal(m Message) []byte { + b, err := json.Marshal(m) + if err != nil { + panic(err) // Message は必ず JSON にできる + } + return b +} + +// compact は JSON を1行にする（SSE の data は1行で送る）。 +func compact(raw json.RawMessage) json.RawMessage { + var buf bytes.Buffer + if err := json.Compact(&buf, raw); err != nil { + return raw + } + return buf.Bytes() +} + +func boolPtr(b bool) *bool { return &b } diff --git a/internal/live/hub_test.go b/internal/live/hub_test.go new file mode 100644 index 0000000..46f1e93 --- /dev/null +++ b/internal/live/hub_test.go @@ -0,0 +1,170 @@ +package live + +import ( + "encoding/json" + "strings" + "testing" + "time" +) + +const solved = "UUUUUUUUURRRRRRRRRFFFFFFFFFDDDDDDDDDLLLLLLLLLBBBBBBBBB" + +func newTestHub() (*Hub, *time.Time) { + now := time.Date(2026, 11, 2, 12, 0, 0, 0, time.UTC) + h := NewHub() + h.now = func() time.Time { return now } + return h, &now +} + +func decode(t *testing.T, b []byte) Message { + t.Helper() + var m Message + if err := json.Unmarshal(b, &m); err != nil { + t.Fatalf("%s: %v", b, err) + } + return m +} + +func recv(t *testing.T, ch <-chan []byte) Message { + t.Helper() + select { + case b, ok := <-ch: + if !ok { + t.Fatal("チャネルが閉じた") + } + return decode(t, b) + default: + t.Fatal("出来事が届いていない") + } + return Message{} +} + +func TestSnapshotBeforeMain(t *testing.T) { + h, _ := newTestHub() + snap, _, unsub := h.Subscribe() + defer unsub() + m := decode(t, snap) + if m.Type != "snapshot" || m.Online == nil || *m.Online || m.Session != nil || m.AgeMS != 0 { + t.Fatalf("%+v", m) + } +} + +func TestPublishAndSnapshot(t *testing.T) { + h, now := newTestHub() + _, ch, unsub := h.Subscribe() + defer unsub() + + h.Publish([]Event{ + {Type: "main", Cube: "connected"}, + {Type: "move", Move: "R", Facelets: solved}, + {Type: "session", Session: json.RawMessage("{\n \"phase\": \"running\"\n}")}, + }) + if m := recv(t, ch); m.Type != "main" || !*m.Online || m.Cube != "connected" { + t.Fatalf("main: %+v", m) + } + if m := recv(t, ch); m.Type != "move" || m.Move != "R" || m.Facelets != solved { + t.Fatalf("move: %+v", m) + } + if m := recv(t, ch); m.Type != "session" || string(m.Session) != `{"phase":"running"}` || m.AgeMS != 0 { + t.Fatalf("session: %+v", m) + } + + // あとから見始めた人には、今の状態と、タイマーの状態を受け取ってからの時間を送る + *now = now.Add(1500 * time.Millisecond) + snap, _, unsub2 := h.Subscribe() + defer unsub2() + m := decode(t, snap) + if !*m.Online || m.Cube != "connected" || m.Facelets != solved || string(m.Session) != `{"phase":"running"}` || m.AgeMS != 1500 { + t.Fatalf("snapshot: %+v", m) + } +} + +func TestOnline(t *testing.T) { + h, now := newTestHub() + _, ch, unsub := h.Subscribe() + defer unsub() + h.Publish([]Event{{Type: "main", Cube: "connected"}}) + recv(t, ch) + + *now = now.Add(StaleAfter - time.Second) + if m := decode(t, h.Ping()); !*m.Online { + t.Fatal("まだ生きているはず") + } + *now = now.Add(2 * time.Second) + if m := decode(t, h.Ping()); *m.Online { + t.Fatal("届かなくなったら、いなくなったとみなす") + } + + // 何も無い POST（生きている知らせ）で戻る + h.Publish(nil) + if m := decode(t, h.Ping()); !*m.Online { + t.Fatal("届いたら戻る") + } + + h.Publish([]Event{{Type: "bye"}}) + if m := recv(t, ch); m.Type != "main" || *m.Online { + t.Fatalf("bye: %+v", m) + } + if m := decode(t, h.Ping()); *m.Online { + t.Fatal("閉じたらいない") + } +} + +func TestSlowSubscriberIsDropped(t *testing.T) { + h, _ := newTestHub() + _, ch, unsub := h.Subscribe() + defer unsub() + for range subscriberBuffer + 1 { + h.Publish([]Event{{Type: "move", Move: "R", Facelets: solved}}) + } + n := 0 + for range ch { + n++ + } + if n != subscriberBuffer { + t.Fatalf("got %d", n) + } +} + +func TestClose(t *testing.T) { + h, _ := newTestHub() + _, ch, unsub := h.Subscribe() + h.Close() + if _, ok := <-ch; ok { + t.Fatal("閉じるはず") + } + unsub() // 二重に閉じない + if snap, _, _ := h.Subscribe(); snap != nil { + t.Fatal("閉じたあとは受け付けない") + } +} + +func TestValidate(t *testing.T) { + ok := []Event{ + {Type: "main", Cube: "disconnected"}, + {Type: "move", Move: "U'", Facelets: solved}, + {Type: "facelets", Facelets: solved}, + {Type: "session", Session: json.RawMessage(`{}`)}, + {Type: "bye"}, + } + for _, e := range ok { + if err := e.Validate(); err != nil { + t.Errorf("%+v: %v", e, err) + } + } + bad := []Event{ + {Type: "nope"}, + {Type: "main", Cube: "maybe"}, + {Type: "move", Facelets: solved}, + {Type: "move", Move: "R", Facelets: "UUU"}, + {Type: "facelets", Facelets: strings.Repeat("X", 54)}, + {Type: "session", Session: json.RawMessage(`[1]`)}, + {Type: "session"}, + {Type: "session", Session: json.RawMessage(`{"a":"` + strings.Repeat("x", maxSession) + `"}`)}, + } + for _, e := range bad { + if err := e.Validate(); err == nil { + t.Errorf("%+v: エラーになるはず", e) + } + } +} diff --git a/internal/server/live.go b/internal/server/live.go new file mode 100644 index 0000000..32c4af8 --- /dev/null +++ b/internal/server/live.go @@ -0,0 +1,81 @@ +package server + +import ( + "encoding/json" + "fmt" + "io" + "net/http" + "time" + + "github.com/riwn/cubalyst/internal/live" +) + +// livePing は、見る側に生きているかを知らせる間隔（プロキシに切られないためでもある）。 +var livePing = 15 * time.Second + +type liveEventsRequest struct { + Events []live.Event `json:"events"` +} + +// handleLiveEvents は、メイン（キューブをつないだブラウザ）からタイマー画面の出来事を受け取る。 +func (s *Server) handleLiveEvents(w http.ResponseWriter, r *http.Request) { + var in liveEventsRequest + if err := json.NewDecoder(io.LimitReader(r.Body, 1<<20)).Decode(&in); err != nil { + writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid json: " + err.Error()}) + return + } + for i, e := range in.Events { + if err := e.Validate(); err != nil { + writeJSON(w, http.StatusBadRequest, map[string]string{"error": fmt.Sprintf("events[%d]: %v", i, err)}) + return + } + } + s.live.Publish(in.Events) + w.WriteHeader(http.StatusNoContent) +} + +// handleLiveStream は、見る側にメインの状態を SSE で送り続ける。最初に今の状態（snapshot）を送る。 +func (s *Server) handleLiveStream(w http.ResponseWriter, r *http.Request) { + flusher, ok := w.(http.Flusher) + if !ok { + writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "streaming is not supported"}) + return + } + snapshot, ch, unsubscribe := s.live.Subscribe() + defer unsubscribe() + if snapshot == nil { + writeJSON(w, http.StatusServiceUnavailable, map[string]string{"error": "shutting down"}) + return + } + h := w.Header() + h.Set("Content-Type", "text/event-stream") + h.Set("Cache-Control", "no-cache") + h.Set("X-Accel-Buffering", "no") + w.WriteHeader(http.StatusOK) + send := func(b []byte) bool { + if _, err := fmt.Fprintf(w, "data: %s\n\n", b); err != nil { + return false + } + flusher.Flush() + return true + } + if !send(snapshot) { + return + } + ticker := time.NewTicker(livePing) + defer ticker.Stop() + for { + select { + case <-r.Context().Done(): + return + case b, ok := <-ch: + if !ok || !send(b) { + return + } + case <-ticker.C: + if !send(s.live.Ping()) { + return + } + } + } +} diff --git a/internal/server/live_test.go b/internal/server/live_test.go new file mode 100644 index 0000000..d586b10 --- /dev/null +++ b/internal/server/live_test.go @@ -0,0 +1,84 @@ +package server + +import ( + "bufio" + "encoding/json" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" +) + +const solvedFacelets = "UUUUUUUUURRRRRRRRRFFFFFFFFFDDDDDDDDDLLLLLLLLLBBBBBBBBB" + +func TestLiveEventsRejectsInvalid(t *testing.T) { + h := newTestServer(t, t.TempDir()) + for _, body := range []string{`{`, `{"events":[{"type":"nope"}]}`, `{"events":[{"type":"move","move":"R"}]}`} { + if rec := do(t, h, http.MethodPost, "/api/live/events", body); rec.Code != http.StatusBadRequest { + t.Errorf("%s: status=%d", body, rec.Code) + } + } + if rec := do(t, h, http.MethodPost, "/api/live/events", `{"events":[]}`); rec.Code != http.StatusNoContent { + t.Errorf("空：status=%d %s", rec.Code, rec.Body) + } +} + +func TestLiveStream(t *testing.T) { + ts := httptest.NewServer(newTestServer(t, t.TempDir())) + defer ts.Close() + + res, err := http.Get(ts.URL + "/api/live/stream") + if err != nil { + t.Fatal(err) + } + defer res.Body.Close() + if ct := res.Header.Get("Content-Type"); ct != "text/event-stream" { + t.Fatalf("Content-Type=%q", ct) + } + lines := make(chan string) + go func() { + sc := bufio.NewScanner(res.Body) + for sc.Scan() { + if l, ok := strings.CutPrefix(sc.Text(), "data: "); ok { + lines <- l + } + } + close(lines) + }() + next := func() map[string]any { + t.Helper() + select { + case l := <-lines: + var m map[string]any + if err := json.Unmarshal([]byte(l), &m); err != nil { + t.Fatalf("%s: %v", l, err) + } + return m + case <-time.After(5 * time.Second): + t.Fatal("届かない") + } + return nil + } + + if m := next(); m["type"] != "snapshot" || m["online"] != false { + t.Fatalf("snapshot: %v", m) + } + body := `{"events":[{"type":"main","cube":"connected"},{"type":"move","move":"R","facelets":"` + solvedFacelets + `"}]}` + req, _ := http.NewRequest(http.MethodPost, ts.URL+"/api/live/events", strings.NewReader(body)) + req.Header.Set("Content-Type", "application/json") + pr, err := http.DefaultClient.Do(req) + if err != nil { + t.Fatal(err) + } + pr.Body.Close() + if pr.StatusCode != http.StatusNoContent { + t.Fatalf("POST status=%d", pr.StatusCode) + } + if m := next(); m["type"] != "main" || m["online"] != true { + t.Fatalf("main: %v", m) + } + if m := next(); m["type"] != "move" || m["move"] != "R" { + t.Fatalf("move: %v", m) + } +} diff --git a/internal/server/server.go b/internal/server/server.go index fb8547b..7200938 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -16,6 +16,7 @@ import ( "github.com/riwn/cubalyst/docs" "github.com/riwn/cubalyst/internal/analysis" "github.com/riwn/cubalyst/internal/db" + "github.com/riwn/cubalyst/internal/live" "github.com/riwn/cubalyst/internal/settings" "github.com/riwn/cubalyst/internal/solves" ) @@ -27,6 +28,7 @@ type Server struct { solves *solves.Store settings *settings.Store analysis *analysis.Service + live *live.Hub } // New はサーバーを作る。 @@ -37,9 +39,15 @@ func New(database *sql.DB, staticDir string) *Server { solves: solves.NewStore(database), settings: settings.NewStore(database), analysis: analysis.NewService(database), + live: live.NewHub(), } } +// Close は、つなぎっぱなしの接続（タイマー画面の中継）を切る。http.Server の RegisterOnShutdown に渡す。 +func (s *Server) Close() { + s.live.Close() +} + // Handler はルーティング済みのハンドラを返す。 func (s *Server) Handler() http.Handler { mux := http.NewServeMux() @@ -66,6 +74,8 @@ func (s *Server) Handler() http.Handler { mux.HandleFunc("GET /api/analysis-input", s.handleMultiAnalysisInput) mux.HandleFunc("GET /api/stage-cases", s.handleStageCases) mux.HandleFunc("GET /api/stats/practice", s.handlePracticeStats) + mux.HandleFunc("POST /api/live/events", s.handleLiveEvents) + mux.HandleFunc("GET /api/live/stream", s.handleLiveStream) mux.HandleFunc("/api/", func(w http.ResponseWriter, r *http.Request) { writeJSON(w, http.StatusNotFound, map[string]string{"error": "not found"}) }) @@ -140,7 +150,8 @@ func logRequests(next http.Handler) http.Handler { return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { start := time.Now() next.ServeHTTP(w, r) - if strings.HasPrefix(r.URL.Path, "/api/") { + // タイマー画面の中継は1手ごとに届くので、ログに残さない + if strings.HasPrefix(r.URL.Path, "/api/") && r.URL.Path != "/api/live/events" { log.Printf("%s %s (%s)", r.Method, r.URL.Path, time.Since(start).Round(time.Microsecond)) } }) diff --git a/web/src/App.tsx b/web/src/App.tsx index 0f3b959..b93027b 100644 --- a/web/src/App.tsx +++ b/web/src/App.tsx @@ -5,6 +5,7 @@ import { CubeConnectionControl, CubeNotices } from './components/ConnectionPanel import { CubeStoreProvider } from './cube/CubeStoreProvider' import { AnalysesPage, AnalysisDetailPage } from './pages/AnalysesPage' import { HistoryPage } from './pages/HistoryPage' +import { LivePage } from './pages/LivePage' import { PracticePage } from './pages/PracticePage' import { SettingsPage } from './pages/SettingsPage' import { SolveDetailPage } from './pages/SolveDetailPage' @@ -37,6 +38,16 @@ const NAV: { to: string; label: string; icon: ReactNode; end?: boolean; match?: ), }, + { + to: '/live', + label: '中継', + icon: ( + <> + + + + ), + }, { to: '/stats', label: '解析', @@ -144,6 +155,7 @@ export default function App() { } /> } /> + } /> } /> } /> } /> diff --git a/web/src/components/ScrambleDisplay.tsx b/web/src/components/ScrambleDisplay.tsx index c480153..50c0339 100644 --- a/web/src/components/ScrambleDisplay.tsx +++ b/web/src/components/ScrambleDisplay.tsx @@ -1,8 +1,8 @@ import type { SessionState, SolveSession } from '../solve/session' import { KebabMenu } from '../ui/Popover' -/** スクランブル手順と進捗。間違えたときは戻す手順を表示する。 */ -export function ScrambleDisplay({ session, state }: { session: SolveSession; state: SessionState }) { +/** スクランブル手順と進捗。間違えたときは戻す手順を表示する。session を省くと（中継の画面）、操作のボタンを出さない。 */ +export function ScrambleDisplay({ session, state }: { session?: SolveSession; state: SessionState }) { const { phase, progress, exercise } = state return ( @@ -36,40 +36,42 @@ export function ScrambleDisplay({ session, state }: { session: SolveSession; sta

)} -
- {phase !== 'running' && ( - - )} - {phase === 'scrambling' && ( - session.forceReady(), - }, - ]} - /> - )} - {phase === 'running' && } -
+ {session && ( +
+ {phase !== 'running' && ( + + )} + {phase === 'scrambling' && ( + session.forceReady(), + }, + ]} + /> + )} + {phase === 'running' && } +
+ )} ) } diff --git a/web/src/components/TimerDisplay.tsx b/web/src/components/TimerDisplay.tsx index a7be7fb..d49f157 100644 --- a/web/src/components/TimerDisplay.tsx +++ b/web/src/components/TimerDisplay.tsx @@ -2,8 +2,11 @@ import { useEffect, useState } from 'react' import { INSPECTION_DNF_MS, INSPECTION_MS, type SessionState } from '../solve/session' import { effectiveTime, formatTime } from '../solve/stats' -/** 大きなタイマー表示。計測中は経過時間、それ以外は直前の記録。 */ -export function TimerDisplay({ state }: { state: SessionState }) { +/** + * 大きなタイマー表示。計測中は経過時間、それ以外は直前の記録。 + * lastMoveCount は直前の記録の手数（回転ログを持たない中継の画面で渡す。省くと回転ログの長さ）。 + */ +export function TimerDisplay({ state, lastMoveCount }: { state: SessionState; lastMoveCount?: number }) { const { phase, startedAt, last, moveCount, readyAt, inspection } = state const inspecting = phase === 'ready' && inspection && readyAt != null const [now, setNow] = useState(() => performance.now()) @@ -47,7 +50,7 @@ export function TimerDisplay({ state }: { state: SessionState }) { : phase === 'scrambling' ? 'スクランブル中' : last - ? `${last.moves?.length ?? 0}手` + ? `${lastMoveCount ?? last.moves?.length ?? 0}手` : '' } return ( diff --git a/web/src/index.css b/web/src/index.css index 85f9735..407e1e7 100644 --- a/web/src/index.css +++ b/web/src/index.css @@ -599,6 +599,21 @@ code { text-align: center; } +/* 中継：メインを映しているかの一行。映している間は緑の点を付ける */ +.live-status { + margin: 8px 0 0; +} + +.live-status--on::before { + content: ''; + display: inline-block; + width: 0.6em; + height: 0.6em; + margin-right: 0.4em; + border-radius: 50%; + background: var(--ok); +} + /* スクランブル：画面上部に左寄せ。済んだ手は薄く、次の手は黄面の下線 */ .scramble { margin: 8px 0 0; diff --git a/web/src/live/feed.ts b/web/src/live/feed.ts new file mode 100644 index 0000000..8a0e09d --- /dev/null +++ b/web/src/live/feed.ts @@ -0,0 +1,116 @@ +// 見る側：中継のサーバー（SSE）から受け取った出来事を、この端末のキューブのストアとタイマーの状態に映す。 + +import type { CubeStore, ConnectionStatus } from '../cube/store' +import type { SessionState } from '../solve/session' +import { fromLiveSession, type LiveMessage } from './protocol' + +const ENDPOINT = '/api/live/stream' + +export type LiveView = { + /** サーバーとの接続。error のときは EventSource がつなぎ直している。 */ + stream: 'connecting' | 'open' | 'error' + /** メイン（キューブをつないだタイマー画面）がいるか。 */ + online: boolean + /** メインのキューブの接続の状態。 */ + cube: ConnectionStatus | null + /** メインのタイマーの状態（この端末の時計に置き直したもの）。 */ + session: SessionState | null + /** 直前のソルブの手数。 */ + lastMoveCount: number +} + +export class LiveFeed { + private view: LiveView = { + stream: 'connecting', + online: false, + cube: null, + session: null, + lastMoveCount: 0, + } + private listeners = new Set<() => void>() + private source: EventSource | null = null + private readonly store: CubeStore + + constructor(store: CubeStore) { + this.store = store + } + + subscribe = (listener: () => void): (() => void) => { + this.listeners.add(listener) + return () => { + this.listeners.delete(listener) + } + } + + getSnapshot = (): LiveView => this.view + + private set(patch: Partial) { + this.view = { ...this.view, ...patch } + for (const l of this.listeners) l() + } + + start() { + if (this.source) return + const source = new EventSource(ENDPOINT) + this.source = source + source.onopen = () => this.set({ stream: 'open' }) + source.onerror = () => this.set({ stream: 'error' }) + source.onmessage = (e: MessageEvent) => { + let msg: LiveMessage + try { + msg = JSON.parse(e.data) as LiveMessage + } catch { + return + } + this.handle(msg, performance.now()) + } + } + + stop() { + this.source?.close() + this.source = null + } + + /** 1つの出来事を映す。now は受け取った時刻（performance.now() 基準）。 */ + handle(msg: LiveMessage, now: number) { + switch (msg.type) { + case 'snapshot': + if (msg.facelets) this.store.handleEvent({ type: 'facelets', facelets: msg.facelets }) + this.set({ + online: msg.online, + cube: msg.cube ?? null, + ...(msg.session ? this.sessionPatch(msg.session, now - (msg.age_ms ?? 0)) : { session: null }), + }) + break + case 'ping': + if (msg.online !== this.view.online) this.set({ online: msg.online }) + break + case 'main': + this.set({ online: msg.online, cube: msg.cube ?? this.view.cube }) + break + case 'move': + this.store.handleEvent({ + type: 'move', + move: msg.move, + localTimestamp: null, + cubeTimestamp: null, + }) + // 回したあとの状態と比べ、ずれていたら合わせ直す（途中の出来事を取りこぼしたときなど） + this.store.handleEvent({ type: 'facelets', facelets: msg.facelets }) + break + case 'facelets': + this.store.handleEvent({ type: 'facelets', facelets: msg.facelets }) + break + case 'session': + this.set(this.sessionPatch(msg.session, now)) + break + } + } + + private sessionPatch(s: Parameters[0], now: number): Partial { + return { + session: fromLiveSession(s, now), + lastMoveCount: s.last?.move_count ?? 0, + } + } +} diff --git a/web/src/live/hooks.ts b/web/src/live/hooks.ts new file mode 100644 index 0000000..d2e2a62 --- /dev/null +++ b/web/src/live/hooks.ts @@ -0,0 +1,37 @@ +import { useEffect, useRef, useState, useSyncExternalStore } from 'react' +import type { NewSolve } from '../api' +import { useCubeStore } from '../cube/hooks' +import type { CubeStore } from '../cube/store' +import { useSettings } from '../settings/hooks' +import { withSplits, type Split } from '../solve/cfop' +import type { SolveSession } from '../solve/session' +import { LiveFeed, type LiveView } from './feed' +import { startLivePublish } from './publisher' + +/** on の間、タイマー画面の状態を中継のサーバーへ送る（キューブがつながっている間だけ送る）。 */ +export function useLivePublish(session: SolveSession | null, on: boolean) { + const store = useCubeStore() + const { settings } = useSettings() + const crossColor = useRef(settings.cross_color) + useEffect(() => { + crossColor.current = settings.cross_color + }) + useEffect(() => { + if (!session || !on) return + const cache = new WeakMap() + return startLivePublish(store, session, (last) => { + if (!cache.has(last)) cache.set(last, withSplits(store.kpuzzle, last, crossColor.current).splits) + return cache.get(last) + }) + }, [store, session, on]) +} + +/** 中継を受け取り、store（見る側だけのキューブのストア）に映す。 */ +export function useLiveFeed(store: CubeStore): LiveView { + const [feed] = useState(() => new LiveFeed(store)) + useEffect(() => { + feed.start() + return () => feed.stop() + }, [feed]) + return useSyncExternalStore(feed.subscribe, feed.getSnapshot) +} diff --git a/web/src/live/live.test.ts b/web/src/live/live.test.ts new file mode 100644 index 0000000..f57628a --- /dev/null +++ b/web/src/live/live.test.ts @@ -0,0 +1,168 @@ +import { cube3x3x3 } from 'cubing/puzzles' +import { describe, expect, it } from 'vitest' +import type { NewSolve } from '../api' +import { MockConnection } from '../cube/mock' +import { CubeStore } from '../cube/store' +import { SolveSession, isSolved } from '../solve/session' +import { LiveFeed } from './feed' +import type { LiveEvent, LiveMessage } from './protocol' +import { LiveQueue, startLivePublish } from './publisher' + +const kpuzzle = await cube3x3x3.kpuzzle() +const flush = () => new Promise((r) => setTimeout(r, 0)) + +/** 送るたびに呼び出しを残し、resolve するまで返さない POST。 */ +function manualPost() { + const calls: { events: LiveEvent[]; resolve: () => void; reject: () => void }[] = [] + const post = (events: LiveEvent[]) => + new Promise((resolve, reject) => { + calls.push({ events, resolve, reject: () => reject(new Error('fail')) }) + }) + return { calls, post } +} + +const move = (m: string): LiveEvent => ({ type: 'move', move: m, facelets: 'x' }) +const session = (n: number): LiveEvent => ({ type: 'session', session: { move_count: n } as never }) + +describe('LiveQueue', () => { + it('sends one request at a time and batches what piles up', async () => { + const { calls, post } = manualPost() + const q = new LiveQueue(() => [], post) + q.push(move('R')) + q.push(move('U')) + q.push(session(1)) + q.push(session(2)) + expect(calls.map((c) => c.events)).toEqual([[move('R')]]) + calls[0].resolve() + await flush() + // タイマーの状態は最新の1つだけ + expect(calls[1].events).toEqual([move('U'), session(2)]) + }) + + it('resends the whole state after a failure', async () => { + const { calls, post } = manualPost() + const snap: LiveEvent[] = [{ type: 'main', cube: 'connected' }] + const q = new LiveQueue(() => snap, post) + q.push(move('R')) + calls[0].reject() + await flush() + q.push(move('U')) + expect(calls[1].events).toEqual(snap) + calls[1].resolve() + await flush() + q.push(move('F')) + expect(calls[2].events).toEqual([move('F')]) + }) + + it('sends a heartbeat when idle and bye on stop', async () => { + const { calls, post } = manualPost() + const q = new LiveQueue(() => [], post) + q.heartbeat() + expect(calls[0].events).toEqual([]) + calls[0].resolve() + await flush() + q.stop() + q.push(move('R')) + expect(calls.slice(1).map((c) => c.events)).toEqual([[{ type: 'bye' }]]) + }) +}) + +/** サーバー（internal/live）と同じように、メインの出来事を見る側の出来事にする。 */ +function relay(e: LiveEvent): LiveMessage { + switch (e.type) { + case 'main': + return { type: 'main', online: true, cube: e.cube } + case 'bye': + return { type: 'main', online: false } + default: + return e + } +} + +describe('mirroring the timer to another browser', () => { + it('shows the same cube and timer as the main', async () => { + const main = new CubeStore(kpuzzle) + const mock = new MockConnection(kpuzzle) + main.attach(mock) + const solves: NewSolve[] = [] + const scrambles = ["R U R'"] + const timer = new SolveSession( + main, + async () => ({ mode: 'full', scramble: scrambles.shift() ?? 'F2', isGoal: isSolved }), + (s) => solves.push(s), + ) + await timer.next() + + const viewer = new CubeStore(kpuzzle) + const feed = new LiveFeed(viewer) + feed.handle({ type: 'snapshot', online: false }, 0) + const stop = startLivePublish( + main, + timer, + () => undefined, + async (events) => { + events.forEach((e) => feed.handle(relay(e), performance.now())) + }, + ) + await flush() + expect(feed.getSnapshot()).toMatchObject({ online: true, cube: 'connected' }) + expect(feed.getSnapshot().session).toMatchObject({ phase: 'scrambling', exercise: { scramble: "R U R'" } }) + + for (const m of ['R', 'U', "R'"]) { + mock.turn(m) + await flush() + } + expect(viewer.getSnapshot().facelets).toBe(main.getSnapshot().facelets) + expect(feed.getSnapshot().session?.phase).toBe('ready') + + mock.turn('R') + await flush() + const running = feed.getSnapshot().session! + expect(running).toMatchObject({ phase: 'running', moveCount: 1 }) + // 計測を始めた時刻は、この端末の時計に置き直してある + expect(Math.abs(running.startedAt! - timer.getSnapshot().startedAt!)).toBeLessThan(50) + + for (const m of ["U'", "R'"]) { + mock.turn(m) + await flush() + } + expect(solves).toHaveLength(1) + await flush() + expect(viewer.getSnapshot().facelets).toBe(main.getSnapshot().facelets) + const after = feed.getSnapshot() + expect(after.session?.last).toMatchObject({ total_ms: solves[0].total_ms, solved_at: solves[0].solved_at }) + expect(after.lastMoveCount).toBe(3) + + stop() + await flush() + expect(feed.getSnapshot().online).toBe(false) + timer.dispose() + }) + + it('shifts the timer by the age of a snapshot', () => { + const feed = new LiveFeed(new CubeStore(kpuzzle)) + feed.handle( + { + type: 'snapshot', + online: true, + cube: 'connected', + session: { + phase: 'running', + scramble: 'R', + label: null, + progress: null, + inspection: true, + inspection_penalty: '', + move_count: 5, + started_ago_ms: 2000, + ready_ago_ms: 5000, + last: null, + error: null, + }, + age_ms: 1000, + }, + 10_000, + ) + expect(feed.getSnapshot().session).toMatchObject({ startedAt: 7000, readyAt: 4000, moveCount: 5 }) + }) +}) diff --git a/web/src/live/protocol.ts b/web/src/live/protocol.ts new file mode 100644 index 0000000..a65886a --- /dev/null +++ b/web/src/live/protocol.ts @@ -0,0 +1,128 @@ +// タイマー画面の中継（/api/live）でやり取りする形。サーバーは session の中身を見ずにそのまま流す。 +// +// 時刻（performance.now()）は端末ごとに基準が違うので、「送った時点で何 ms たったか」（*_ago_ms）にして送り、 +// 受け取った側が自分の時計に置き直す。 + +import type { NewSolve } from '../api' +import type { ConnectionStatus } from '../cube/store' +import type { Split } from '../solve/cfop' +import type { ScrambleProgress } from '../solve/scramble-tracker' +import type { SessionPhase, SessionState } from '../solve/session' +import type { Penalty } from '../solve/stats' + +/** 直前のソルブ。回転ログやジャイロは送らず、表示に使う分だけ。 */ +export type LiveLast = { + solved_at: string + total_ms: number + penalty: Penalty + move_count: number + splits?: Split[] +} + +/** タイマーの状態。 */ +export type LiveSession = { + phase: SessionPhase + scramble: string | null + label: string | null + progress: ScrambleProgress | null + inspection: boolean + inspection_penalty: Penalty + move_count: number + /** 計測を始めてから送るまでの時間。計測中でなければ null。 */ + started_ago_ms: number | null + /** 計測待ち（インスペクションの起点）になってから送るまでの時間。 */ + ready_ago_ms: number | null + last: LiveLast | null + error: string | null +} + +/** メインが送る出来事。 */ +export type LiveEvent = + | { type: 'main'; cube: ConnectionStatus } + | { type: 'move'; move: string; facelets: string } + | { type: 'facelets'; facelets: string } + | { type: 'session'; session: LiveSession } + | { type: 'bye' } + +/** 見る側が受け取る出来事。 */ +export type LiveMessage = + | { + type: 'snapshot' + online: boolean + cube?: ConnectionStatus + facelets?: string + session?: LiveSession + /** session をサーバーが受け取ってからの時間。 */ + age_ms?: number + } + | { type: 'ping'; online: boolean } + | { type: 'main'; online: boolean; cube?: ConnectionStatus } + | { type: 'move'; move: string; facelets: string } + | { type: 'facelets'; facelets: string } + | { type: 'session'; session: LiveSession } + +const ago = (now: number, at: number | null) => (at == null ? null : Math.max(0, Math.round(now - at))) + +/** メインのタイマーの状態を、送る形にする。splits は直前のソルブの段階（表示している分）。 */ +export function toLiveSession(state: SessionState, now: number, splits?: Split[]): LiveSession { + const { last } = state + return { + phase: state.phase, + scramble: state.exercise?.scramble ?? null, + label: state.exercise?.label ?? null, + progress: state.progress, + inspection: state.inspection, + inspection_penalty: state.inspectionPenalty, + move_count: state.moveCount, + started_ago_ms: ago(now, state.startedAt), + ready_ago_ms: ago(now, state.readyAt), + last: last + ? { + solved_at: last.solved_at, + total_ms: last.total_ms, + penalty: last.penalty, + move_count: last.moves?.length ?? 0, + splits, + } + : null, + error: state.error, + } +} + +/** 受け取ったタイマーの状態を、この端末の時計での SessionState にする（画面の部品をそのまま使うため）。 */ +export function fromLiveSession(s: LiveSession, now: number): SessionState { + const at = (agoMs: number | null) => (agoMs == null ? null : now - agoMs) + const last: NewSolve | null = s.last + ? { + mode: 'full', + solved_at: s.last.solved_at, + total_ms: s.last.total_ms, + penalty: s.last.penalty, + scramble: '', + case_id: null, + step: null, + note: '', + splits: s.last.splits, + } + : null + return { + phase: s.phase, + exercise: + s.scramble == null + ? null + : { + mode: 'full', + scramble: s.scramble, + label: s.label ?? undefined, + isGoal: () => false, + }, + progress: s.progress, + startedAt: at(s.started_ago_ms), + readyAt: at(s.ready_ago_ms), + inspection: s.inspection, + inspectionPenalty: s.inspection_penalty, + moveCount: s.move_count, + last, + error: s.error, + } +} diff --git a/web/src/live/publisher.ts b/web/src/live/publisher.ts new file mode 100644 index 0000000..7fc338f --- /dev/null +++ b/web/src/live/publisher.ts @@ -0,0 +1,159 @@ +// メイン（キューブをつないだタイマー画面）から、中継のサーバーへ出来事を送る。 +// +// 送る順番を崩さないよう、一度に1つの POST だけを送り、その間に溜まった出来事はまとめて次に送る。 +// 送れなかったときは捨て、次に送るときに今の状態を丸ごと送り直す（見る側はそれで合わせ直す）。 + +import type { NewSolve } from '../api' +import type { CubeStore } from '../cube/store' +import type { Split } from '../solve/cfop' +import type { SolveSession } from '../solve/session' +import { toLiveSession, type LiveEvent } from './protocol' + +const ENDPOINT = '/api/live/events' +/** 何もなくても送る間隔。サーバーは40秒届かなければメインがいなくなったとみなす。 */ +const HEARTBEAT_MS = 15_000 + +export type PostEvents = (events: LiveEvent[]) => Promise + +async function postEvents(events: LiveEvent[]): Promise { + const res = await fetch(ENDPOINT, { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ events }), + keepalive: true, + }) + if (!res.ok) throw new Error(`HTTP ${res.status}`) +} + +/** 出来事を順に送る列。snapshot は、送り直すときの今の状態。 */ +export class LiveQueue { + private queue: LiveEvent[] = [] + private sending = false + private resync = false + private stopped = false + private readonly post: PostEvents + private readonly snapshot: () => LiveEvent[] + + constructor(snapshot: () => LiveEvent[], post: PostEvents = postEvents) { + this.snapshot = snapshot + this.post = post + } + + push(e: LiveEvent) { + if (this.stopped) return + // タイマーの状態は最新の1つだけ送れば足りる + if (e.type === 'session') this.queue = this.queue.filter((q) => q.type !== 'session') + this.queue.push(e) + void this.flush() + } + + /** 何もなくても送る（生きている知らせ）。送っている途中なら、それで足りる。 */ + heartbeat() { + if (!this.stopped && !this.sending) void this.flush(true) + } + + /** 止める。最後に bye を送る。 */ + stop() { + if (this.stopped) return + this.queue.push({ type: 'bye' }) + void this.flush() + this.stopped = true + } + + private async flush(force = false) { + if (this.sending) return + this.sending = true + try { + while (this.queue.length > 0 || force) { + force = false + const batch = this.queue + this.queue = [] + const events = this.resync ? [...this.snapshot(), ...batch.filter((e) => e.type === 'bye')] : batch + try { + await this.post(events) + this.resync = false + } catch { + this.resync = !this.stopped + } + } + } finally { + this.sending = false + } + } +} + +/** + * キューブとタイマーの出来事を中継のサーバーへ送り始める。戻り値で止める（bye を送る）。 + * splitsOf は直前のソルブの段階（画面に出している分）を返す。 + */ +export function startLivePublish( + store: CubeStore, + session: SolveSession, + splitsOf: (last: NewSolve) => Split[] | undefined, + post?: PostEvents, +): () => void { + const sessionEvent = (): LiveEvent => { + const state = session.getSnapshot() + return { + type: 'session', + session: toLiveSession(state, performance.now(), state.last ? splitsOf(state.last) : undefined), + } + } + const snapshot = (): LiveEvent[] => { + const cube = store.getSnapshot() + return [{ type: 'main', cube: cube.status }, { type: 'facelets', facelets: cube.facelets }, sessionEvent()] + } + const queue = new LiveQueue(snapshot, post) + snapshot().forEach((e) => queue.push(e)) + + let status = store.getSnapshot().status + const unsubscribes = [ + store.moves.subscribe(({ move }) => + queue.push({ + type: 'move', + move: move.move, + facelets: store.getSnapshot().facelets, + }), + ), + store.view.subscribe((u) => { + if (u.type === 'replace') + queue.push({ + type: 'facelets', + facelets: store.getSnapshot().facelets, + }) + }), + store.subscribe(() => { + const next = store.getSnapshot().status + if (next !== status) { + status = next + queue.push({ type: 'main', cube: next }) + } + }), + session.subscribe(() => queue.push(sessionEvent())), + ] + const timer = setInterval(() => queue.heartbeat(), HEARTBEAT_MS) + // タブを閉じたときは fetch が間に合わないことがあるので sendBeacon で知らせる + const onPageHide = () => { + navigator.sendBeacon?.( + ENDPOINT, + new Blob([JSON.stringify({ events: [{ type: 'bye' }] })], { + type: 'application/json', + }), + ) + } + // 戻る・進むでページが戻ってきたら、今の状態を送り直す + const onPageShow = (e: PageTransitionEvent) => { + if (e.persisted) snapshot().forEach((ev) => queue.push(ev)) + } + const win = typeof window === 'undefined' ? null : window // テスト（Node）には無い + win?.addEventListener('pagehide', onPageHide) + win?.addEventListener('pageshow', onPageShow) + + return () => { + unsubscribes.forEach((u) => u()) + clearInterval(timer) + win?.removeEventListener('pagehide', onPageHide) + win?.removeEventListener('pageshow', onPageShow) + queue.stop() + } +} diff --git a/web/src/pages/LivePage.tsx b/web/src/pages/LivePage.tsx new file mode 100644 index 0000000..310c26e --- /dev/null +++ b/web/src/pages/LivePage.tsx @@ -0,0 +1,56 @@ +import { useState } from 'react' +import { CubeView } from '../components/CubeView' +import { FitLayout } from '../components/FitLayout' +import { ScrambleDisplay } from '../components/ScrambleDisplay' +import { StageRibbon } from '../components/StageRibbon' +import { TimerDisplay } from '../components/TimerDisplay' +import { CubeStoreContext } from '../cube/context' +import { useCubeStore } from '../cube/hooks' +import { CubeStore } from '../cube/store' +import { useScreenWakeLock } from '../cube/wake-lock' +import { useLiveFeed } from '../live/hooks' + +/** + * 中継：キューブをつないだブラウザ（メイン）のタイマー画面を映す。見るだけで、記録も操作もしない。 + * この端末のキューブ（右上の Connect）とは別のストアに映すので、ここでキューブをつないでも混ざらない。 + */ +export function LivePage() { + const { kpuzzle } = useCubeStore() + const [store] = useState(() => new CubeStore(kpuzzle)) + const view = useLiveFeed(store) + const { session } = view + const watching = view.online && view.cube === 'connected' + useScreenWakeLock(watching) + + let notice: string | null = null + if (view.stream !== 'open') + notice = + view.stream === 'connecting' ? 'サーバーにつないでいます…' : 'サーバーとの接続が切れました。つなぎ直しています…' + else if (!view.online) notice = 'メインがいません。キューブをつないだブラウザでタイマー画面を開くと、ここに映ります' + else if (view.cube !== 'connected') notice = 'メインのキューブがつながっていません' + + return ( + + +

+ {notice ?? 'メインのタイマー画面を映しています'} +

+ {session && ( + <> + + + {session.phase !== 'running' && session.last && ( + + )} + + )} + + } + cube={} + /> + + ) +} diff --git a/web/src/pages/TimerPage.tsx b/web/src/pages/TimerPage.tsx index 143cc3a..657e676 100644 --- a/web/src/pages/TimerPage.tsx +++ b/web/src/pages/TimerPage.tsx @@ -11,6 +11,7 @@ import { ScrambleDisplay } from '../components/ScrambleDisplay' import { StageRibbon } from '../components/StageRibbon' import { TimerDisplay } from '../components/TimerDisplay' import { useCubeState, useCubeStore } from '../cube/hooks' +import { useLivePublish } from '../live/hooks' import { useSettings } from '../settings/hooks' import { withSplits } from '../solve/cfop' import { fullSolveSource } from '../solve/exercises' @@ -35,6 +36,8 @@ export function TimerPage() { ) const state = useSessionState(session) const { status } = useCubeState() + // キューブがつながっている間は、ほかのブラウザ（/live）に画面を映す + useLivePublish(session, status === 'connected') const debug = useDebug() const screen = useRef(null) const last = state?.last