Add broadcaster subscription lifecycles

src/app.ts

@@ -1,5 +1,6 @@
 import { Config } from './services/config.ts';
 import { Database, MigrationService } from './services/storage/index.ts';
+import { Broadcaster } from './services/broadcaster.ts';
 import { ApplicationRouter } from './services/router.ts';
 import { Logger } from './utils/logger.ts';
 
@@ -18,6 +19,8 @@ export class App {
         const migrations = new MigrationService(database, debug);
         await migrations.migrateToLatest();
 
+        // Domain services are shared across all transports and route modules.
+        const broadcaster = new Broadcaster(debug);
         const routes = [];
 
         // Route loading is an explicit startup phase, not first-request work.
@@ -25,13 +28,14 @@ export class App {
         // before any client can connect.
         const router = await ApplicationRouter.create(routes);
 
-        return new App(database, router);
+        return new App(database, broadcaster, router);
     }
 
     private stopPromise: Promise<void> | undefined;
 
     constructor(
         private readonly database: Database,
+        private readonly broadcaster: Broadcaster,
         private readonly router: ApplicationRouter,
     ) {}
 

src/constants.ts

@@ -0,0 +1,4 @@
+/**
+ * HTTP status code for "Not Acceptable" error.
+ */
+export const HTTP_STATUS_CODE_NOT_ACCEPTED = 406;

src/services/broadcaster.ts

@@ -0,0 +1,313 @@
+import { HTTP_STATUS_CODE_NOT_ACCEPTED } from '../constants.ts';
+
+import type { BaseStream, StreamEvent } from './stream/base-stream.ts';
+import type { Logger } from '../utils/logger.ts';
+import { ApplicationError } from '../errors/index.ts';
+
+/** Request-scoped view from which the broadcaster obtains a stable connection. */
+export interface BroadcastStream {
+    /** Connection identity shared by every request on the same transport session. */
+    readonly connection: BaseStream;
+
+    /** Whether the connection can remain open to receive published events. */
+    readonly streaming: boolean;
+}
+
+/** One pending subscribe call and the topics whose removal will resolve it. */
+interface SubscriptionWaiter {
+    /** Only topics newly introduced by this particular subscribe call. */
+    readonly remainingTopics: Set<string>;
+
+    /** Completes the promise returned to the subscribing route. */
+    readonly resolve: () => void;
+}
+
+/** Topic delivery contract consumed by domain routes. */
+export abstract class BaseBroadcaster {
+    /**
+     * Subscribe a connection and wait until the topics added by this call are removed.
+     *
+     * Fully duplicate subscriptions resolve immediately.
+     *
+     * @param stream - The stream to subscribe.
+     * @param topics - The topics to subscribe to.
+     */
+    abstract subscribe(stream: BroadcastStream, topics: string[]): Promise<void>;
+
+    /**
+     * Unsubscribes a stream from a list of topics.
+     * @param stream - The stream to unsubscribe.
+     * @param topics - The topics to unsubscribe from.
+     */
+    abstract unsubscribe(stream: BroadcastStream, topics?: string[]): Promise<void>;
+
+    /**
+     * Publishes an event to a topic.
+     * @param topic - The topic to publish to.
+     * @param event - The event to publish.
+     * @returns The published event.
+     */
+    abstract publish(topic: string, event: Omit<StreamEvent, 'id'>): Promise<StreamEvent>;
+
+    /**
+     * Sends an event to a stream.
+     * @param stream - The stream to send the event to.
+     * @param event - The event to send.
+     */
+    abstract sendEvent(stream: BaseStream, event: StreamEvent): Promise<void>;
+}
+
+/**
+ * In-memory topic index with reverse lookup for deterministic stream cleanup.
+ *
+ * A stream is stored strongly only while it has topics. The WeakSet records that
+ * its close observer has already been installed without extending its lifetime.
+ */
+export class Broadcaster extends BaseBroadcaster {
+    /** Namespaced diagnostic logger for subscription and publication activity. */
+    private readonly debug: Logger;
+
+    /** Forward index: topic name to the connections receiving that topic. */
+    private readonly topicStreams = new Map<string, Set<BaseStream>>();
+
+    /** Reverse index: connection to all topics currently attached to it. */
+    private readonly streamTopics = new Map<BaseStream, Set<string>>();
+
+    /** Pending subscribe calls grouped by their stable connection identity. */
+    private readonly subscriptionWaiters = new Map<BaseStream, Set<SubscriptionWaiter>>();
+
+    /** Connections which already have the single required close observer. */
+    private readonly observedStreams = new WeakSet<BaseStream>();
+
+    /**
+     * Creates a new Broadcaster.
+     * @param debug - The debug logger.
+     */
+    constructor(debug: Logger) {
+        super();
+
+        // Extend the debug logger to include the broadcaster namespace.
+        this.debug = debug.extend('broadcaster');
+    }
+
+    /**
+     * Subscribe a connection and return a promise for this call's additions.
+     *
+     * Registration is synchronous. The returned promise resolves after all
+     * topics newly added by this call are removed, or when the connection closes.
+     * If every requested topic already exists, it resolves immediately.
+     *
+     * @param stream - The stream to subscribe.
+     * @param topics - The topics to subscribe to.
+     */
+    subscribe(stream: BroadcastStream, topics: string[]): Promise<void> {
+        // Normal HTTP cannot receive later publications, so fail before
+        // mutating either subscription index.
+        if (!stream.streaming) {
+            throw new ApplicationError(HTTP_STATUS_CODE_NOT_ACCEPTED, 'This route requires a stream-capable connection');
+        }
+
+        // ApplicationRouteStream is request-scoped, but subscriptions must survive
+        // across requests. Always index by the shared underlying connection.
+        const connection = stream.connection;
+
+        // Reuse the connection's reverse-index entry when it already has topics.
+        // A new Set is not stored until this call actually introduces a topic.
+        const trackedTopics = this.streamTopics.get(connection) ?? new Set<string>();
+
+        // Deduplicate the request itself
+        const deduplicatedTopics = Array.from(new Set(topics));
+
+        // Filter topics that are already subscribed to by the connection.
+        const topicsToAdd = deduplicatedTopics.filter((topic) => !trackedTopics.has(topic));
+
+        // A fully duplicate (or empty) subscription adds no lifetime to track.
+        if (topicsToAdd.length === 0) {
+            return Promise.resolve();
+        }
+
+        // Store the reverse index before installing the close observer. An
+        // already-closed connection invokes onClose immediately and must be able
+        // to remove the topics registered by this call.
+        this.streamTopics.set(connection, trackedTopics);
+
+        for (const topic of topicsToAdd) {
+            // Find or create the forward-index set for this topic.
+            let streams = this.topicStreams.get(topic);
+
+            if (!streams) {
+                streams = new Set();
+                this.topicStreams.set(topic, streams);
+            }
+
+            // Update both indexes together: publication uses the forward index,
+            // while unsubscribe and connection cleanup use the reverse index.
+            streams.add(connection);
+            trackedTopics.add(topic);
+        }
+
+        // Create the lifecycle promise returned to the route. Its waiter owns only
+        // the topics added above, not duplicate topics owned by earlier calls.
+        const removed = new Promise<void>((resolve) => {
+            // Several non-overlapping subscription requests can remain active on
+            // one WebSocket connection, so each connection stores a set of waiters.
+            const waiters = this.subscriptionWaiters.get(connection) ?? new Set<SubscriptionWaiter>();
+
+            waiters.add({
+                remainingTopics: new Set(topicsToAdd),
+                resolve,
+            });
+
+            this.subscriptionWaiters.set(connection, waiters);
+        });
+
+        // Register the waiter before observing closure: onClose invokes its
+        // callback immediately when registration races with an already-closed stream.
+        if (!this.observedStreams.has(connection)) {
+            // WeakSet prevents repeated subscribe requests from adding duplicate
+            // close callbacks without retaining an otherwise unused connection.
+            this.observedStreams.add(connection);
+
+            // Remote disconnect, local close, and shutdown all use the same cleanup
+            // path, which also resolves every affected subscription promise.
+            connection.onClose(() => this.removeSubscriptions(connection));
+        }
+
+        // Log only the topics introduced by this call; duplicates were no-ops.
+        this.debug('subscribed stream to topics %o', topicsToAdd);
+
+        // Keep the route dispatch pending for exactly this subscription's
+        // lifetime. Duplicate calls return an already-resolved promise above.
+        return removed;
+    }
+
+    /**
+     * Unsubscribes a stream from a list of topics.
+     * @param stream - The stream to unsubscribe.
+     * @param topics - The topics to unsubscribe from.
+     */
+    async unsubscribe(stream: BroadcastStream, topics?: string[]): Promise<void> {
+        // Resolve request-scoped facades to the same stable connection key used
+        // during subscribe, allowing a later WebSocket request to unsubscribe.
+        const topicsToRemove = this.removeSubscriptions(stream.connection, topics);
+
+        // Logging remains useful even for idempotent removal of missing topics.
+        this.debug('unsubscribed stream from topics %o', topicsToRemove);
+    }
+
+    /**
+     * Remove topics from a connection and resolve affected subscription calls.
+     *
+     * @param connection - Stable connection stored in the topic index.
+     * @param topics - Specific topics to remove, or all current topics.
+     * @returns The topics considered for removal.
+     */
+    private removeSubscriptions(connection: BaseStream, topics?: string[]): string[] {
+        // Missing connections are valid: unsubscribe is deliberately idempotent.
+        const trackedTopics = this.streamTopics.get(connection);
+
+        // Omitting topics means connection cleanup, so remove every tracked topic.
+        const topicsToRemove = topics ?? Array.from(trackedTopics ?? []);
+
+        // Waiters should advance only for topics that were genuinely active.
+        const removedTopics = new Set<string>();
+
+        for (const topic of topicsToRemove) {
+            // Deleting from the reverse index reports whether this call actually
+            // removed an active connection/topic relationship.
+            if (trackedTopics?.delete(topic)) {
+                removedTopics.add(topic);
+            }
+
+            // Remove the same relationship from the publication index.
+            this.topicStreams.get(topic)?.delete(connection);
+
+            // Empty topic sets have no value and would unnecessarily retain maps.
+            if (this.topicStreams.get(topic)?.size === 0) {
+                this.topicStreams.delete(topic);
+            }
+        }
+
+        // Stop strongly retaining connections after their final topic is removed.
+        if (!trackedTopics || trackedTopics.size === 0) {
+            this.streamTopics.delete(connection);
+        }
+
+        // Resolve subscribe calls whose newly added topics have all disappeared.
+        const waiters = this.subscriptionWaiters.get(connection);
+
+        if (waiters) {
+            for (const waiter of waiters) {
+                // Partial unsubscribe removes only the affected portion of each
+                // waiter's outstanding topic set.
+                for (const topic of removedTopics) {
+                    waiter.remainingTopics.delete(topic);
+                }
+
+                // The route completes once every topic introduced by its call has
+                // been removed, even if the connection still has other topics.
+                if (waiter.remainingTopics.size === 0) {
+                    waiters.delete(waiter);
+                    waiter.resolve();
+                }
+            }
+
+            // Avoid retaining an empty waiter collection after all routes settle.
+            if (waiters.size === 0) {
+                this.subscriptionWaiters.delete(connection);
+            }
+        }
+
+        // Return the requested removal list for consistent unsubscribe logging.
+        return topicsToRemove;
+    }
+
+    /**
+     * Publishes an event to a topic.
+     * @param topic - The topic to publish to.
+     * @param event - The event to publish.
+     * @returns The published event.
+     */
+    async publish(topic: string, event: Omit<StreamEvent, 'id'>): Promise<StreamEvent> {
+        // Get the current timestamp.
+        const timestamp = Date.now();
+
+        // Add an ID to the event.
+        const eventWithId: StreamEvent = {
+            ...event,
+            id: String(timestamp),
+        };
+
+        // Copy the current subscriber set and start every send immediately.
+        // Promise.all provides concurrent fan-out while still allowing publish to
+        // wait until every local delivery attempt has settled.
+        await Promise.all(Array.from(this.topicStreams.get(topic) ?? [], (stream) => this.sendEvent(stream, eventWithId)));
+
+        // Log the published event.
+        this.debug('published %s to topic %s', event.type, topic);
+
+        return eventWithId;
+    }
+
+    /**
+     * Sends an event to a stream.
+     * @param stream - The stream to send the event to.
+     * @param event - The event to send.
+     */
+    async sendEvent(stream: BaseStream, event: StreamEvent): Promise<void> {
+        try {
+            // Broadcaster messages bypass the request facade because pushed events
+            // are connection-level and must not inherit a request correlation ID.
+            await stream.send(event);
+        } catch (error) {
+            // Log the error.
+            this.debug('failed to send event to stream: %O', error);
+
+            // A failed connection cannot receive future publications. Closing it
+            // triggers the normal observer cleanup; the explicit removal also
+            // makes this path safe for unusual stream implementations.
+            stream.close();
+            this.removeSubscriptions(stream);
+        }
+    }
+}

tests/services/broadcaster.test.ts

@@ -0,0 +1,199 @@
+import { describe, expect, it, vi } from "vitest";
+
+import { ApplicationError } from "../../src/errors/index.js";
+import { Broadcaster } from "../../src/services/broadcaster.js";
+import { ApplicationRouteStream } from "../../src/services/route-stream.js";
+import { Logger } from "../../src/utils/logger.js";
+import { TestConnection } from "../helpers/test-connection.js";
+
+function createBroadcaster(): Broadcaster {
+  return new Broadcaster(new Logger("broadcaster-test"));
+}
+
+function routeStream(connection: TestConnection): ApplicationRouteStream {
+  return new ApplicationRouteStream(connection, undefined);
+}
+
+async function expectPending(promise: Promise<void>): Promise<void> {
+  const settled = vi.fn();
+  void promise.then(settled);
+  await Promise.resolve();
+  expect(settled).not.toHaveBeenCalled();
+}
+
+describe("Broadcaster subscriptions", () => {
+  it("delivers events and resolves after a later request removes the topic", async () => {
+    const broadcaster = createBroadcaster();
+    const connection = new TestConnection(true, true);
+    const subscribed = broadcaster.subscribe(routeStream(connection), [
+      "items",
+    ]);
+
+    await expectPending(subscribed);
+    await broadcaster.publish("items", {
+      type: "item-changed",
+      data: { id: "a" },
+    });
+
+    expect(connection.messages).toEqual([
+      expect.objectContaining({
+        type: "item-changed",
+        data: { id: "a" },
+      }),
+    ]);
+
+    // A different request-scoped facade still resolves the connection's
+    // original subscription.
+    await broadcaster.unsubscribe(routeStream(connection), ["items"]);
+    await expect(subscribed).resolves.toBeUndefined();
+  });
+
+  it("resolves fully duplicate subscriptions immediately", async () => {
+    const broadcaster = createBroadcaster();
+    const connection = new TestConnection(true, true);
+    const first = broadcaster.subscribe(routeStream(connection), [
+      "items",
+      "items",
+    ]);
+    const duplicate = broadcaster.subscribe(routeStream(connection), ["items"]);
+
+    await expect(duplicate).resolves.toBeUndefined();
+    await expectPending(first);
+    expect(connection.closeCallbacks).toHaveLength(1);
+
+    await broadcaster.unsubscribe(routeStream(connection), ["items"]);
+    await first;
+  });
+
+  it("waits only for topics newly added by a partially overlapping call", async () => {
+    const broadcaster = createBroadcaster();
+    const connection = new TestConnection(true, true);
+    const first = broadcaster.subscribe(routeStream(connection), ["a"]);
+    const second = broadcaster.subscribe(routeStream(connection), ["a", "b"]);
+
+    await broadcaster.unsubscribe(routeStream(connection), ["a"]);
+    await expect(first).resolves.toBeUndefined();
+    await expectPending(second);
+
+    await broadcaster.unsubscribe(routeStream(connection), ["b"]);
+    await expect(second).resolves.toBeUndefined();
+  });
+
+  it("resolves every pending subscription and removes topics on close", async () => {
+    const broadcaster = createBroadcaster();
+    const connection = new TestConnection(true, false);
+    const first = broadcaster.subscribe(routeStream(connection), ["a"]);
+    const second = broadcaster.subscribe(routeStream(connection), ["b"]);
+
+    connection.close();
+    await Promise.all([first, second]);
+    await broadcaster.publish("a", { type: "changed", data: null });
+    await broadcaster.publish("b", { type: "changed", data: null });
+
+    expect(connection.messages).toEqual([]);
+  });
+
+  it("immediately resolves registration against an already-closed connection", async () => {
+    const broadcaster = createBroadcaster();
+    const connection = new TestConnection(true, false);
+    connection.close();
+
+    await expect(
+      broadcaster.subscribe(routeStream(connection), ["items"]),
+    ).resolves.toBeUndefined();
+  });
+
+  it("rejects subscriptions on a non-streaming connection", () => {
+    const broadcaster = createBroadcaster();
+    const connection = new TestConnection(false, false);
+
+    expect(() =>
+      broadcaster.subscribe(routeStream(connection), ["items"]),
+    ).toThrowError(
+      expect.objectContaining({ statusCode: 406 }),
+    );
+  });
+
+  it("treats an empty subscription and repeated unsubscribe as no-ops", async () => {
+    const broadcaster = createBroadcaster();
+    const connection = new TestConnection(true, true);
+
+    await expect(
+      broadcaster.subscribe(routeStream(connection), []),
+    ).resolves.toBeUndefined();
+    await broadcaster.unsubscribe(routeStream(connection), ["missing"]);
+    await broadcaster.unsubscribe(routeStream(connection));
+
+    expect(connection.closeCallbacks).toHaveLength(0);
+  });
+
+  it("fans out concurrently to independent connections", async () => {
+    const broadcaster = createBroadcaster();
+    const first = new TestConnection(true, false);
+    const second = new TestConnection(true, false);
+    const originalFirstSend = first.send.bind(first);
+    let releaseFirst: () => void = () => undefined;
+    const firstReleased = new Promise<void>((resolve) => {
+      releaseFirst = resolve;
+    });
+    let markSecondSent: () => void = () => undefined;
+    const secondSent = new Promise<void>((resolve) => {
+      markSecondSent = resolve;
+    });
+
+    first.send = async (message) => {
+      await firstReleased;
+      await originalFirstSend(message);
+    };
+    second.send = async (message) => {
+      await TestConnection.prototype.send.call(second, message);
+      markSecondSent();
+    };
+
+    const firstSubscription = broadcaster.subscribe(routeStream(first), [
+      "items",
+    ]);
+    const secondSubscription = broadcaster.subscribe(routeStream(second), [
+      "items",
+    ]);
+    const publication = broadcaster.publish("items", {
+      type: "item-changed",
+      data: {},
+    });
+
+    await secondSent;
+    releaseFirst();
+    await publication;
+
+    expect(first.messages).toHaveLength(1);
+    expect(second.messages).toHaveLength(1);
+
+    first.close();
+    second.close();
+    await Promise.all([firstSubscription, secondSubscription]);
+  });
+
+  it("closes and removes a connection whose event delivery fails", async () => {
+    const broadcaster = createBroadcaster();
+    const connection = new TestConnection(true, true);
+    connection.send = vi.fn().mockRejectedValue(new Error("socket failed"));
+    const subscribed = broadcaster.subscribe(routeStream(connection), [
+      "items",
+    ]);
+
+    await broadcaster.publish("items", {
+      type: "item-changed",
+      data: {},
+    });
+    await subscribed;
+
+    expect(connection.closed).toBe(true);
+    expect(connection.send).toHaveBeenCalledOnce();
+
+    await broadcaster.publish("items", {
+      type: "item-changed",
+      data: {},
+    });
+    expect(connection.send).toHaveBeenCalledOnce();
+  });
+});

tests/subscription-flow.test.ts

@@ -0,0 +1,76 @@
+import { describe, expect, it, vi } from "vitest";
+
+import type { RouteDefinition, RouteModule } from "../src/routes/types.js";
+import { ApplicationRouter } from "../src/services/router.js";
+import { Broadcaster } from "../src/services/broadcaster.js";
+import { Logger } from "../src/utils/logger.js";
+import { TestConnection } from "./helpers/test-connection.js";
+
+function moduleWith(routes: RouteDefinition[]): RouteModule {
+  return {
+    async getRoutes() {
+      return routes;
+    },
+  };
+}
+
+async function expectPending(promise: Promise<void>): Promise<void> {
+  const settled = vi.fn();
+  void promise.then(settled);
+  await Promise.resolve();
+  expect(settled).not.toHaveBeenCalled();
+}
+
+describe("long-lived subscription dispatch", () => {
+  it("allows duplicate subscribe and unsubscribe requests while the original request waits", async () => {
+    const broadcaster = new Broadcaster(new Logger("subscription-flow-test"));
+    const router = await ApplicationRouter.create([
+      moduleWith([
+        {
+          url: "/items/subscribe",
+          handler: async (stream) => {
+            await broadcaster.subscribe(stream, ["items"]);
+          },
+        },
+        {
+          url: "/items/unsubscribe",
+          handler: async (stream) => {
+            await broadcaster.unsubscribe(stream, ["items"]);
+            await stream.send({});
+          },
+        },
+      ]),
+    ]);
+    const connection = new TestConnection(true, true);
+
+    const original = router.dispatch(
+      { path: "/items/subscribe", requestId: "subscribe-1" },
+      connection,
+    );
+    await vi.waitFor(() => expect(connection.closeCallbacks).toHaveLength(1));
+    await expectPending(original);
+
+    // This request uses a different ApplicationRouteStream over the same
+    // connection. Since the topic already exists, its dispatch completes.
+    await router.dispatch(
+      { path: "/items/subscribe", requestId: "subscribe-2" },
+      connection,
+    );
+    await expectPending(original);
+
+    await router.dispatch(
+      { path: "/items/unsubscribe", requestId: "unsubscribe-1" },
+      connection,
+    );
+    await original;
+
+    expect(connection.messages).toEqual([
+      {
+        id: "unsubscribe-1",
+        type: "response",
+        statusCode: 200,
+        body: {},
+      },
+    ]);
+  });
+});