feat: implement SSE streaming for real-time API error updates - #816
Conversation
…2 / #807) - add in-memory `apiErrorBroadcaster` for fan-out of `api_errors` updates - expose `GET /api/admin/errors/stream` (text/event-stream, adminRequired) with initial `ready` event, periodic keep-alive comments, and subscriber cap - publish from Sentry webhook, GitHub AI callback, and admin PATCH so the UI receives push updates from every mutation path - switch `useApiErrors` to subscribe via `EventSource` with reconnect/cleanup; fallback polling kicks in only when the stream is down - tests: broadcaster unit, SSE route headers + push delivery, hook merge/filter Closes #807
|
No actionable comments were generated in the recent review. 🎉 ℹ️ Recent review info⚙️ Run configurationConfiguration used: Path: .coderabbit.yaml Review profile: CHILL Plan: Pro Run ID: 📒 Files selected for processing (1)
🚧 Files skipped from review as they are similar to previous changes (1)
📝 WalkthroughWalkthroughAdds an in-memory server-side SSE broadcaster, an admin SSE endpoint that streams ApiError updates (with subscriber cap and keepalive), server publishers that call the broadcaster from webhooks and a PATCH handler, a client hook that consumes SSE with reconnection/filtering and polling suspension, tests, and exports ChangesAPI Errors Real-Time Streaming
Sequence DiagramsequenceDiagram
participant AC as Admin Client
participant SE as SSE Endpoint
participant BC as Broadcaster
participant WH as Webhook / Patch Handler
participant DB as Database
AC->>SE: GET /api/admin/errors/stream
SE->>AC: event: ready (retry: 30000)
SE->>BC: subscribe listener
rect rgba(100, 200, 100, 0.5)
WH->>DB: update / upsert ApiError
DB-->>WH: updated row
WH->>BC: publishApiErrorUpdate(row)
end
rect rgba(100, 150, 200, 0.5)
BC->>SE: invoke listener (row)
SE->>AC: event: update {row}
end
rect rgba(200, 150, 100, 0.5)
AC->>AC: parse update, apply filter, merge/prepend
AC->>AC: set streamConnected, suspend polling
end
Estimated code review effort🎯 4 (Complex) | ⏱️ ~50 minutes Possibly related PRs
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Comment |
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: bf380fab13
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
|
|
||
| const open = () => { | ||
| if (cancelled) return; | ||
| es = new EventSource("/api/admin/errors/stream", { withCredentials: true }); |
There was a problem hiding this comment.
In the production admin build, REST calls are sent to VITE_API_BASE_URL (admin/README.md:41, .github/workflows/deploy-prod.yml:169), but this relative EventSource URL connects to the Cloudflare Pages admin origin instead of https://api.zedi-note.app. In that split-origin environment /api/admin/errors/stream will not reach the Hono API, so the new real-time stream never receives updates and the page falls back to polling; build the stream URL with the same API base as adminFetch.
Useful? React with 👍 / 👎.
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request implements real-time updates for the API errors dashboard using Server-Sent Events (SSE). It introduces an in-memory broadcaster service on the server to fan out updates from Sentry webhooks, AI analysis callbacks, and manual status changes. The client-side useApiErrors hook was updated to subscribe to the SSE stream, handle automatic reconnections with exponential backoff, and merge incoming updates into the local state. Review feedback identifies a potential resource leak where timers are not cleared during reconnection, a timing issue with HTTP status codes when the subscriber limit is reached, and a suggestion to improve the user experience by moving updated rows to the top of the list.
| const onVisible = () => { | ||
| if (typeof document !== "undefined" && document.hidden) return; | ||
| // 可視化されたタイミングで未接続なら即時再接続を試みる。 | ||
| // When the tab becomes visible again, eagerly reconnect if we lost the | ||
| // stream while hidden. | ||
| if (!es) { | ||
| backoff = SSE_RECONNECT_INITIAL_MS; | ||
| open(); | ||
| } | ||
| }; |
There was a problem hiding this comment.
タブが可視化された際に open() を呼び出していますが、既存の reconnectTimer がクリアされていないため、タイマーが生きている場合に EventSource が重複して作成される可能性があります。これにより古い接続がリークし、リソースを無駄に消費する原因となります。open() を呼ぶ前に clearTimeout を行うべきです。
| const onVisible = () => { | |
| if (typeof document !== "undefined" && document.hidden) return; | |
| // 可視化されたタイミングで未接続なら即時再接続を試みる。 | |
| // When the tab becomes visible again, eagerly reconnect if we lost the | |
| // stream while hidden. | |
| if (!es) { | |
| backoff = SSE_RECONNECT_INITIAL_MS; | |
| open(); | |
| } | |
| }; | |
| const onVisible = () => { | |
| if (typeof document !== "undefined" && document.hidden) return; | |
| // 可視化されたタイミングで未接続なら即時再接続を試みる。 | |
| // When the tab becomes visible again, eagerly reconnect if we lost the | |
| // stream while hidden. | |
| if (!es) { | |
| if (reconnectTimer != null) { | |
| window.clearTimeout(reconnectTimer); | |
| reconnectTimer = null; | |
| } | |
| backoff = SSE_RECONNECT_INITIAL_MS; | |
| open(); | |
| } | |
| }; |
| app.get("/stream", (c) => { | ||
| return streamSSE( | ||
| c, | ||
| async (stream) => { |
There was a problem hiding this comment.
購読者数の上限チェックを streamSSE の内部で行っていますが、この時点では既に HTTP ステータス 200 OK でレスポンスが開始されています。上限に達している場合にクライアントが再試行（バックオフ）の判断を適切に行えるよう、streamSSE を呼び出す前に apiErrorSubscriberCount() を確認し、上限を超えている場合は 503 Service Unavailable を返すべきです。なお、apiErrorSubscriberCount を ../../services/apiErrorBroadcaster.js からインポートする必要があります。
| app.get("/stream", (c) => { | |
| return streamSSE( | |
| c, | |
| async (stream) => { | |
| app.get("/stream", (c) => { | |
| if (apiErrorSubscriberCount() >= API_ERROR_STREAM_MAX_SUBSCRIBERS) { | |
| c.status(503); | |
| return c.json({ error: "subscriber cap reached" }); | |
| } | |
| return streamSSE( | |
| c, | |
| async (stream) => { |
| const idx = prev.errors.findIndex((r) => r.id === row.id); | ||
| if (idx >= 0) { | ||
| const next = prev.errors.slice(); | ||
| next[idx] = row; | ||
| return { ...prev, errors: next }; | ||
| } | ||
| return { ...prev, errors: [row, ...prev.errors], total: prev.total + 1 }; |
There was a problem hiding this comment.
既存の ID が見つかった場合に元のインデックスで置換していますが、エラー一覧は通常 lastSeenAt の降順でソートされているため、更新された行（最新の発生）はリストの先頭に移動させるのが適切です。現在の実装では、更新された行がリストの下方に埋もれたままになり、リアルタイム更新の利便性が損なわれます。また、ページネーション（offset > 0）利用時の整合性についても考慮が必要です。
const filtered = prev.errors.filter((r) => r.id !== row.id);
const isUpdate = filtered.length < prev.errors.length;
// 常に先頭に配置することで、最新の更新をユーザーに提示する。
// ただし、offset > 0 の場合はページ間の重複や欠落が発生しうる点に注意。
return { ...prev, errors: [row, ...filtered], total: isUpdate ? prev.total : prev.total + 1 };- route SSE URL through `getApiUrl` so split-origin admin builds reach the API host instead of the Cloudflare Pages origin - pre-check subscriber cap before `streamSSE` so capacity exhaustion returns a real 503 instead of 200 + SSE error event - clear pending `reconnectTimer` in the visibilitychange handler to avoid spawning a duplicate `EventSource` - move SSE-updated rows to the front of the list (matches the server's `last_seen_at DESC` ordering on a fresh REST fetch)
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
🧹 Nitpick comments (1)
server/api/src/routes/admin/errors.ts (1)
239-246: 💤 Low valuePing event is not an SSE comment line; it will be delivered to clients.
Lines 242-244 comment that "SSE comment lines (
:) are ignored by EventSource", but the code sends{ data: "", event: "ping" }which is a named event, not a comment. Clients will receive this as anevent: pingmessage.If the intent is truly a keep-alive that's invisible to the client, use a raw comment line instead. If clients should handle ping events, update the comment accordingly.
Option A: Use an actual SSE comment (invisible to EventSource)
The
writeSSEmethod may not support raw comments directly. If true invisible keep-alives are needed, you'd need to write directly to the underlying stream (e.g.,stream.write(": ping\n\n")). Verify Hono's streaming API supports this.Option B: Update the comment to reflect actual behavior
- // SSE コメント行（`:` で始まる）はクライアントには配信されない。 - // SSE comment lines (leading `:`) are ignored by the EventSource API - // but keep the underlying connection alive. + // Named `ping` event keeps the connection alive. The client can ignore + // it or use it for debugging; what matters is that the TCP stays open. await stream.writeSSE({ data: "", event: "ping" });🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@server/api/src/routes/admin/errors.ts` around lines 239 - 246, The loop currently uses stream.writeSSE({ data: "", event: "ping" }) which emits a named SSE "ping" event (clients will receive it) but the comment claims it's an invisible SSE comment; change this by either: (A) sending a true SSE comment line to the underlying stream (write a line that begins with ':' and ends with the SSE record terminator) instead of using writeSSE so it stays invisible, or (B) update the comment to accurately state that writeSSE is emitting a named "ping" event that clients will receive; update references in the while loop that uses SSE_KEEPALIVE_MS and stream.writeSSE accordingly.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Nitpick comments:
In `@server/api/src/routes/admin/errors.ts`:
- Around line 239-246: The loop currently uses stream.writeSSE({ data: "",
event: "ping" }) which emits a named SSE "ping" event (clients will receive it)
but the comment claims it's an invisible SSE comment; change this by either: (A)
sending a true SSE comment line to the underlying stream (write a line that
begins with ':' and ends with the SSE record terminator) instead of using
writeSSE so it stays invisible, or (B) update the comment to accurately state
that writeSSE is emitting a named "ping" event that clients will receive; update
references in the while loop that uses SSE_KEEPALIVE_MS and stream.writeSSE
accordingly.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro
Run ID: 73878cf9-99bf-4319-ae49-3bc490141f46
📒 Files selected for processing (5)
admin/src/api/client.tsadmin/src/pages/errors/useApiErrors.test.tsadmin/src/pages/errors/useApiErrors.tsserver/api/src/__tests__/routes/admin/errors.test.tsserver/api/src/routes/admin/errors.ts
✅ Files skipped from review due to trivial changes (1)
- admin/src/pages/errors/useApiErrors.test.ts
🚧 Files skipped from review as they are similar to previous changes (1)
- admin/src/pages/errors/useApiErrors.ts
`stream.writeSSE({ event: "ping" })` always emits an `event:` field, so the
client receives a named ping event — not the invisible keep-alive the
comment claimed. Switch to `stream.write(": ping\n\n")` so the heartbeat
is a true SSE comment line (ignored by EventSource per the spec) while
still keeping idle proxies from tearing down the TCP connection.
PR #816 review feedback.
2 participants
概要
管理画面の API エラー一覧をサーバー送信イベント (SSE) でリアルタイム更新する機能を実装しました (Epic #616 Phase 2 / issue #807)。REST API で初期取得した後、
/api/admin/errors/streamエンドポイントをEventSourceで購読し、Sentry Webhook・GitHub AI コールバック・管理画面の PATCH 操作による更新をリアルタイムで受信します。変更点
クライアント側 (
admin/src/pages/errors/)useApiErrors.ts: SSE 購読機能を追加EventSourceで/api/admin/errors/streamを購読updateイベントで受信した行を既存リストにマージ（同 ID なら置換、新規なら先頭に追加）enableStreamパラメータで SSE を無効化可能（テスト用）useApiErrors.test.ts(新規): 単体テストenableStream: falseでの SSE 無効化サーバー側 (
server/api/src/)services/apiErrorBroadcaster.ts(新規): in-memory pub/subsubscribeApiErrorUpdates(): SSE 購読者を登録（上限 64 接続）publishApiErrorUpdate(): 全購読者へ行を配信ApiErrorStreamCapacityExceededErrorを投げ、SSE ルートで 503 にマップservices/apiErrorBroadcaster.test.ts(新規): 単体テストroutes/admin/errors.ts: SSE エンドポイント追加GET /api/admin/errors/stream:text/event-streamで更新を配信readyイベントとretry: 30000ヒントを送信publishApiErrorUpdate()を呼び出しroutes/webhooks/sentry.ts: Sentry Webhook でpublishApiErrorUpdate()を呼び出しroutes/webhooks/githubAiCallback.ts: GitHub AI コールバックでpublishApiErrorUpdate()を呼び出しroutes/admin/errors.test.ts: SSE エンドポイントのテスト追加readyイベントの確認updateイベントの配信確認変
https://claude.ai/code/session_01BKJQ9SDjcRmHdt7uW7LcY5
Summary by CodeRabbit
New Features
Reliability
Behavior
Tests