Uh oh!
There was an error while loading. Please reload this page.
feat(node-message-broker): implement Hub-link client with SSE wakeup - #3
Conversation
Replace the throwing HubClient stub with a real adapter over the published @privateaim/messenger-http-kit client. send/pull/ack relay to the Hub durable mailbox, authenticated as the node client via the Authup client-credentials hook. The Hub's payload-free `messagePending` wakeup is consumed through a new SseWakeupSource over GET /messages/stream. EventSource can't carry the node's Authorization header, so the stream is read over fetch behind an IWakeupSource port, with a pure SSE parser, an abort-aware auto-reconnect loop, and ping-heartbeat filtering. The IHubClient port is corrected to the real messenger contract: send resolves with the persisted ids (string[]), and pull takes a MessagePullQuery (limit/wait) — the speculative `after` cursor and Message[] return are gone. The onWakeup -> pull -> decrypt -> deliver loop and the container-facing message endpoints remain for follow-up slices (the loop gates on the crypto adapter).
Warning Review limit reached
More reviews will be available in 21 minutes and 2 seconds. Learn how PR review limits work. Your organization has used up its prepaid credits, and credit purchases are no longer available. Enable the review add-on in the billing tab to keep reviews running — you're only billed for reviews past your plan's rate limits ($0.25/file). ⌛ How to resolve this issue?After more reviews become available, a review can be triggered using the To avoid repeated limits, reduce automatic review volume by pausing incremental auto-reviews earlier, using label-based review opt-in, excluding WIP or generated PR titles, or requesting reviews manually when the PR is ready. If your team needs uninterrupted high-volume reviews, an organization admin can enable usage-based credits. 🚦 How do rate limits work?CodeRabbit enforces per-developer PR review limits for each organization. Most developers receive the normal plan review availability. For paid Pro and Pro+ PR reviews, CodeRabbit uses adaptive limits for sustained high-volume activity. When a developer's recent PR review activity reaches the 95th percentile or higher among CodeRabbit users, additional reviews become available more gradually as earlier reviews age out of the rolling window. Please see our Fair Usage Limits Policy for further information. ℹ️ Review info⚙️ Run configurationConfiguration used: defaults Review profile: CHILL Plan: Pro Run ID: 📒 Files selected for processing (4)
📝 WalkthroughWalkthroughReplaces the Phase-4 stubbed ChangesHub adapter: REST + SSE wakeup
Sequence Diagram(s)sequenceDiagram
participant ComponentsModule
participant SseWakeupSource
participant HubClient
participant MessengerHTTPClient
participant HubMessengerAPI
ComponentsModule->>MessengerHTTPClient: new Client(hubURL) + auth hook
ComponentsModule->>SseWakeupSource: new SseWakeupSource({ authorization, url })
ComponentsModule->>HubClient: new HubClient({ client, wakeup })
ComponentsModule->>HubClient: start()
HubClient->>SseWakeupSource: wakeup.start()
SseWakeupSource->>HubMessengerAPI: GET /messages/stream (Bearer token)
HubMessengerAPI-->>SseWakeupSource: SSE stream
SseWakeupSource-->>HubClient: onWakeup listener(recipient)
rect rgba(100, 149, 237, 0.5)
Note over HubClient, HubMessengerAPI: Message send/pull/ack
HubClient->>MessengerHTTPClient: client.message.send(request)
MessengerHTTPClient->>HubMessengerAPI: POST /messages
HubMessengerAPI-->>MessengerHTTPClient: string[]
MessengerHTTPClient-->>HubClient: string[]
end
Estimated code review effort🎯 4 (Complex) | ⏱️ ~50 minutes Poem
✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
tada5hi
commented
Jun 25, 2026
@coderabbitai review |
✅ Action performedReview finished.
|
Stability and robustness fixes for the SSE wakeup source and its wiring: - add an idle watchdog (resets on every event incl. heartbeats) so a silently half-open stream is dropped and reconnected instead of hanging - split lifecycle vs per-connection AbortControllers so stop() can always cancel the reconnect backoff, even after an error-triggered abort - normalise CR/CRLF line endings in the SSE parser, handling a CRLF split across chunk boundaries - tighten messagePending validation to require recipient type and id - cache the node-client grant for the SSE Authorization header so reconnects reuse the token instead of minting a fresh one each time - normalise HUB_URL trailing slash before building the stream URL - clear the auth hook's refresh timer on teardown to avoid a dangling timer - cover reconnect, idle drop, clean stop, malformed payloads and CRLF in tests
Uh oh!
There was an error while loading. Please reload this page.
What
Replaces the throwing
HubClientstub (plan 013 Track B, phase 4) with a real Hub-link adapter.HubClient—send/pull/ackrelay to the Hub durable mailbox via the published@privateaim/messenger-http-kitclient, authenticated as the node client through the Authup client-credentials hook.SseWakeupSource— consumes the Hub's payload-freemessagePendingwakeup overGET /messages/stream.EventSourcecan't carry the node'sAuthorizationheader, so the stream is read overfetchbehind anIWakeupSourceport, with a pureparseSseStreamgenerator, an abort-aware auto-reconnect loop, andping-heartbeat filtering.ComponentsModule— builds the messengerClient+ auth hook and the SSE source (reusing onetokenCreator), skips opening the stream underNODE_ENV=test.Port correction
IHubClientis aligned to the real messenger contract discovered frommessenger-http-kit/messenger-kit:sendresolves with the persisted ids (string[]), notMessage[].pulltakes aMessagePullQuery(limit/wait) — the speculativeaftercursor is gone (the Hub mailbox is delete-on-ack, no cursor).Tests
parseSseStream(chunk-boundary splits, multi-linedata, comment skipping),SseWakeupSourcedispatch (messagePending dispatched, heartbeats ignored, clean stop), andHubClientdelegation via colocated fakes. Build (JS + types), lint, and 10/10 tests green.Not in this slice
The
onWakeup → pull → decrypt → deliverloop (gated on the crypto adapter) and the container-facing message endpoints.Summary by CodeRabbit
New Features
Bug Fixes
Tests