From 2faa9e9471ef84e50b48678553c6047b6a3497f8 Mon Sep 17 00:00:00 2001 From: tada5hi Date: Wed, 24 Jun 2026 12:56:25 +0200 Subject: [PATCH 1/2] feat(node-message-broker): implement Hub-link client with SSE wakeup MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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). --- apps/node-message-broker/package.json | 1 + .../src/adapters/hub/client.ts | 59 ++-- .../src/adapters/hub/index.ts | 1 + .../src/adapters/hub/sse-wakeup-source.ts | 256 ++++++++++++++++++ .../src/app/modules/components/module.ts | 58 +++- .../node-message-broker/src/core/hub/types.ts | 57 +++- .../test/unit/adapters/hub/client.spec.ts | 89 ++++++ .../adapters/hub/fake-messenger-client.ts | 45 +++ .../unit/adapters/hub/fake-wakeup-source.ts | 43 +++ .../adapters/hub/sse-wakeup-source.spec.ts | 107 ++++++++ package-lock.json | 11 + 11 files changed, 691 insertions(+), 36 deletions(-) create mode 100644 apps/node-message-broker/src/adapters/hub/sse-wakeup-source.ts create mode 100644 apps/node-message-broker/test/unit/adapters/hub/client.spec.ts create mode 100644 apps/node-message-broker/test/unit/adapters/hub/fake-messenger-client.ts create mode 100644 apps/node-message-broker/test/unit/adapters/hub/fake-wakeup-source.ts create mode 100644 apps/node-message-broker/test/unit/adapters/hub/sse-wakeup-source.spec.ts diff --git a/apps/node-message-broker/package.json b/apps/node-message-broker/package.json index 1774a8f..3467372 100644 --- a/apps/node-message-broker/package.json +++ b/apps/node-message-broker/package.json @@ -22,6 +22,7 @@ "dependencies": { "@privateaim/errors": "^0.8.42", "@privateaim/kit": "^0.11.5", + "@privateaim/messenger-http-kit": "^0.11.5", "@privateaim/messenger-kit": "^0.11.5", "@privateaim/server-http-kit": "^0.11.5", "@privateaim/server-kit": "^0.11.5", diff --git a/apps/node-message-broker/src/adapters/hub/client.ts b/apps/node-message-broker/src/adapters/hub/client.ts index 1782cbd..7c22333 100644 --- a/apps/node-message-broker/src/adapters/hub/client.ts +++ b/apps/node-message-broker/src/adapters/hub/client.ts @@ -5,38 +5,61 @@ * view the LICENSE file that was distributed with this source code. */ -import type { Message, MessagePullResponse } from '@privateaim/messenger-kit'; -import type { IHubClient } from '../../core/hub/index.ts'; +import type { + MessageAckRequest, + MessageParty, + MessagePullQuery, + MessagePullResponse, + SendMessageRequest, +} from '@privateaim/messenger-kit'; +import type { + IHubClient, + IMessengerClient, + IWakeupSource, +} from '../../core/hub/index.ts'; -const NOT_IMPLEMENTED = 'HubClient is a Phase 4 stub (Plan 013 Track B): implement against @privateaim/messenger-http-kit + the SSE wakeup stream.'; +type HubClientContext = { + client: IMessengerClient, + wakeup: IWakeupSource +}; /** - * Hub-link adapter. Phase 4 implements this with the `@privateaim/messenger-http-kit` - * Hapic client (REST `send` / `pull` / `ack`) authenticating as the node client, plus - * the SSE wakeup stream (`GET /messages/stream`) feeding `onWakeup`. + * Hub-link adapter. REST `send` / `pull` / `ack` go through the + * `@privateaim/messenger-http-kit` client (authenticated as the node client); + * `onWakeup` rides the SSE wakeup source. The node relays opaque end-to-end + * payloads — encryption/decryption is the caller's concern, not the Hub's. */ export class HubClient implements IHubClient { - async send(): Promise { - throw new Error(NOT_IMPLEMENTED); + protected client: IMessengerClient; + + protected wakeup: IWakeupSource; + + constructor(ctx: HubClientContext) { + this.client = ctx.client; + this.wakeup = ctx.wakeup; + } + + send(input: SendMessageRequest): Promise { + return this.client.message.send(input); } - async pull(): Promise { - throw new Error(NOT_IMPLEMENTED); + pull(query?: MessagePullQuery): Promise { + return this.client.message.pull(query); } - async ack(): Promise { - throw new Error(NOT_IMPLEMENTED); + ack(input: MessageAckRequest): Promise { + return this.client.message.ack(input); } - onWakeup(): () => void { - return () => {}; + onWakeup(listener: (recipient: MessageParty) => void): () => void { + return this.wakeup.subscribe((event) => listener(event.recipient)); } - async start(): Promise { - // no-op until the Phase 4 SSE subscription lands + start(): Promise { + return this.wakeup.start(); } - async stop(): Promise { - // no-op until the Phase 4 SSE subscription lands + stop(): Promise { + return this.wakeup.stop(); } } diff --git a/apps/node-message-broker/src/adapters/hub/index.ts b/apps/node-message-broker/src/adapters/hub/index.ts index ab87f0e..7e00d62 100644 --- a/apps/node-message-broker/src/adapters/hub/index.ts +++ b/apps/node-message-broker/src/adapters/hub/index.ts @@ -6,3 +6,4 @@ */ export * from './client.ts'; +export * from './sse-wakeup-source.ts'; diff --git a/apps/node-message-broker/src/adapters/hub/sse-wakeup-source.ts b/apps/node-message-broker/src/adapters/hub/sse-wakeup-source.ts new file mode 100644 index 0000000..85c9050 --- /dev/null +++ b/apps/node-message-broker/src/adapters/hub/sse-wakeup-source.ts @@ -0,0 +1,256 @@ +/* + * Copyright (c) 2026. + * Author Peter Placzek (tada5hi) + * For the full copyright and license information, + * view the LICENSE file that was distributed with this source code. + */ + +import type { MessagePendingEvent } from '@privateaim/messenger-kit'; +import { WakeupEventName } from '@privateaim/messenger-kit'; +import type { Logger } from '@privateaim/server-kit'; +import type { IWakeupSource } from '../../core/hub/index.ts'; + +type SseEvent = { + event: string, + data: string +}; + +type SseWakeupSourceContext = { + /** Absolute URL of the Hub's `GET /messages/stream` SSE endpoint. */ + url: string, + /** Resolves the `Authorization` header value (e.g. `Bearer `) per connection. */ + authorization: () => Promise, + /** Injectable for tests; defaults to the global `fetch`. */ + fetchFn?: typeof fetch, + /** Backoff between reconnect attempts. */ + reconnectDelayMs?: number, + logger?: Logger +}; + +const DEFAULT_RECONNECT_DELAY_MS = 3000; + +/** + * Parse a byte stream of Server-Sent Events into `{ event, data }` records. + * Pure and transport-agnostic: events are separated by a blank line, `event:` + * sets the type (default `message`), and consecutive `data:` lines are joined + * with newlines (per the SSE spec). Comment lines (`:`) and unknown fields are + * ignored. + */ +export async function* parseSseStream( + source: AsyncIterable, +): AsyncGenerator { + const decoder = new TextDecoder(); + let buffer = ''; + + for await (const chunk of source) { + buffer += decoder.decode(chunk, { stream: true }); + + let boundary = buffer.indexOf('\n\n'); + while (boundary !== -1) { + const block = buffer.slice(0, boundary); + buffer = buffer.slice(boundary + 2); + + const event = parseSseBlock(block); + if (event) { + yield event; + } + + boundary = buffer.indexOf('\n\n'); + } + } +} + +function parseSseBlock(block: string): SseEvent | undefined { + let event = 'message'; + const data: string[] = []; + + for (const line of block.split('\n')) { + if (line.length === 0 || line.startsWith(':')) { + continue; + } + + const separator = line.indexOf(':'); + const field = separator === -1 ? line : line.slice(0, separator); + let value = separator === -1 ? '' : line.slice(separator + 1); + if (value.startsWith(' ')) { + value = value.slice(1); + } + + if (field === 'event') { + event = value; + } else if (field === 'data') { + data.push(value); + } + } + + if (data.length === 0) { + return undefined; + } + + return { event, data: data.join('\n') }; +} + +function isMessagePendingEvent(value: unknown): value is MessagePendingEvent { + if (typeof value !== 'object' || value === null) { + return false; + } + + const { recipient } = value as { recipient?: unknown }; + return typeof recipient === 'object' && recipient !== null; +} + +async function* readableToAsyncIterable( + stream: ReadableStream, +): AsyncIterable { + const reader = stream.getReader(); + try { + for (;;) { + const { done, value } = await reader.read(); + if (done) { + break; + } + if (value) { + yield value; + } + } + } finally { + reader.releaseLock(); + } +} + +/** + * Consumes the Hub's `messagePending` SSE stream (`GET /messages/stream`) and + * forwards each signal to subscribers. `EventSource` can't carry the node's + * `Authorization` header, so the stream is read over `fetch`; the loop + * auto-reconnects with backoff until {@link stop}. `ping` heartbeats and any + * non-`messagePending` events are ignored. + */ +export class SseWakeupSource implements IWakeupSource { + protected url: string; + + protected authorization: () => Promise; + + protected fetchFn: typeof fetch; + + protected reconnectDelayMs: number; + + protected logger: Logger | undefined; + + protected listeners = new Set<(event: MessagePendingEvent) => void>(); + + protected controller: AbortController | undefined; + + protected loop: Promise | undefined; + + protected running = false; + + constructor(ctx: SseWakeupSourceContext) { + this.url = ctx.url; + this.authorization = ctx.authorization; + this.fetchFn = ctx.fetchFn ?? fetch; + this.reconnectDelayMs = ctx.reconnectDelayMs ?? DEFAULT_RECONNECT_DELAY_MS; + this.logger = ctx.logger; + } + + subscribe(listener: (event: MessagePendingEvent) => void): () => void { + this.listeners.add(listener); + return () => { + this.listeners.delete(listener); + }; + } + + async start(): Promise { + if (this.running) { + return; + } + + this.running = true; + this.loop = this.run(); + } + + async stop(): Promise { + this.running = false; + if (this.controller) { + this.controller.abort(); + } + if (this.loop) { + await this.loop; + this.loop = undefined; + } + } + + protected async run(): Promise { + while (this.running) { + try { + await this.connect(); + } catch (error) { + if (this.running) { + this.logger?.warn(`Message wakeup stream disconnected: ${(error as Error).message}`); + } + } + + if (this.running) { + await this.delay(this.reconnectDelayMs); + } + } + } + + protected async connect(): Promise { + const controller = new AbortController(); + this.controller = controller; + + const authorization = await this.authorization(); + const response = await this.fetchFn(this.url, { + method: 'GET', + headers: { + accept: 'text/event-stream', + authorization, + }, + signal: controller.signal, + }); + + if (!response.ok || !response.body) { + throw new Error(`unexpected response (status ${response.status})`); + } + + for await (const event of parseSseStream(readableToAsyncIterable(response.body))) { + if (event.event === WakeupEventName.MESSAGE_PENDING) { + this.dispatch(event.data); + } + } + } + + protected dispatch(data: string): void { + let parsed: unknown; + try { + parsed = JSON.parse(data); + } catch { + return; + } + + if (!isMessagePendingEvent(parsed)) { + return; + } + + for (const listener of this.listeners) { + try { + listener(parsed); + } catch (error) { + this.logger?.warn(`Message wakeup listener failed: ${(error as Error).message}`); + } + } + } + + protected delay(ms: number): Promise { + return new Promise((resolve) => { + const timer = setTimeout(resolve, ms); + if (typeof timer.unref === 'function') { + timer.unref(); + } + this.controller?.signal.addEventListener('abort', () => { + clearTimeout(timer); + resolve(); + }, { once: true }); + }); + } +} diff --git a/apps/node-message-broker/src/app/modules/components/module.ts b/apps/node-message-broker/src/app/modules/components/module.ts index e6ff14c..cad4f14 100644 --- a/apps/node-message-broker/src/app/modules/components/module.ts +++ b/apps/node-message-broker/src/app/modules/components/module.ts @@ -7,14 +7,26 @@ import type { IContainer } from 'eldin'; import type { IModule } from 'orkos'; +import { Client } from '@privateaim/messenger-http-kit'; +import { + EnvironmentName, + LoggerInjectionKey, + createAuthupClientAuthenticationHook, + createAuthupClientTokenCreator, +} from '@privateaim/server-kit'; import { MemoryDeliveryService } from '../../../adapters/delivery/index.ts'; -import { HubClient } from '../../../adapters/hub/index.ts'; +import { HubClient, SseWakeupSource } from '../../../adapters/hub/index.ts'; +import { ConfigInjectionKey } from '../config/constants.ts'; import { ComponentsInjectionKey } from './constants.ts'; /** * Wires the broker's moving parts: local delivery (webhook registry) and the - * Hub link. Phase 4 connects the Hub client's `onWakeup` → pull → decrypt → - * `delivery.deliver()` loop; today it registers the ports with a stub client. + * Hub link — REST send/pull/ack via `@privateaim/messenger-http-kit` plus the + * SSE wakeup stream, both authenticated as the node client via client + * credentials. + * + * Phase 4 (Plan 013 Track B): the `onWakeup` → pull → decrypt → `delivery.deliver()` + * loop still needs the crypto adapter before it can be connected here. */ export class ComponentsModule implements IModule { readonly name = 'components'; @@ -24,16 +36,44 @@ export class ComponentsModule implements IModule { private hubClient: HubClient | undefined; async setup(container: IContainer): Promise { + const config = container.resolve(ConfigInjectionKey); + + const loggerResult = container.tryResolve(LoggerInjectionKey); + const logger = loggerResult.success ? loggerResult.data : undefined; + const delivery = new MemoryDeliveryService(); container.register(ComponentsInjectionKey.Delivery, { useValue: delivery }); - // Phase 4 (Plan 013 Track B): construct the real Hub client from config - // (@privateaim/messenger-http-kit + the SSE wakeup stream) and wire - // onWakeup → pull → decrypt → delivery.deliver(). - const hubClient = new HubClient(); - await hubClient.start(); - this.hubClient = hubClient; + // node-client credentials authenticate every Hub interaction; the same + // token creator backs the REST auth hook and the SSE Authorization header. + const tokenCreator = createAuthupClientTokenCreator({ + baseURL: config.authupURL, + clientId: config.clientId, + clientSecret: config.clientSecret, + realm: config.realm, + }); + + const client = new Client({ baseURL: config.hubURL }); + createAuthupClientAuthenticationHook({ + baseURL: config.authupURL, + tokenCreator, + }).attach(client); + + const wakeup = new SseWakeupSource({ + url: new URL('messages/stream', config.hubURL).toString(), + authorization: async () => `Bearer ${(await tokenCreator()).access_token}`, + logger, + }); + + const hubClient = new HubClient({ client, wakeup }); container.register(ComponentsInjectionKey.HubClient, { useValue: hubClient }); + + // tests don't reach a live Hub; skip opening the reconnecting stream. + if (config.env !== EnvironmentName.TEST) { + await hubClient.start(); + } + + this.hubClient = hubClient; } async teardown(): Promise { diff --git a/apps/node-message-broker/src/core/hub/types.ts b/apps/node-message-broker/src/core/hub/types.ts index 2d88fd6..f43942e 100644 --- a/apps/node-message-broker/src/core/hub/types.ts +++ b/apps/node-message-broker/src/core/hub/types.ts @@ -6,33 +6,72 @@ */ import type { - Message, MessageAckRequest, MessageParty, + MessagePendingEvent, + MessagePullQuery, MessagePullResponse, SendMessageRequest, } from '@privateaim/messenger-kit'; +/** + * The slice of the `@privateaim/messenger-http-kit` message API the broker relies + * on. Declared structurally so the Hub adapter can be tested with a fake instead + * of a live HTTP client. + */ +export interface IMessengerMessageApi { + send(data: SendMessageRequest): Promise; + + pull(query?: MessagePullQuery): Promise; + + ack(data: MessageAckRequest): Promise; +} + +/** The shape of the `@privateaim/messenger-http-kit` `Client` the broker depends on. */ +export interface IMessengerClient { + message: IMessengerMessageApi; +} + +/** + * Payload-free wakeup channel. The Hub emits a `messagePending` signal — carrying + * only the recipient identity, never the payload — over SSE; the source forwards + * each signal to its subscribers, which respond by pulling via + * {@link IHubClient.pull}. Implemented in `adapters/hub` over the Hub's + * `GET /messages/stream` endpoint. + */ +export interface IWakeupSource { + /** Register a wakeup listener; returns an unsubscribe fn. */ + subscribe(listener: (event: MessagePendingEvent) => void): () => void; + + /** Open the wakeup channel (auto-reconnecting until {@link stop}). */ + start(): Promise; + + /** Close the wakeup channel. */ + stop(): Promise; +} + /** * Port to the Hub durable message broker. The node relays sends to the Hub - * mailbox and pulls inbound ciphertext from it; a payload-free wakeup (SSE - * preferred) triggers an immediate pull, with long-poll as the fallback. + * mailbox and pulls inbound ciphertext from it; a payload-free wakeup (SSE) + * triggers an immediate pull, with the pull's own `wait` long-poll as fallback. * - * Implemented in `adapters/hub` via `@privateaim/messenger-http-kit`. + * Implemented in `adapters/hub` via `@privateaim/messenger-http-kit` plus an + * {@link IWakeupSource}. */ export interface IHubClient { - /** Persist one row per recipient in the Hub mailbox. */ - send(input: SendMessageRequest): Promise; + /** Persist one row per recipient in the Hub mailbox; resolves with their ids. */ + send(input: SendMessageRequest): Promise; - /** Cursor-based pull of messages addressed to this node; `wait` long-polls. */ - pull(input?: { after?: string; wait?: number }): Promise; + /** Pull this node's pending messages (oldest first); `wait` long-polls. */ + pull(query?: MessagePullQuery): Promise; - /** Advance the recipient cursor (deletes acknowledged rows). */ + /** Acknowledge messages by id — the Hub deletes them for this node. */ ack(input: MessageAckRequest): Promise; /** Subscribe to payload-free `messagePending` wakeups; returns an unsubscribe fn. */ onWakeup(listener: (recipient: MessageParty) => void): () => void; start(): Promise; + stop(): Promise; } diff --git a/apps/node-message-broker/test/unit/adapters/hub/client.spec.ts b/apps/node-message-broker/test/unit/adapters/hub/client.spec.ts new file mode 100644 index 0000000..f13e7bd --- /dev/null +++ b/apps/node-message-broker/test/unit/adapters/hub/client.spec.ts @@ -0,0 +1,89 @@ +/* + * Copyright (c) 2026. + * Author Peter Placzek (tada5hi) + * For the full copyright and license information, + * view the LICENSE file that was distributed with this source code. + */ + +import type { MessageParty, SendMessageRequest } from '@privateaim/messenger-kit'; +import { describe, expect, it } from 'vitest'; +import { HubClient } from '../../../../src/adapters/hub/index.ts'; +import { FakeMessengerClient } from './fake-messenger-client.ts'; +import { FakeWakeupSource } from './fake-wakeup-source.ts'; + +function setup() { + const client = new FakeMessengerClient(); + const wakeup = new FakeWakeupSource(); + const hubClient = new HubClient({ client, wakeup }); + return { + client, + wakeup, + hubClient, + }; +} + +describe('adapters/hub/client', () => { + const recipient: MessageParty = { type: 'client', id: 'node-1' }; + + it('relays send to the messenger client and returns the persisted ids', async () => { + const { client, hubClient } = setup(); + client.sendResult = ['m1', 'm2']; + + const request: SendMessageRequest = { recipients: [recipient], data: 'cipher' }; + const ids = await hubClient.send(request); + + expect(ids).toEqual(['m1', 'm2']); + expect(client.sent).toEqual([request]); + }); + + it('forwards the pull query and returns the response', async () => { + const { client, hubClient } = setup(); + client.pullResult = { + messages: [{ + id: 'm1', + sender_type: 'client', + sender_id: 'other', + recipient_type: 'client', + recipient_id: 'node-1', + data: 'cipher', + metadata: null, + created_at: '2026-01-01T00:00:00.000Z', + }], + }; + + const response = await hubClient.pull({ limit: 10, wait: 5000 }); + + expect(client.pulled).toEqual([{ limit: 10, wait: 5000 }]); + expect(response.messages).toHaveLength(1); + }); + + it('relays ack to the messenger client', async () => { + const { client, hubClient } = setup(); + + await hubClient.ack({ ids: ['m1', 'm2'] }); + + expect(client.acked).toEqual([{ ids: ['m1', 'm2'] }]); + }); + + it('delivers wakeups as the recipient and supports unsubscribe', () => { + const { wakeup, hubClient } = setup(); + const received: MessageParty[] = []; + + const unsubscribe = hubClient.onWakeup((party) => received.push(party)); + wakeup.emit({ recipient }); + unsubscribe(); + wakeup.emit({ recipient }); + + expect(received).toEqual([recipient]); + }); + + it('delegates start and stop to the wakeup source', async () => { + const { wakeup, hubClient } = setup(); + + await hubClient.start(); + await hubClient.stop(); + + expect(wakeup.started).toBe(1); + expect(wakeup.stopped).toBe(1); + }); +}); diff --git a/apps/node-message-broker/test/unit/adapters/hub/fake-messenger-client.ts b/apps/node-message-broker/test/unit/adapters/hub/fake-messenger-client.ts new file mode 100644 index 0000000..b54c4c4 --- /dev/null +++ b/apps/node-message-broker/test/unit/adapters/hub/fake-messenger-client.ts @@ -0,0 +1,45 @@ +/* + * Copyright (c) 2026. + * Author Peter Placzek (tada5hi) + * For the full copyright and license information, + * view the LICENSE file that was distributed with this source code. + */ + +import type { + MessageAckRequest, + MessagePullQuery, + MessagePullResponse, + SendMessageRequest, +} from '@privateaim/messenger-kit'; +import type { IMessengerClient, IMessengerMessageApi } from '../../../../src/core/hub/index.ts'; + +/** + * In-memory `IMessengerClient` that records every send/pull/ack and returns + * configurable canned results — stands in for the `@privateaim/messenger-http-kit` + * `Client` so the Hub adapter is testable without a live Hub. + */ +export class FakeMessengerClient implements IMessengerClient { + sent: SendMessageRequest[] = []; + + pulled: (MessagePullQuery | undefined)[] = []; + + acked: MessageAckRequest[] = []; + + sendResult: string[] = []; + + pullResult: MessagePullResponse = { messages: [] }; + + message: IMessengerMessageApi = { + send: async (data) => { + this.sent.push(data); + return this.sendResult; + }, + pull: async (query) => { + this.pulled.push(query); + return this.pullResult; + }, + ack: async (data) => { + this.acked.push(data); + }, + }; +} diff --git a/apps/node-message-broker/test/unit/adapters/hub/fake-wakeup-source.ts b/apps/node-message-broker/test/unit/adapters/hub/fake-wakeup-source.ts new file mode 100644 index 0000000..a62c6c6 --- /dev/null +++ b/apps/node-message-broker/test/unit/adapters/hub/fake-wakeup-source.ts @@ -0,0 +1,43 @@ +/* + * Copyright (c) 2026. + * Author Peter Placzek (tada5hi) + * For the full copyright and license information, + * view the LICENSE file that was distributed with this source code. + */ + +import type { MessagePendingEvent } from '@privateaim/messenger-kit'; +import type { IWakeupSource } from '../../../../src/core/hub/index.ts'; + +/** + * In-memory `IWakeupSource` whose {@link emit} drives wakeups synchronously and + * which counts start/stop calls — lets the Hub adapter be tested without the SSE + * transport. + */ +export class FakeWakeupSource implements IWakeupSource { + listeners = new Set<(event: MessagePendingEvent) => void>(); + + started = 0; + + stopped = 0; + + subscribe(listener: (event: MessagePendingEvent) => void): () => void { + this.listeners.add(listener); + return () => { + this.listeners.delete(listener); + }; + } + + async start(): Promise { + this.started += 1; + } + + async stop(): Promise { + this.stopped += 1; + } + + emit(event: MessagePendingEvent): void { + for (const listener of this.listeners) { + listener(event); + } + } +} diff --git a/apps/node-message-broker/test/unit/adapters/hub/sse-wakeup-source.spec.ts b/apps/node-message-broker/test/unit/adapters/hub/sse-wakeup-source.spec.ts new file mode 100644 index 0000000..7c17bfb --- /dev/null +++ b/apps/node-message-broker/test/unit/adapters/hub/sse-wakeup-source.spec.ts @@ -0,0 +1,107 @@ +/* + * Copyright (c) 2026. + * Author Peter Placzek (tada5hi) + * For the full copyright and license information, + * view the LICENSE file that was distributed with this source code. + */ + +import type { MessagePendingEvent } from '@privateaim/messenger-kit'; +import { WakeupEventName } from '@privateaim/messenger-kit'; +import { describe, expect, it } from 'vitest'; +import { SseWakeupSource, parseSseStream } from '../../../../src/adapters/hub/index.ts'; + +async function* chunksOf(parts: string[]): AsyncGenerator { + const encoder = new TextEncoder(); + for (const part of parts) { + yield encoder.encode(part); + } +} + +function streamFrom(parts: string[]): ReadableStream { + const encoder = new TextEncoder(); + return new ReadableStream({ + start(controller) { + for (const part of parts) { + controller.enqueue(encoder.encode(part)); + } + controller.close(); + }, + }); +} + +describe('adapters/hub/parseSseStream', () => { + it('parses events split across chunk boundaries, joins data, skips comments', async () => { + const events = []; + for await (const event of parseSseStream(chunksOf([ + 'event: messagePen', + 'ding\ndata: {"recipient"', + ':{"type":"client","id":"n1"}}\n\n', + ': heartbeat-comment\n\n', + 'event: ping\ndata: 1\n\n', + ]))) { + events.push(event); + } + + expect(events).toEqual([ + { event: 'messagePending', data: '{"recipient":{"type":"client","id":"n1"}}' }, + { event: 'ping', data: '1' }, + ]); + }); +}); + +describe('adapters/hub/SseWakeupSource', () => { + const recipient = { type: 'client', id: 'node-1' }; + + it('dispatches messagePending events and ignores heartbeats', async () => { + const received: MessagePendingEvent[] = []; + let signal: () => void = () => {}; + const fired = new Promise((resolve) => { + signal = resolve; + }); + + const fetchFn: typeof fetch = async () => new Response( + streamFrom([ + 'event: ping\ndata: 1\n\n', + `event: ${WakeupEventName.MESSAGE_PENDING}\ndata: ${JSON.stringify({ recipient })}\n\n`, + ]), + { status: 200, headers: { 'content-type': 'text/event-stream' } }, + ); + + const source = new SseWakeupSource({ + url: 'http://hub.test/messages/stream', + authorization: async () => 'Bearer test-token', + fetchFn, + reconnectDelayMs: 60_000, + }); + + source.subscribe((event) => { + received.push(event); + signal(); + }); + + await source.start(); + await fired; + await source.stop(); + + expect(received).toEqual([{ recipient }]); + }); + + it('stops cleanly without ever connecting', async () => { + let calls = 0; + const fetchFn: typeof fetch = async () => { + calls += 1; + return new Response(streamFrom([]), { status: 200 }); + }; + + const source = new SseWakeupSource({ + url: 'http://hub.test/messages/stream', + authorization: async () => 'Bearer test-token', + fetchFn, + reconnectDelayMs: 60_000, + }); + + await source.stop(); + + expect(calls).toBe(0); + }); +}); diff --git a/package-lock.json b/package-lock.json index 2b87979..acba798 100644 --- a/package-lock.json +++ b/package-lock.json @@ -37,6 +37,7 @@ "dependencies": { "@privateaim/errors": "^0.8.42", "@privateaim/kit": "^0.11.5", + "@privateaim/messenger-http-kit": "^0.11.5", "@privateaim/messenger-kit": "^0.11.5", "@privateaim/server-http-kit": "^0.11.5", "@privateaim/server-kit": "^0.11.5", @@ -2018,6 +2019,16 @@ "@authup/kit": "^1.0.0-beta.48" } }, + "node_modules/@privateaim/messenger-http-kit": { + "version": "0.11.5", + "resolved": "https://registry.npmjs.org/@privateaim/messenger-http-kit/-/messenger-http-kit-0.11.5.tgz", + "integrity": "sha512-BJ/G+0gQADQ+xYxUB6i7XSfRnBtOuPuose6WsQpH6I/hGdgcSL1vEAECRbSl9BCPUEMqOo4BScs6+eVZomf16g==", + "license": "Apache-2.0", + "peerDependencies": { + "@privateaim/messenger-kit": "^0.11.5", + "hapic": "^2.8.2" + } + }, "node_modules/@privateaim/messenger-kit": { "version": "0.11.5", "resolved": "https://registry.npmjs.org/@privateaim/messenger-kit/-/messenger-kit-0.11.5.tgz", From 699562f54e33987b84e742cb47ecd533a94f6f0e Mon Sep 17 00:00:00 2001 From: tada5hi Date: Thu, 25 Jun 2026 11:34:07 +0200 Subject: [PATCH 2/2] fix(node-message-broker): harden Hub SSE wakeup link 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 --- .../src/adapters/hub/sse-wakeup-source.ts | 111 +++++++-- .../src/app/modules/components/module.ts | 61 ++++- .../adapters/hub/sse-wakeup-source.spec.ts | 217 +++++++++++++++++- 3 files changed, 358 insertions(+), 31 deletions(-) diff --git a/apps/node-message-broker/src/adapters/hub/sse-wakeup-source.ts b/apps/node-message-broker/src/adapters/hub/sse-wakeup-source.ts index 85c9050..ec2a950 100644 --- a/apps/node-message-broker/src/adapters/hub/sse-wakeup-source.ts +++ b/apps/node-message-broker/src/adapters/hub/sse-wakeup-source.ts @@ -24,17 +24,26 @@ type SseWakeupSourceContext = { fetchFn?: typeof fetch, /** Backoff between reconnect attempts. */ reconnectDelayMs?: number, + /** + * Abort and reconnect if no event — including the Hub's `ping` heartbeats — + * arrives within this window; guards against silently half-open connections + * that never emit FIN/RST. Must exceed the Hub's heartbeat interval. `<= 0` + * disables the watchdog. + */ + idleTimeoutMs?: number, logger?: Logger }; const DEFAULT_RECONNECT_DELAY_MS = 3000; +const DEFAULT_IDLE_TIMEOUT_MS = 60_000; + /** * Parse a byte stream of Server-Sent Events into `{ event, data }` records. * Pure and transport-agnostic: events are separated by a blank line, `event:` * sets the type (default `message`), and consecutive `data:` lines are joined - * with newlines (per the SSE spec). Comment lines (`:`) and unknown fields are - * ignored. + * with newlines (per the SSE spec). CR / CRLF line endings are normalised to LF, + * comment lines (`:`) and unknown fields are ignored. */ export async function* parseSseStream( source: AsyncIterable, @@ -45,6 +54,15 @@ export async function* parseSseStream( for await (const chunk of source) { buffer += decoder.decode(chunk, { stream: true }); + // Normalise CR / CRLF to LF (SSE spec). A trailing CR is held back: it + // may be the first half of a CRLF that is split across two chunks. + let trailingCr = ''; + if (buffer.endsWith('\r')) { + trailingCr = '\r'; + buffer = buffer.slice(0, -1); + } + buffer = buffer.replace(/\r\n?/g, '\n') + trailingCr; + let boundary = buffer.indexOf('\n\n'); while (boundary !== -1) { const block = buffer.slice(0, boundary); @@ -96,7 +114,12 @@ function isMessagePendingEvent(value: unknown): value is MessagePendingEvent { } const { recipient } = value as { recipient?: unknown }; - return typeof recipient === 'object' && recipient !== null; + if (typeof recipient !== 'object' || recipient === null) { + return false; + } + + const { type, id } = recipient as { type?: unknown, id?: unknown }; + return typeof type === 'string' && typeof id === 'string'; } async function* readableToAsyncIterable( @@ -122,8 +145,9 @@ async function* readableToAsyncIterable( * Consumes the Hub's `messagePending` SSE stream (`GET /messages/stream`) and * forwards each signal to subscribers. `EventSource` can't carry the node's * `Authorization` header, so the stream is read over `fetch`; the loop - * auto-reconnects with backoff until {@link stop}. `ping` heartbeats and any - * non-`messagePending` events are ignored. + * auto-reconnects with backoff until {@link stop}, and an idle watchdog drops a + * connection that has gone silent (no heartbeats) so it can be re-established. + * `ping` heartbeats and any non-`messagePending` events are ignored. */ export class SseWakeupSource implements IWakeupSource { protected url: string; @@ -134,11 +158,17 @@ export class SseWakeupSource implements IWakeupSource { protected reconnectDelayMs: number; + protected idleTimeoutMs: number; + protected logger: Logger | undefined; protected listeners = new Set<(event: MessagePendingEvent) => void>(); - protected controller: AbortController | undefined; + /** Aborted once, by {@link stop}, to tear the whole loop down. */ + protected stopController: AbortController | undefined; + + /** Aborted per connection — by {@link stop} or by the idle watchdog. */ + protected connController: AbortController | undefined; protected loop: Promise | undefined; @@ -149,6 +179,7 @@ export class SseWakeupSource implements IWakeupSource { this.authorization = ctx.authorization; this.fetchFn = ctx.fetchFn ?? fetch; this.reconnectDelayMs = ctx.reconnectDelayMs ?? DEFAULT_RECONNECT_DELAY_MS; + this.idleTimeoutMs = ctx.idleTimeoutMs ?? DEFAULT_IDLE_TIMEOUT_MS; this.logger = ctx.logger; } @@ -165,14 +196,14 @@ export class SseWakeupSource implements IWakeupSource { } this.running = true; + this.stopController = new AbortController(); this.loop = this.run(); } async stop(): Promise { this.running = false; - if (this.controller) { - this.controller.abort(); - } + this.stopController?.abort(); + this.connController?.abort(); if (this.loop) { await this.loop; this.loop = undefined; @@ -196,8 +227,13 @@ export class SseWakeupSource implements IWakeupSource { } protected async connect(): Promise { - const controller = new AbortController(); - this.controller = controller; + const { stopController } = this; + if (!stopController) { + return; + } + + const conn = new AbortController(); + this.connController = conn; const authorization = await this.authorization(); const response = await this.fetchFn(this.url, { @@ -206,16 +242,38 @@ export class SseWakeupSource implements IWakeupSource { accept: 'text/event-stream', authorization, }, - signal: controller.signal, + signal: AbortSignal.any([stopController.signal, conn.signal]), }); if (!response.ok || !response.body) { throw new Error(`unexpected response (status ${response.status})`); } - for await (const event of parseSseStream(readableToAsyncIterable(response.body))) { - if (event.event === WakeupEventName.MESSAGE_PENDING) { - this.dispatch(event.data); + let idleTimer: ReturnType | undefined; + const armIdle = () => { + if (this.idleTimeoutMs <= 0) { + return; + } + if (idleTimer) { + clearTimeout(idleTimer); + } + idleTimer = setTimeout(() => conn.abort(), this.idleTimeoutMs); + if (typeof idleTimer.unref === 'function') { + idleTimer.unref(); + } + }; + + try { + armIdle(); + for await (const event of parseSseStream(readableToAsyncIterable(response.body))) { + armIdle(); + if (event.event === WakeupEventName.MESSAGE_PENDING) { + this.dispatch(event.data); + } + } + } finally { + if (idleTimer) { + clearTimeout(idleTimer); } } } @@ -243,14 +301,27 @@ export class SseWakeupSource implements IWakeupSource { protected delay(ms: number): Promise { return new Promise((resolve) => { - const timer = setTimeout(resolve, ms); - if (typeof timer.unref === 'function') { - timer.unref(); + const signal = this.stopController?.signal; + if (signal?.aborted) { + resolve(); + return; } - this.controller?.signal.addEventListener('abort', () => { + + let timer: ReturnType; + const onAbort = () => { clearTimeout(timer); resolve(); - }, { once: true }); + }; + + timer = setTimeout(() => { + signal?.removeEventListener('abort', onAbort); + resolve(); + }, ms); + if (typeof timer.unref === 'function') { + timer.unref(); + } + + signal?.addEventListener('abort', onAbort, { once: true }); }); } } diff --git a/apps/node-message-broker/src/app/modules/components/module.ts b/apps/node-message-broker/src/app/modules/components/module.ts index cad4f14..88c0d82 100644 --- a/apps/node-message-broker/src/app/modules/components/module.ts +++ b/apps/node-message-broker/src/app/modules/components/module.ts @@ -35,6 +35,8 @@ export class ComponentsModule implements IModule { private hubClient: HubClient | undefined; + private authHook: ReturnType | undefined; + async setup(container: IContainer): Promise { const config = container.resolve(ConfigInjectionKey); @@ -44,8 +46,9 @@ export class ComponentsModule implements IModule { const delivery = new MemoryDeliveryService(); container.register(ComponentsInjectionKey.Delivery, { useValue: delivery }); - // node-client credentials authenticate every Hub interaction; the same - // token creator backs the REST auth hook and the SSE Authorization header. + // node-client credentials authenticate every Hub interaction; this creator + // backs the REST auth hook directly and the SSE Authorization header via a + // caching wrapper (below). const tokenCreator = createAuthupClientTokenCreator({ baseURL: config.authupURL, clientId: config.clientId, @@ -54,14 +57,25 @@ export class ComponentsModule implements IModule { }); const client = new Client({ baseURL: config.hubURL }); - createAuthupClientAuthenticationHook({ + const authHook = createAuthupClientAuthenticationHook({ baseURL: config.authupURL, tokenCreator, - }).attach(client); + }); + authHook.attach(client); + this.authHook = authHook; + + // The SSE stream reads over raw `fetch`, so it can't piggyback on the REST + // hook's cached token; cache the node-client grant ourselves so reconnects + // reuse it instead of minting a fresh grant per (re)connect. + const wakeupTokenCreator = createCachedTokenCreator(tokenCreator); + + // `new URL(relative, base)` drops the last path segment of a base without a + // trailing slash — guard so a sub-pathed HUB_URL still resolves correctly. + const hubBaseURL = config.hubURL.endsWith('/') ? config.hubURL : `${config.hubURL}/`; const wakeup = new SseWakeupSource({ - url: new URL('messages/stream', config.hubURL).toString(), - authorization: async () => `Bearer ${(await tokenCreator()).access_token}`, + url: new URL('messages/stream', hubBaseURL).toString(), + authorization: async () => `Bearer ${(await wakeupTokenCreator()).access_token}`, logger, }); @@ -81,5 +95,40 @@ export class ComponentsModule implements IModule { await this.hubClient.stop(); this.hubClient = undefined; } + + // The auth hook owns a token-refresh timer; drop it so it can't fire (and + // hit Authup) after shutdown. + if (this.authHook) { + this.authHook.disable(); + this.authHook.clearTimer(); + this.authHook = undefined; + } } } + +/** + * Wrap a {@link createAuthupClientTokenCreator} result so the grant is reused + * until shortly before it expires, instead of minting a fresh one on every call. + * Calls are serial (the SSE source reconnects one at a time), so no in-flight + * de-duplication is needed. + */ +function createCachedTokenCreator( + inner: ReturnType, +): ReturnType { + const EXPIRY_MARGIN_SECONDS = 30; + + let cached: Awaited> | undefined; + let expiresAt = 0; + + return async () => { + const now = Date.now(); + if (cached && now < expiresAt) { + return cached; + } + + const grant = await inner(); + cached = grant; + expiresAt = now + Math.max(0, grant.expires_in - EXPIRY_MARGIN_SECONDS) * 1000; + return grant; + }; +} diff --git a/apps/node-message-broker/test/unit/adapters/hub/sse-wakeup-source.spec.ts b/apps/node-message-broker/test/unit/adapters/hub/sse-wakeup-source.spec.ts index 7c17bfb..25970f7 100644 --- a/apps/node-message-broker/test/unit/adapters/hub/sse-wakeup-source.spec.ts +++ b/apps/node-message-broker/test/unit/adapters/hub/sse-wakeup-source.spec.ts @@ -17,6 +17,7 @@ async function* chunksOf(parts: string[]): AsyncGenerator { } } +/** Stream that emits `parts` then closes. */ function streamFrom(parts: string[]): ReadableStream { const encoder = new TextEncoder(); return new ReadableStream({ @@ -29,6 +30,36 @@ function streamFrom(parts: string[]): ReadableStream { }); } +/** Stream that emits `parts` then stays open until `signal` aborts. */ +function openStream(parts: string[], signal?: AbortSignal | null): ReadableStream { + const encoder = new TextEncoder(); + return new ReadableStream({ + start(controller) { + for (const part of parts) { + controller.enqueue(encoder.encode(part)); + } + const abort = () => { + try { + controller.error(new Error('aborted')); + } catch { + // stream already closed/errored + } + }; + if (signal) { + if (signal.aborted) { + abort(); + } else { + signal.addEventListener('abort', abort, { once: true }); + } + } + }, + }); +} + +function pendingEvent(recipient: { type: string, id: string }): string { + return `event: ${WakeupEventName.MESSAGE_PENDING}\ndata: ${JSON.stringify({ recipient })}\n\n`; +} + describe('adapters/hub/parseSseStream', () => { it('parses events split across chunk boundaries, joins data, skips comments', async () => { const events = []; @@ -47,10 +78,37 @@ describe('adapters/hub/parseSseStream', () => { { event: 'ping', data: '1' }, ]); }); + + it('parses CRLF-delimited streams', async () => { + const events = []; + for await (const event of parseSseStream(chunksOf([ + 'event: messagePending\r\ndata: {"recipient":{"type":"client","id":"n1"}}\r\n\r\n', + ]))) { + events.push(event); + } + + expect(events).toEqual([ + { event: 'messagePending', data: '{"recipient":{"type":"client","id":"n1"}}' }, + ]); + }); + + it('handles a CRLF that straddles a chunk boundary', async () => { + const events = []; + for await (const event of parseSseStream(chunksOf([ + 'data: a\r', + '\ndata: b\r\n\r\n', + ]))) { + events.push(event); + } + + expect(events).toEqual([{ event: 'message', data: 'a\nb' }]); + }); }); describe('adapters/hub/SseWakeupSource', () => { const recipient = { type: 'client', id: 'node-1' }; + const url = 'http://hub.test/messages/stream'; + const authorization = async () => 'Bearer test-token'; it('dispatches messagePending events and ignores heartbeats', async () => { const received: MessagePendingEvent[] = []; @@ -62,16 +120,92 @@ describe('adapters/hub/SseWakeupSource', () => { const fetchFn: typeof fetch = async () => new Response( streamFrom([ 'event: ping\ndata: 1\n\n', - `event: ${WakeupEventName.MESSAGE_PENDING}\ndata: ${JSON.stringify({ recipient })}\n\n`, + pendingEvent(recipient), ]), { status: 200, headers: { 'content-type': 'text/event-stream' } }, ); const source = new SseWakeupSource({ - url: 'http://hub.test/messages/stream', - authorization: async () => 'Bearer test-token', + url, + authorization, fetchFn, reconnectDelayMs: 60_000, + idleTimeoutMs: 0, + }); + + source.subscribe((event) => { + received.push(event); + signal(); + }); + + await source.start(); + await fired; + await source.stop(); + + expect(received).toEqual([{ recipient }]); + }); + + it('ignores malformed JSON and recipients missing type/id', async () => { + const received: MessagePendingEvent[] = []; + let signal: () => void = () => {}; + const fired = new Promise((resolve) => { + signal = resolve; + }); + + const fetchFn: typeof fetch = async () => new Response( + streamFrom([ + `event: ${WakeupEventName.MESSAGE_PENDING}\ndata: not-json\n\n`, + `event: ${WakeupEventName.MESSAGE_PENDING}\ndata: ${JSON.stringify({ recipient: { id: 'x' } })}\n\n`, + pendingEvent(recipient), + ]), + { status: 200, headers: { 'content-type': 'text/event-stream' } }, + ); + + const source = new SseWakeupSource({ + url, + authorization, + fetchFn, + reconnectDelayMs: 60_000, + idleTimeoutMs: 0, + }); + + source.subscribe((event) => { + received.push(event); + signal(); + }); + + await source.start(); + await fired; + await source.stop(); + + expect(received).toEqual([{ recipient }]); + }); + + it('reconnects after a failed connection and then dispatches', async () => { + const received: MessagePendingEvent[] = []; + let signal: () => void = () => {}; + const fired = new Promise((resolve) => { + signal = resolve; + }); + + let calls = 0; + const fetchFn: typeof fetch = async (_input, init) => { + calls += 1; + if (calls === 1) { + return new Response(streamFrom([]), { status: 503 }); + } + return new Response( + openStream([pendingEvent(recipient)], init?.signal), + { status: 200, headers: { 'content-type': 'text/event-stream' } }, + ); + }; + + const source = new SseWakeupSource({ + url, + authorization, + fetchFn, + reconnectDelayMs: 5, + idleTimeoutMs: 0, }); source.subscribe((event) => { @@ -83,9 +217,82 @@ describe('adapters/hub/SseWakeupSource', () => { await fired; await source.stop(); + expect(calls).toBeGreaterThanOrEqual(2); expect(received).toEqual([{ recipient }]); }); + it('drops a silent connection via the idle watchdog and reconnects', async () => { + const received: MessagePendingEvent[] = []; + let signal: () => void = () => {}; + const fired = new Promise((resolve) => { + signal = resolve; + }); + + let calls = 0; + const fetchFn: typeof fetch = async (_input, init) => { + calls += 1; + if (calls === 1) { + // open but silent — only the idle watchdog can end it + return new Response( + openStream([], init?.signal), + { status: 200, headers: { 'content-type': 'text/event-stream' } }, + ); + } + return new Response( + openStream([pendingEvent(recipient)], init?.signal), + { status: 200, headers: { 'content-type': 'text/event-stream' } }, + ); + }; + + const source = new SseWakeupSource({ + url, + authorization, + fetchFn, + reconnectDelayMs: 5, + idleTimeoutMs: 15, + }); + + source.subscribe((event) => { + received.push(event); + signal(); + }); + + await source.start(); + await fired; + await source.stop(); + + expect(calls).toBeGreaterThanOrEqual(2); + expect(received).toEqual([{ recipient }]); + }); + + it('stops promptly while a connection is open', async () => { + let calls = 0; + const fetchFn: typeof fetch = async (_input, init) => { + calls += 1; + return new Response( + openStream([], init?.signal), + { status: 200, headers: { 'content-type': 'text/event-stream' } }, + ); + }; + + const source = new SseWakeupSource({ + url, + authorization, + fetchFn, + reconnectDelayMs: 60_000, + idleTimeoutMs: 0, + }); + + await source.start(); + // let the loop open the connection before tearing it down + await new Promise((resolve) => { + setTimeout(resolve, 5); + }); + await source.stop(); + + expect(calls).toBe(1); + }); + it('stops cleanly without ever connecting', async () => { let calls = 0; const fetchFn: typeof fetch = async () => { @@ -94,8 +301,8 @@ describe('adapters/hub/SseWakeupSource', () => { }; const source = new SseWakeupSource({ - url: 'http://hub.test/messages/stream', - authorization: async () => 'Bearer test-token', + url, + authorization, fetchFn, reconnectDelayMs: 60_000, });