Skip to content
Cocot Wiki

アップロード状態のプッシュ通知（LISTEN/NOTIFY + SSE）

背景

トランスコードの完了/失敗はworkerコンテナ（internal/transcode）で確定するが、それをリアルタイムでクライアントへ伝えるHTTP接続を持っているのはappコンテナだけである。worker/appは別プロセス（別コンテナ）なので、Goのチャネルのようなプロセス内通知では橋渡しできない。

追加のミドルウェア（Redis pub/sub等）を持ち込まず、両者が既に接続している共有インフラである**PostgresのLISTEN/NOTIFY**で仲介する。WebSocketは双方向通信が要らない（サーバ→クライアントの一方向プッシュで十分な）ため採用せず、**Server-Sent Events（SSE）**をクライアント配信に使う。

全体の流れ

  1. worker → Postgres: finishUpload（変換成功）・failUpload（変換失敗、いずれもinternal/transcode/service.go）が、DB更新（CompleteUpload/FailUpload）成功直後にNotifyUploadStatus（backend/db/queries/uploads.sqlのSELECT pg_notify('upload_status_changed', $1)）を呼ぶ。payloadはnotify.Event（upload_id/work_id/status/error_message）のJSON。ベストエフォートであり、失敗してもログに残すだけで呼び出し元には伝播させない（NOTIFY自体が届かなくても、DB上のアップロード状態確定処理は既に成功しているため）。
  2. Postgres → app: internal/notify.RunListener（cmd/app/main.goがgoroutineとして起動）が、pool.Acquireで確保した専用接続でLISTEN upload_status_changedを張り続ける。WaitForNotificationで受信するたびにparseNotifyPayloadでデコードし、Broker.Publishへ渡す。接続が切れた場合は2秒後に再接続する（listenOnce/RunListenerの分離）。
  3. app内のfan-out: notify.Brokerがwork_idごとに購読チャネルの集合を持ち、Publishで該当work_idの全購読者へブロードキャストする。購読者のバッファ（4件）が詰まっている場合はブロックせずdropする——SSE接続側は再接続時にGET /api/works/:id等で最新状態を取り直せるので、取りこぼしが致命傷にならない設計。
  4. app → クライアント: works.Handler.StreamUploadEvents（GET /api/works/:id/upload-events、所有者本人のみ）が、まずBroker.Subscribeで購読してから現在のアップロード状態をDBスナップショットとして1件送出し、その後は購読チャネルに届くイベントをSSEフレーム（data: <json>\n\n）として転送する。この順序（購読 → スナップショット取得）が重要で、逆にすると「スナップショット取得後・購読開始前」にworkerがNOTIFYを送出した場合に購読者0件としてイベントがdropされ、クライアントがdone/failedを永久に受け取れずハングする競合状態になる。15秒間隔のハートビートコメント（: keepalive\n\n）でプロキシのアイドルタイムアウトを防ぐ。statusがdone/failedの終端状態に達したら接続を閉じる。

なぜEventSourceではなくfetch()+ReadableStreamか

ブラウザ標準のEventSource APIはカスタムヘッダー（Authorization）を送信できない。トークンをクエリパラメータに載せる方法はログや履歴に残るリスクがあるため避け、fetch() + ReadableStreamでレスポンスボディを手動パースする方式にした（backend/cmd/dev/dashboard/upload.htmlのstreamUploadEvents参照）。将来Flutterクライアントを接続する場合も、標準のEventSource相当のAPIはヘッダー制御ができないことが多く、同様にストリーミングレスポンスを手動パースする実装（httpパッケージのClient.sendでストリームを読む等）が必要になる見込み。

テスト方針

  • notify.Broker（Subscribe/Publish/Unsubscribeのfan-outロジック）とparseNotifyPayloadは実DB接続なしでユニットテスト済み（backend/internal/notify/broker_test.go）。
  • transcode/service.goのnotifyUploadStatusヘルパー（NOTIFYペイロードの組み立てとベストエフォート呼び出し）はfakeベースでユニットテスト済み（backend/internal/transcode/service_test.go）。
  • internal/notify/listener.go（実際のPostgres接続でのLISTEN）とworks.Handler.StreamUploadEvents（実際のSSEストリーミング）は実DB・実HTTP接続を要するため、ユニットテスト対象外。動作確認はdocker compose up環境でdev dashboard（dashboard/upload.html）からアップロード→トランスコード完了/失敗を実際に発生させて目視確認する。