`subscription-bus` — typed fan-out with the cross-tenant capability gate

(no submodules)

Purpose

SubscriptionBus is the broadcast surface every plugin uses to fan a payload out to every subscribed connection on the host. It owns the (topic, tenant) lookup against the registry, the per-connection filter predicate, and the cross-tenant capability gate documented in design doc §11.2. This module exposes the abstract SubscriptionBus<Payload> contract and ships InMemorySubscriptionBus — the single-host default used by the conformance harness.

The bus encodes each broadcast into a WsEnvelope exactly once and pushes the same encoded frame to every matching connection; per-connection filters can drop the delivery without re-encoding the payload.

Features

Generic SubscriptionBus<Payload> interface with subscribe, unsubscribe, broadcast, and asyncIterator.
BusCaller — identity (tenant + capabilities) used to enforce the ws.broadcast-cross-tenant capability gate.
BusSubscribeOptions and BusBroadcastOptions carry the optional tenant override and filter predicate.
InMemorySubscriptionBus — single-host default; keeps subscriptions indexed by (topic, tenant) and integrates with the registry's recordSubscription / dropSubscription hooks.
asyncIterator(topic, { tenant? }) — for await (const event of …) consumption for plugins that prefer pull-style.

Usage: Basic

import { InMemorySubscriptionBus } from "@suaveplan/api-ws/subscription-bus";

const bus = new InMemorySubscriptionBus<{ orderId: string }>();
bus.subscribe(connection, "orders.created");
bus.broadcast(
    { tenant: connection.tenantId, capabilities: new Set() },
    "orders.created",
    { tenant: connection.tenantId, payload: { orderId: "o1" } },
);

Usage: Advanced

import {
    InMemorySubscriptionBus,
    type BusCaller,
} from "@suaveplan/api-ws/subscription-bus";

interface Alert { priority: number }
const bus = new InMemorySubscriptionBus<Alert>();

bus.subscribe(highPriorityConn, "alerts", {
    filter: (alert) => alert.priority >= 5,
});

const caller: BusCaller = {
    tenant: orgA,
    capabilities: new Set(["ws.broadcast-cross-tenant"]),
};

bus.broadcast(caller, "alerts", {
    tenant: orgB,
    payload: { priority: 9 },
});

const iterator = bus.asyncIterator("alerts", { tenant: orgA });
for await (const evt of iterator) {
    console.log(evt.payload.priority);
    break;
}
await iterator.return?.();

API Reference

interface SubscriptionBus<Payload> { subscribe, unsubscribe, broadcast, asyncIterator }.
interface BusCaller { tenant: TenantId; capabilities: ReadonlySet<CapabilityId> }.
interface BusBroadcastOptions<Payload> { tenant, payload, filter? }.
interface BusSubscribeOptions<Payload> { tenant?, filter? }.
class InMemorySubscriptionBus<Payload> implements SubscriptionBus<Payload>.

Implementation Notes

The cross-tenant gate fires when opts.tenant !== caller.tenant AND the caller is missing ws.broadcast-cross-tenant. Same-tenant broadcasts never need the capability.
The bus and the registry share the (topic, tenant) indexing key format so a future audit pass can cross-check the two without a schema gymnastic.
The asyncIterator is unbounded — callers MUST await iterator.return?.() (or rely on for await to do so when the loop exits) to release the bus's listener slot.
Internal envelope ids are produced from the topic + a timestamp + a short random suffix; engines that need stronger guarantees can wrap the bus and override send in their own adapter.

3.6 KiB Raw Permalink Blame History

subscription-bus — typed fan-out with the cross-tenant capability gate