diff --git a/apps/node-message-broker/package.json b/apps/node-message-broker/package.json index c687cf2..1244f66 100644 --- a/apps/node-message-broker/package.json +++ b/apps/node-message-broker/package.json @@ -25,6 +25,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..ec2a950 --- /dev/null +++ b/apps/node-message-broker/src/adapters/hub/sse-wakeup-source.ts @@ -0,0 +1,327 @@ +/* + * 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, + /** + * 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). CR / CRLF line endings are normalised to LF, + * 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 }); + + // 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); + 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 }; + 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( + 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}, 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; + + protected authorization: () => Promise; + + protected fetchFn: typeof fetch; + + protected reconnectDelayMs: number; + + protected idleTimeoutMs: number; + + protected logger: Logger | undefined; + + protected listeners = new Set<(event: MessagePendingEvent) => void>(); + + /** 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; + + 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.idleTimeoutMs = ctx.idleTimeoutMs ?? DEFAULT_IDLE_TIMEOUT_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.stopController = new AbortController(); + this.loop = this.run(); + } + + async stop(): Promise { + this.running = false; + this.stopController?.abort(); + this.connController?.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 { stopController } = this; + if (!stopController) { + return; + } + + const conn = new AbortController(); + this.connController = conn; + + const authorization = await this.authorization(); + const response = await this.fetchFn(this.url, { + method: 'GET', + headers: { + accept: 'text/event-stream', + authorization, + }, + signal: AbortSignal.any([stopController.signal, conn.signal]), + }); + + if (!response.ok || !response.body) { + throw new Error(`unexpected response (status ${response.status})`); + } + + 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); + } + } + } + + 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 signal = this.stopController?.signal; + if (signal?.aborted) { + resolve(); + return; + } + + let timer: ReturnType; + const onAbort = () => { + clearTimeout(timer); + resolve(); + }; + + 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 e6ff14c..88c0d82 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'; @@ -23,17 +35,59 @@ export class ComponentsModule implements IModule { private hubClient: HubClient | undefined; + private authHook: ReturnType | 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; 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, + clientSecret: config.clientSecret, + realm: config.realm, + }); + + const client = new Client({ baseURL: config.hubURL }); + const authHook = createAuthupClientAuthenticationHook({ + baseURL: config.authupURL, + tokenCreator, + }); + 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', hubBaseURL).toString(), + authorization: async () => `Bearer ${(await wakeupTokenCreator()).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 { @@ -41,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/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..25970f7 --- /dev/null +++ b/apps/node-message-broker/test/unit/adapters/hub/sse-wakeup-source.spec.ts @@ -0,0 +1,314 @@ +/* + * 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); + } +} + +/** Stream that emits `parts` then closes. */ +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(); + }, + }); +} + +/** 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 = []; + 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' }, + ]); + }); + + 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[] = []; + let signal: () => void = () => {}; + const fired = new Promise((resolve) => { + signal = resolve; + }); + + const fetchFn: typeof fetch = async () => new Response( + streamFrom([ + 'event: ping\ndata: 1\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('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) => { + received.push(event); + signal(); + }); + + await source.start(); + 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 () => { + calls += 1; + return new Response(streamFrom([]), { status: 200 }); + }; + + const source = new SseWakeupSource({ + url, + authorization, + 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",