···11+/** Decorator that wraps a spaces StorageAdapter and publishes realtime events
22+ * after successful writes. Spaces and community modules stay unaware of
33+ * realtime; the decorator is the only integration seam. */
44+55+import type { StorageAdapter, SpaceMemberRow } from "../spaces/types";
66+import type { PubSub, RealtimeEvent } from "./types";
77+import { communityTopic, spaceTopic } from "./types";
88+99+export interface PublishingAdapterOptions {
1010+ /** Optional lookup: given a space's ownerDid, return true if that DID is a
1111+ * community in the local `communities` table. When provided, writes also
1212+ * publish to `community:<ownerDid>` so subscribers who expanded that alias
1313+ * at ticket-mint time receive the event.
1414+ *
1515+ * The lookup is expected to be cheap (cached in the caller) — the decorator
1616+ * calls it on every write. */
1717+ isCommunityDid?: (did: string) => Promise<boolean> | boolean;
1818+}
1919+2020+export function wrapWithPublishing(
2121+ inner: StorageAdapter,
2222+ pubsub: PubSub,
2323+ opts: PublishingAdapterOptions = {}
2424+): StorageAdapter {
2525+ const publishSpaceAndCommunity = async (
2626+ spaceUri: string,
2727+ ownerDid: string | null,
2828+ build: (topic: string) => RealtimeEvent
2929+ ): Promise<void> => {
3030+ await pubsub.publish(build(spaceTopic(spaceUri)));
3131+ if (ownerDid && opts.isCommunityDid && (await opts.isCommunityDid(ownerDid))) {
3232+ await pubsub.publish(build(communityTopic(ownerDid)));
3333+ }
3434+ };
3535+3636+ const ownerOf = async (spaceUri: string): Promise<string | null> => {
3737+ const s = await inner.getSpace(spaceUri);
3838+ return s?.ownerDid ?? null;
3939+ };
4040+4141+ const wrapped: StorageAdapter = {
4242+ ...inner,
4343+ createSpace: inner.createSpace.bind(inner),
4444+ getSpace: inner.getSpace.bind(inner),
4545+ listSpaces: inner.listSpaces.bind(inner),
4646+ deleteSpace: inner.deleteSpace.bind(inner),
4747+ updateSpaceAppPolicy: inner.updateSpaceAppPolicy.bind(inner),
4848+ getMember: inner.getMember.bind(inner),
4949+ listMembers: inner.listMembers.bind(inner),
5050+ createInvite: inner.createInvite.bind(inner),
5151+ listInvites: inner.listInvites.bind(inner),
5252+ revokeInvite: inner.revokeInvite.bind(inner),
5353+ getInvite: inner.getInvite.bind(inner),
5454+ redeemInvite: inner.redeemInvite.bind(inner),
5555+ getRecord: inner.getRecord.bind(inner),
5656+ listRecords: inner.listRecords.bind(inner),
5757+ listCollections: inner.listCollections.bind(inner),
5858+5959+ async addMember(spaceUri, did, addedBy) {
6060+ await inner.addMember(spaceUri, did, addedBy);
6161+ const owner = await ownerOf(spaceUri);
6262+ const now = Date.now();
6363+ await publishSpaceAndCommunity(spaceUri, owner, (topic) => ({
6464+ topic,
6565+ kind: "member.added",
6666+ payload: { spaceUri, did },
6767+ ts: now,
6868+ }));
6969+ },
7070+7171+ async removeMember(spaceUri, did) {
7272+ await inner.removeMember(spaceUri, did);
7373+ const owner = await ownerOf(spaceUri);
7474+ const now = Date.now();
7575+ await publishSpaceAndCommunity(spaceUri, owner, (topic) => ({
7676+ topic,
7777+ kind: "member.removed",
7878+ payload: { spaceUri, did },
7979+ ts: now,
8080+ }));
8181+ },
8282+8383+ async applyMembershipDiff(spaceUri, adds, removes, addedBy) {
8484+ await inner.applyMembershipDiff(spaceUri, adds, removes, addedBy);
8585+ if (adds.length === 0 && removes.length === 0) return;
8686+ const owner = await ownerOf(spaceUri);
8787+ const now = Date.now();
8888+ for (const did of adds) {
8989+ await publishSpaceAndCommunity(spaceUri, owner, (topic) => ({
9090+ topic,
9191+ kind: "member.added",
9292+ payload: { spaceUri, did },
9393+ ts: now,
9494+ }));
9595+ }
9696+ for (const did of removes) {
9797+ await publishSpaceAndCommunity(spaceUri, owner, (topic) => ({
9898+ topic,
9999+ kind: "member.removed",
100100+ payload: { spaceUri, did },
101101+ ts: now,
102102+ }));
103103+ }
104104+ },
105105+106106+ async putRecord(record) {
107107+ await inner.putRecord(record);
108108+ const owner = await ownerOf(record.spaceUri);
109109+ const now = Date.now();
110110+ await publishSpaceAndCommunity(record.spaceUri, owner, (topic) => ({
111111+ topic,
112112+ kind: "record.created",
113113+ payload: {
114114+ spaceUri: record.spaceUri,
115115+ collection: record.collection,
116116+ authorDid: record.authorDid,
117117+ rkey: record.rkey,
118118+ },
119119+ ts: now,
120120+ }));
121121+ },
122122+123123+ async deleteRecord(spaceUri, collection, authorDid, rkey) {
124124+ await inner.deleteRecord(spaceUri, collection, authorDid, rkey);
125125+ const owner = await ownerOf(spaceUri);
126126+ const now = Date.now();
127127+ await publishSpaceAndCommunity(spaceUri, owner, (topic) => ({
128128+ topic,
129129+ kind: "record.deleted",
130130+ payload: { spaceUri, collection, authorDid, rkey },
131131+ ts: now,
132132+ }));
133133+ },
134134+ };
135135+ return wrapped;
136136+}
137137+138138+// Keep this import hint for types that downstream code might pull from here.
139139+export type { SpaceMemberRow };
+90
src/core/realtime/resolve.ts
···11+/** Resolve a raw topic request (as given by a caller) to the concrete set of
22+ * delivery topics they are authorized to subscribe to.
33+ *
44+ * Rules (v1):
55+ * - `space:<uri>` → allowed iff caller is owner or a member of the space.
66+ * - `community:<did>` → expanded to `space:<uri>` for every space in the
77+ * community reachable by the caller (direct grants
88+ * or via delegation). `resolveReachableSpaces`
99+ * already has exactly this semantics.
1010+ * - `actor:<did>` → self-only in v1.
1111+ * - `collection:<nsid>` → rejected unless the deployment opts in
1212+ * (not yet implemented). */
1313+1414+import type { StorageAdapter } from "../spaces/types";
1515+import type { CommunityAdapter } from "../community/adapter";
1616+import { resolveReachableSpaces } from "../community/acl";
1717+import { spaceTopic, parseCommunityTopic, parseSpaceTopic } from "./types";
1818+1919+export interface TopicResolutionContext {
2020+ spaces: StorageAdapter;
2121+ /** May be null if the community module is not enabled. */
2222+ community: CommunityAdapter | null;
2323+}
2424+2525+export interface TopicResolution {
2626+ ok: true;
2727+ topics: string[];
2828+}
2929+3030+export interface TopicResolutionError {
3131+ ok: false;
3232+ error: "Forbidden" | "InvalidRequest" | "NotFound" | "NotSupported";
3333+ reason: string;
3434+}
3535+3636+export async function resolveTopicForCaller(
3737+ rawTopic: string,
3838+ callerDid: string,
3939+ ctx: TopicResolutionContext
4040+): Promise<TopicResolution | TopicResolutionError> {
4141+ // space:<uri>
4242+ const spaceUri = parseSpaceTopic(rawTopic);
4343+ if (spaceUri) {
4444+ const space = await ctx.spaces.getSpace(spaceUri);
4545+ if (!space) return { ok: false, error: "NotFound", reason: "space-not-found" };
4646+ if (space.ownerDid === callerDid) return { ok: true, topics: [rawTopic] };
4747+ const member = await ctx.spaces.getMember(spaceUri, callerDid);
4848+ if (!member) return { ok: false, error: "Forbidden", reason: "not-member" };
4949+ return { ok: true, topics: [rawTopic] };
5050+ }
5151+5252+ // community:<did>
5353+ const communityDid = parseCommunityTopic(rawTopic);
5454+ if (communityDid) {
5555+ if (!ctx.community) {
5656+ return { ok: false, error: "NotSupported", reason: "community-module-disabled" };
5757+ }
5858+ const row = await ctx.community.getCommunity(communityDid);
5959+ if (!row) return { ok: false, error: "NotFound", reason: "community-not-found" };
6060+ const reachable = await resolveReachableSpaces(ctx.community, callerDid);
6161+ // Filter to spaces owned by THIS community — reachable may include spaces
6262+ // from other communities via cross-community delegation.
6363+ const ownedList = await ctx.spaces.listSpaces({ ownerDid: communityDid, limit: 1000 });
6464+ const owned: Set<string> = new Set(ownedList.spaces.map((s) => s.uri));
6565+ const topics: string[] = [];
6666+ for (const uri of reachable) {
6767+ if (owned.has(uri)) topics.push(spaceTopic(uri));
6868+ }
6969+ if (topics.length === 0) {
7070+ return { ok: false, error: "Forbidden", reason: "no-reachable-spaces-in-community" };
7171+ }
7272+ return { ok: true, topics };
7373+ }
7474+7575+ // actor:<did> — self-only in v1.
7676+ if (rawTopic.startsWith("actor:")) {
7777+ const did = rawTopic.slice("actor:".length);
7878+ if (did !== callerDid) {
7979+ return { ok: false, error: "Forbidden", reason: "actor-self-only" };
8080+ }
8181+ return { ok: true, topics: [rawTopic] };
8282+ }
8383+8484+ // collection:<nsid> — public firehose, not exposed in v1.
8585+ if (rawTopic.startsWith("collection:")) {
8686+ return { ok: false, error: "NotSupported", reason: "collection-firehose-disabled" };
8787+ }
8888+8989+ return { ok: false, error: "InvalidRequest", reason: "unknown-topic" };
9090+}
+221
src/core/realtime/router.ts
···11+/** Realtime XRPC routes: ticket mint + subscribe (SSE | WS). */
22+33+import type { Context, Hono, MiddlewareHandler } from "hono";
44+import type { ContrailConfig } from "../types";
55+import type { ServiceAuth } from "../spaces/auth";
66+import type { StorageAdapter } from "../spaces/types";
77+import type { CommunityAdapter } from "../community/adapter";
88+import { InMemoryPubSub } from "./in-memory";
99+import { TicketSigner } from "./ticket";
1010+import { sseResponse } from "./sse";
1111+import { pumpWebSocket, type WebSocketLike } from "./websocket";
1212+import { mergeAsyncIterables } from "./merge";
1313+import { resolveTopicForCaller } from "./resolve";
1414+import type { PubSub, RealtimeEvent } from "./types";
1515+import { DEFAULT_TICKET_TTL_MS, DEFAULT_KEEPALIVE_MS } from "./types";
1616+1717+export interface RealtimeRoutesOptions {
1818+ /** Required: auth middleware for `<ns>.realtime.ticket` (JWT required).
1919+ * The subscribe endpoint uses the *ticket* path primarily, so it does not
2020+ * require this middleware — but bots can subscribe with Authorization
2121+ * header, in which case this middleware is used as a fallback. */
2222+ authMiddleware: MiddlewareHandler;
2323+ pubsub?: PubSub;
2424+}
2525+2626+/** WebSocketPair exists on Cloudflare Workers; on Node/Bun it's absent.
2727+ * When absent, a platform-provided WebSocket accept hook is used instead. */
2828+interface WebSocketPairCtor {
2929+ new (): { 0: WebSocketLike & { accept?: () => void }; 1: WebSocketLike & { accept?: () => void } };
3030+}
3131+3232+export function registerRealtimeRoutes(
3333+ app: Hono,
3434+ config: ContrailConfig,
3535+ spaces: StorageAdapter,
3636+ community: CommunityAdapter | null,
3737+ options: RealtimeRoutesOptions
3838+): void {
3939+ const cfg = config.realtime;
4040+ if (!cfg) return;
4141+4242+ const pubsub: PubSub = options.pubsub ?? cfg.pubsub ?? new InMemoryPubSub({ queueBound: cfg.queueBound });
4343+ const signer = new TicketSigner(cfg.ticketSecret);
4444+ const ticketTtl = cfg.ticketTtlMs ?? DEFAULT_TICKET_TTL_MS;
4545+ const keepaliveMs = cfg.keepaliveMs ?? DEFAULT_KEEPALIVE_MS;
4646+4747+ const NS = `${config.namespace}.realtime`;
4848+4949+ // POST /<ns>.realtime.ticket — { topic } → { ticket, topics, expiresAt }
5050+ app.post(`/xrpc/${NS}.ticket`, options.authMiddleware, async (c) => {
5151+ const sa = getAuth(c);
5252+ const body = (await c.req.json().catch(() => null)) as { topic?: string } | null;
5353+ if (!body?.topic) {
5454+ return c.json({ error: "InvalidRequest", message: "topic required" }, 400);
5555+ }
5656+ const resolved = await resolveTopicForCaller(body.topic, sa.issuer, { spaces, community });
5757+ if (!resolved.ok) {
5858+ const status = resolved.error === "NotFound" ? 404 : resolved.error === "Forbidden" ? 403 : 400;
5959+ return c.json({ error: resolved.error, reason: resolved.reason }, status);
6060+ }
6161+ const ticket = await signer.sign({
6262+ topics: resolved.topics,
6363+ did: sa.issuer,
6464+ ttlMs: ticketTtl,
6565+ });
6666+ return c.json({
6767+ ticket,
6868+ topics: resolved.topics,
6969+ expiresAt: Date.now() + ticketTtl,
7070+ });
7171+ });
7272+7373+ // GET /<ns>.realtime.subscribe — SSE or WS, driven by ?ticket= or JWT.
7474+ app.get(`/xrpc/${NS}.subscribe`, async (c) => {
7575+ const url = new URL(c.req.url);
7676+ const ticketParam = url.searchParams.get("ticket");
7777+ const collectionFilter = url.searchParams.get("collection");
7878+7979+ let callerDid: string;
8080+ let topics: string[];
8181+8282+ if (ticketParam) {
8383+ const payload = await signer.verify(ticketParam);
8484+ if (!payload) {
8585+ return c.json({ error: "AuthRequired", reason: "invalid-or-expired-ticket" }, 401);
8686+ }
8787+ callerDid = payload.did;
8888+ topics = payload.topics;
8989+ // Optional: narrow to topics the query explicitly requests.
9090+ const requested = url.searchParams.get("topic");
9191+ if (requested) {
9292+ if (!payload.topics.includes(requested)) {
9393+ return c.json({ error: "Forbidden", reason: "topic-not-in-ticket" }, 403);
9494+ }
9595+ topics = [requested];
9696+ }
9797+ } else {
9898+ // JWT path for server-side bots. Requires the auth middleware to have
9999+ // run and attached `serviceAuth` to the context — so we invoke it
100100+ // inline.
101101+ const runAuth = options.authMiddleware;
102102+ let authed = false;
103103+ await runAuth(c, async () => {
104104+ authed = true;
105105+ });
106106+ if (!authed) return c.res; // middleware already responded with 401
107107+ const sa = getAuth(c);
108108+ callerDid = sa.issuer;
109109+ const topic = url.searchParams.get("topic");
110110+ if (!topic) {
111111+ return c.json({ error: "InvalidRequest", message: "topic required" }, 400);
112112+ }
113113+ const resolved = await resolveTopicForCaller(topic, callerDid, { spaces, community });
114114+ if (!resolved.ok) {
115115+ const status = resolved.error === "NotFound" ? 404 : resolved.error === "Forbidden" ? 403 : 400;
116116+ return c.json({ error: resolved.error, reason: resolved.reason }, status);
117117+ }
118118+ topics = resolved.topics;
119119+ }
120120+121121+ if (topics.length === 0) {
122122+ return c.json({ error: "InvalidRequest", reason: "no-topics" }, 400);
123123+ }
124124+125125+ // Build the merged iterable, with an inline filter that closes the stream
126126+ // on a matching `member.removed` event (self-kick on revocation).
127127+ const ac = new AbortController();
128128+ const signals: AbortSignal[] = [ac.signal];
129129+ const reqSignal = c.req.raw.signal;
130130+ if (reqSignal) signals.push(reqSignal);
131131+ const combined = anySignal(signals);
132132+133133+ const sources = topics.map((t) => pubsub.subscribe(t, combined));
134134+ const merged = withSelfKickAndFilter(
135135+ mergeAsyncIterables(sources, combined),
136136+ callerDid,
137137+ collectionFilter,
138138+ ac
139139+ );
140140+141141+ // Content negotiation: Upgrade: websocket → WS, else SSE.
142142+ if (c.req.header("Upgrade")?.toLowerCase() === "websocket") {
143143+ const Pair = (globalThis as unknown as { WebSocketPair?: WebSocketPairCtor })
144144+ .WebSocketPair;
145145+ if (!Pair) {
146146+ return c.json(
147147+ { error: "NotSupported", reason: "websockets-require-worker-or-ws-adapter" },
148148+ 426
149149+ );
150150+ }
151151+ const pair = new Pair();
152152+ const clientWs = pair[0];
153153+ const serverWs = pair[1];
154154+ serverWs.accept?.();
155155+ // Pump in the background; don't await.
156156+ void pumpWebSocket(serverWs, merged, combined, { keepaliveMs });
157157+ return new Response(null, {
158158+ status: 101,
159159+ // Hono/undici-compat: some runtimes honor `webSocket` on the init.
160160+ // @ts-expect-error - Workers-specific init field
161161+ webSocket: clientWs,
162162+ });
163163+ }
164164+165165+ return sseResponse(merged, combined, { keepaliveMs });
166166+ });
167167+}
168168+169169+// ============================================================================
170170+// Helpers
171171+// ============================================================================
172172+173173+function getAuth(c: Context): ServiceAuth {
174174+ const a = c.get("serviceAuth") as ServiceAuth | undefined;
175175+ if (!a) throw new Error("service auth not set");
176176+ return a;
177177+}
178178+179179+/** Merge multiple AbortSignals into one. Aborts when any source aborts. */
180180+function anySignal(signals: AbortSignal[]): AbortSignal {
181181+ const ac = new AbortController();
182182+ for (const s of signals) {
183183+ if (s.aborted) {
184184+ ac.abort();
185185+ return ac.signal;
186186+ }
187187+ s.addEventListener("abort", () => ac.abort(), { once: true });
188188+ }
189189+ return ac.signal;
190190+}
191191+192192+/** Wrap an iterable: drop events that don't pass the collection filter (if
193193+ * any), and close the outer controller as soon as we see a `member.removed`
194194+ * for the caller's own DID. */
195195+function withSelfKickAndFilter(
196196+ source: AsyncIterable<RealtimeEvent>,
197197+ callerDid: string,
198198+ collectionFilter: string | null,
199199+ ac: AbortController
200200+): AsyncIterable<RealtimeEvent> {
201201+ return {
202202+ async *[Symbol.asyncIterator]() {
203203+ for await (const event of source) {
204204+ if (event.kind === "member.removed" && event.payload.did === callerDid) {
205205+ // Deliver the kick event so the client sees why, then close.
206206+ yield event;
207207+ ac.abort();
208208+ return;
209209+ }
210210+ if (
211211+ collectionFilter &&
212212+ (event.kind === "record.created" || event.kind === "record.deleted") &&
213213+ event.payload.collection !== collectionFilter
214214+ ) {
215215+ continue;
216216+ }
217217+ yield event;
218218+ }
219219+ },
220220+ };
221221+}
+98
src/core/realtime/sse.ts
···11+/** Server-Sent Events transport.
22+ *
33+ * Wraps an AsyncIterable<RealtimeEvent> as a streaming Response. The caller
44+ * (the router) has already done auth and has an AbortSignal it can use to
55+ * tear the stream down (e.g. on `member.removed` for the subscriber's DID). */
66+77+import type { RealtimeEvent } from "./types";
88+import { DEFAULT_KEEPALIVE_MS } from "./types";
99+1010+export interface SseOptions {
1111+ keepaliveMs?: number;
1212+ /** Called before the stream closes. Useful for cleanup that the caller
1313+ * can't do via the signal (e.g. removing a subscriber from a set). */
1414+ onClose?: () => void;
1515+}
1616+1717+export function sseResponse(
1818+ iter: AsyncIterable<RealtimeEvent>,
1919+ signal: AbortSignal,
2020+ opts: SseOptions = {}
2121+): Response {
2222+ const keepaliveMs = opts.keepaliveMs ?? DEFAULT_KEEPALIVE_MS;
2323+ const encoder = new TextEncoder();
2424+2525+ const stream = new ReadableStream<Uint8Array>({
2626+ start(controller) {
2727+ let closed = false;
2828+ let keepalive: ReturnType<typeof setInterval> | null = null;
2929+3030+ const close = () => {
3131+ if (closed) return;
3232+ closed = true;
3333+ if (keepalive) clearInterval(keepalive);
3434+ try {
3535+ controller.close();
3636+ } catch {
3737+ /* already closed */
3838+ }
3939+ opts.onClose?.();
4040+ };
4141+4242+ signal.addEventListener("abort", close, { once: true });
4343+4444+ keepalive = setInterval(() => {
4545+ if (closed) return;
4646+ try {
4747+ controller.enqueue(encoder.encode(`: keepalive\n\n`));
4848+ } catch {
4949+ close();
5050+ }
5151+ }, keepaliveMs);
5252+5353+ (async () => {
5454+ // Opening comment — helps some clients / proxies initialize promptly.
5555+ controller.enqueue(encoder.encode(`: open\n\n`));
5656+ try {
5757+ for await (const event of iter) {
5858+ if (closed) break;
5959+ controller.enqueue(encoder.encode(frameEvent(event)));
6060+ }
6161+ } catch (err) {
6262+ if (!closed) {
6363+ try {
6464+ controller.enqueue(
6565+ encoder.encode(
6666+ `event: error\ndata: ${JSON.stringify({
6767+ message: err instanceof Error ? err.message : String(err),
6868+ })}\n\n`
6969+ )
7070+ );
7171+ } catch {
7272+ /* stream already torn down */
7373+ }
7474+ }
7575+ } finally {
7676+ close();
7777+ }
7878+ })();
7979+ },
8080+ cancel() {
8181+ opts.onClose?.();
8282+ },
8383+ });
8484+8585+ return new Response(stream, {
8686+ status: 200,
8787+ headers: {
8888+ "content-type": "text/event-stream",
8989+ "cache-control": "no-cache, no-transform",
9090+ connection: "keep-alive",
9191+ "x-accel-buffering": "no",
9292+ },
9393+ });
9494+}
9595+9696+function frameEvent(event: RealtimeEvent): string {
9797+ return `event: ${event.kind}\ndata: ${JSON.stringify(event)}\n\n`;
9898+}
+159
src/core/realtime/ticket.ts
···11+/** Subscription tickets — HMAC-signed short-lived `{topics, did, exp}` blobs.
22+ *
33+ * Wire format: `<payload>.<sig>` where
44+ * payload = base64url(JSON({ topics, did, exp, iat }))
55+ * sig = base64url(HMAC-SHA256(key, payload))
66+ *
77+ * Tickets are integrity-only (not encrypted). Browsers use them because
88+ * EventSource / WebSocket can't send Authorization headers; server-side
99+ * consumers skip the ticket dance and send their JWT directly. */
1010+1111+export interface TicketPayload {
1212+ /** Concrete delivery topics this ticket authorizes. `community:<did>` is
1313+ * expanded to the caller's visible spaces before signing — a ticket never
1414+ * carries a community alias. */
1515+ topics: string[];
1616+ did: string;
1717+ /** Unix ms. */
1818+ exp: number;
1919+ /** Unix ms — useful for debugging; ignored on verify. */
2020+ iat: number;
2121+}
2222+2323+function normalizeSecret(secret: Uint8Array | string): Uint8Array {
2424+ if (typeof secret !== "string") {
2525+ if (secret.length !== 32) {
2626+ throw new Error(`realtime ticketSecret must be 32 bytes, got ${secret.length}`);
2727+ }
2828+ return secret;
2929+ }
3030+ // 64 hex chars would also round-trip as base64 (to 48 bytes). Prefer hex
3131+ // when the input matches the hex alphabet exactly; fall back to base64.
3232+ const hex = tryHex(secret);
3333+ if (hex && hex.length === 32) return hex;
3434+ const b64 = tryBase64(secret);
3535+ if (b64 && b64.length === 32) return b64;
3636+ if (hex || b64) {
3737+ const got = (hex ?? b64)!.length;
3838+ throw new Error(`realtime ticketSecret must decode to 32 bytes, got ${got}`);
3939+ }
4040+ throw new Error("realtime ticketSecret must be a 32-byte Uint8Array or base64/hex string");
4141+}
4242+4343+function tryBase64(s: string): Uint8Array | null {
4444+ try {
4545+ const normal = s.replace(/-/g, "+").replace(/_/g, "/");
4646+ const padded = normal + "=".repeat((4 - (normal.length % 4)) % 4);
4747+ if (!/^[A-Za-z0-9+/]*=*$/.test(padded)) return null;
4848+ const bin = atob(padded);
4949+ const out = new Uint8Array(bin.length);
5050+ for (let i = 0; i < bin.length; i++) out[i] = bin.charCodeAt(i);
5151+ return out;
5252+ } catch {
5353+ return null;
5454+ }
5555+}
5656+5757+function tryHex(s: string): Uint8Array | null {
5858+ if (!/^[0-9a-fA-F]+$/.test(s) || s.length % 2 !== 0) return null;
5959+ const out = new Uint8Array(s.length / 2);
6060+ for (let i = 0; i < out.length; i++) {
6161+ out[i] = parseInt(s.slice(i * 2, i * 2 + 2), 16);
6262+ }
6363+ return out;
6464+}
6565+6666+function b64urlFromBytes(bytes: Uint8Array): string {
6767+ let bin = "";
6868+ for (let i = 0; i < bytes.length; i++) bin += String.fromCharCode(bytes[i]!);
6969+ return btoa(bin).replace(/\+/g, "-").replace(/\//g, "_").replace(/=+$/, "");
7070+}
7171+7272+function b64urlToBytes(s: string): Uint8Array {
7373+ const normal = s.replace(/-/g, "+").replace(/_/g, "/");
7474+ const padded = normal + "=".repeat((4 - (normal.length % 4)) % 4);
7575+ const bin = atob(padded);
7676+ const out = new Uint8Array(bin.length);
7777+ for (let i = 0; i < bin.length; i++) out[i] = bin.charCodeAt(i);
7878+ return out;
7979+}
8080+8181+function b64urlFromString(s: string): string {
8282+ return b64urlFromBytes(new TextEncoder().encode(s));
8383+}
8484+8585+function stringFromB64url(s: string): string {
8686+ return new TextDecoder().decode(b64urlToBytes(s));
8787+}
8888+8989+function constantTimeEq(a: Uint8Array, b: Uint8Array): boolean {
9090+ if (a.length !== b.length) return false;
9191+ let diff = 0;
9292+ for (let i = 0; i < a.length; i++) diff |= a[i]! ^ b[i]!;
9393+ return diff === 0;
9494+}
9595+9696+export class TicketSigner {
9797+ private readonly keyPromise: Promise<CryptoKey>;
9898+9999+ constructor(secret: Uint8Array | string) {
100100+ const raw = normalizeSecret(secret);
101101+ this.keyPromise = crypto.subtle.importKey(
102102+ "raw",
103103+ raw as BufferSource,
104104+ { name: "HMAC", hash: "SHA-256" },
105105+ false,
106106+ ["sign", "verify"]
107107+ );
108108+ }
109109+110110+ async sign(input: { topics: string[]; did: string; ttlMs: number }): Promise<string> {
111111+ const now = Date.now();
112112+ const payload: TicketPayload = {
113113+ topics: input.topics,
114114+ did: input.did,
115115+ exp: now + input.ttlMs,
116116+ iat: now,
117117+ };
118118+ const payloadPart = b64urlFromString(JSON.stringify(payload));
119119+ const sig = await crypto.subtle.sign(
120120+ "HMAC",
121121+ await this.keyPromise,
122122+ new TextEncoder().encode(payloadPart) as BufferSource
123123+ );
124124+ const sigPart = b64urlFromBytes(new Uint8Array(sig));
125125+ return `${payloadPart}.${sigPart}`;
126126+ }
127127+128128+ /** Returns the decoded payload if the ticket is valid + unexpired, else null. */
129129+ async verify(ticket: string): Promise<TicketPayload | null> {
130130+ const dot = ticket.indexOf(".");
131131+ if (dot < 0) return null;
132132+ const payloadPart = ticket.slice(0, dot);
133133+ const sigPart = ticket.slice(dot + 1);
134134+ let expectedSig: Uint8Array;
135135+ try {
136136+ expectedSig = b64urlToBytes(sigPart);
137137+ } catch {
138138+ return null;
139139+ }
140140+ const computedRaw = await crypto.subtle.sign(
141141+ "HMAC",
142142+ await this.keyPromise,
143143+ new TextEncoder().encode(payloadPart) as BufferSource
144144+ );
145145+ const computed = new Uint8Array(computedRaw);
146146+ if (!constantTimeEq(expectedSig, computed)) return null;
147147+ let parsed: TicketPayload;
148148+ try {
149149+ parsed = JSON.parse(stringFromB64url(payloadPart));
150150+ } catch {
151151+ return null;
152152+ }
153153+ if (!parsed || !Array.isArray(parsed.topics) || typeof parsed.did !== "string") {
154154+ return null;
155155+ }
156156+ if (typeof parsed.exp !== "number" || parsed.exp <= Date.now()) return null;
157157+ return parsed;
158158+ }
159159+}
+95
src/core/realtime/types.ts
···11+/** Realtime module — canonical types + interfaces. See docs/realtime.md. */
22+33+/** Discriminated union of every event kind that flows through the PubSub. */
44+export type RealtimeEvent =
55+ | {
66+ topic: string;
77+ kind: "record.created";
88+ payload: { spaceUri: string; collection: string; authorDid: string; rkey: string };
99+ ts: number;
1010+ }
1111+ | {
1212+ topic: string;
1313+ kind: "record.deleted";
1414+ payload: { spaceUri: string; collection: string; authorDid: string; rkey: string };
1515+ ts: number;
1616+ }
1717+ | {
1818+ topic: string;
1919+ kind: "member.added";
2020+ payload: { spaceUri: string; did: string };
2121+ ts: number;
2222+ }
2323+ | {
2424+ topic: string;
2525+ kind: "member.removed";
2626+ payload: { spaceUri: string; did: string };
2727+ ts: number;
2828+ };
2929+3030+export type RealtimeEventKind = RealtimeEvent["kind"];
3131+3232+/** Core pubsub abstraction. Implementations: InMemoryPubSub, DurableObjectPubSub. */
3333+export interface PubSub {
3434+ publish(event: RealtimeEvent): Promise<void>;
3535+ /** Stream events on the topic until the signal aborts (or the iterator is
3636+ * returned/broken out of). Implementations use a bounded per-subscriber
3737+ * queue with drop-oldest semantics — a slow subscriber can't stall publishers. */
3838+ subscribe(topic: string, signal?: AbortSignal): AsyncIterable<RealtimeEvent>;
3939+}
4040+4141+// ---- Canonical topic strings -----------------------------------------------
4242+// `community:<did>` is an alias resolved at ticket-mint time to the concrete
4343+// set of `space:<uri>` topics the caller can see; it is never a real delivery
4444+// topic. The other three are real.
4545+4646+export function spaceTopic(uri: string): string {
4747+ return `space:${uri}`;
4848+}
4949+5050+export function communityTopic(did: string): string {
5151+ return `community:${did}`;
5252+}
5353+5454+export function collectionTopic(nsid: string): string {
5555+ return `collection:${nsid}`;
5656+}
5757+5858+export function actorTopic(did: string): string {
5959+ return `actor:${did}`;
6060+}
6161+6262+export function isCommunityTopic(topic: string): boolean {
6363+ return topic.startsWith("community:");
6464+}
6565+6666+export function parseCommunityTopic(topic: string): string | null {
6767+ return isCommunityTopic(topic) ? topic.slice("community:".length) : null;
6868+}
6969+7070+export function parseSpaceTopic(topic: string): string | null {
7171+ return topic.startsWith("space:") ? topic.slice("space:".length) : null;
7272+}
7373+7474+// ---- Config -----------------------------------------------------------------
7575+7676+export interface RealtimeConfig {
7777+ /** Backing pubsub. Default: new InMemoryPubSub() (single-process only). On
7878+ * Workers, pass `new DurableObjectPubSub(env.REALTIME)`. */
7979+ pubsub?: PubSub;
8080+ /** HMAC secret used to sign subscription tickets. 32 bytes. Accepts raw
8181+ * Uint8Array or base64 / hex string. Envelope-encrypts nothing — tickets
8282+ * are integrity-only, not confidential. */
8383+ ticketSecret: Uint8Array | string;
8484+ /** Ticket lifetime in ms. Default 120_000 (2 minutes). */
8585+ ticketTtlMs?: number;
8686+ /** SSE/WS keepalive interval in ms. Default 15_000. */
8787+ keepaliveMs?: number;
8888+ /** Per-subscriber queue bound. Default 1024. Events beyond this are dropped
8989+ * oldest-first and the subscriber receives a `lag` signal (out of band). */
9090+ queueBound?: number;
9191+}
9292+9393+export const DEFAULT_TICKET_TTL_MS = 120_000;
9494+export const DEFAULT_KEEPALIVE_MS = 15_000;
9595+export const DEFAULT_QUEUE_BOUND = 1024;
+84
src/core/realtime/websocket.ts
···11+/** WebSocket transport.
22+ *
33+ * Accepts a new WebSocket connection (either via `WebSocketPair` on Workers
44+ * or a platform-provided server-side socket) and pumps events to it from an
55+ * AsyncIterable<RealtimeEvent>. Messages are UTF-8 JSON, one event per frame.
66+ *
77+ * Close codes (subset, RFC 6455 + app-custom):
88+ * - 4001: server error pumping
99+ * - 4003: membership revoked
1010+ * - 4008: ticket/auth invalid (used by the router, not here)
1111+ */
1212+1313+import type { RealtimeEvent } from "./types";
1414+import { DEFAULT_KEEPALIVE_MS } from "./types";
1515+1616+export interface WebSocketLike {
1717+ send(data: string): void;
1818+ close(code?: number, reason?: string): void;
1919+ addEventListener(type: "message" | "close" | "error", listener: (ev: any) => void): void;
2020+}
2121+2222+export interface WebSocketPumpOptions {
2323+ keepaliveMs?: number;
2424+ onClose?: () => void;
2525+}
2626+2727+/** Pump events from `iter` to `ws` until the signal aborts or the iter ends.
2828+ * Caller is responsible for having already accept()ed the socket. */
2929+export async function pumpWebSocket(
3030+ ws: WebSocketLike,
3131+ iter: AsyncIterable<RealtimeEvent>,
3232+ signal: AbortSignal,
3333+ opts: WebSocketPumpOptions = {}
3434+): Promise<void> {
3535+ const keepaliveMs = opts.keepaliveMs ?? DEFAULT_KEEPALIVE_MS;
3636+ let closed = false;
3737+3838+ const close = (code?: number, reason?: string) => {
3939+ if (closed) return;
4040+ closed = true;
4141+ try {
4242+ ws.close(code, reason);
4343+ } catch {
4444+ /* already closed */
4545+ }
4646+ opts.onClose?.();
4747+ };
4848+4949+ ws.addEventListener("close", () => {
5050+ closed = true;
5151+ opts.onClose?.();
5252+ });
5353+ ws.addEventListener("error", () => {
5454+ closed = true;
5555+ opts.onClose?.();
5656+ });
5757+ signal.addEventListener("abort", () => close(1000, "aborted"), { once: true });
5858+5959+ const keepalive = setInterval(() => {
6060+ if (closed) return;
6161+ try {
6262+ ws.send(JSON.stringify({ kind: "$keepalive" }));
6363+ } catch {
6464+ close();
6565+ }
6666+ }, keepaliveMs);
6767+6868+ try {
6969+ for await (const event of iter) {
7070+ if (closed) break;
7171+ try {
7272+ ws.send(JSON.stringify(event));
7373+ } catch {
7474+ close(4001, "send-failed");
7575+ break;
7676+ }
7777+ }
7878+ } catch {
7979+ close(4001, "pump-error");
8080+ } finally {
8181+ clearInterval(keepalive);
8282+ close();
8383+ }
8484+}
+54-1
src/core/router/index.ts
···1414import type { ServiceJwtVerifier } from "@atcute/xrpc-server/auth";
1515import { registerCommunityRoutes } from "../community/router";
1616import type { CommunityRoutesOptions } from "../community/router";
1717+import { CommunityAdapter } from "../community/adapter";
1818+import { registerRealtimeRoutes } from "../realtime/router";
1919+import type { RealtimeRoutesOptions } from "../realtime/router";
2020+import { InMemoryPubSub } from "../realtime/in-memory";
2121+import { wrapWithPublishing } from "../realtime/publishing-adapter";
2222+import type { PubSub } from "../realtime/types";
1723import { resolveActor } from "../identity";
1824import { resolveProfiles } from "./profiles";
1925import { backfillUser } from "../backfill";
···2632export interface CreateAppOptions {
2733 spaces?: SpacesRoutesOptions;
2834 community?: CommunityRoutesOptions;
3535+ realtime?: Partial<RealtimeRoutesOptions>;
2936 /** Separate DB for the spaces tables. Defaults to `db`. */
3037 spacesDb?: Database;
3138 /** Full spaces context override (escape hatch for tests). */
···7279 // Shared spaces context — verifier + adapter — reused by both the per-collection
7380 // routes (for `?spaceUri=...` dispatch) and the `<ns>.space.*` routes.
7481 const spacesDb = options.spacesDb ?? db;
7575- const spacesCtx: SpacesContext | null =
8282+ let spacesCtx: SpacesContext | null =
7683 options.spacesCtx !== undefined
7784 ? options.spacesCtx
7885 : config.spaces
···8289 }
8390 : null;
84919292+ // Realtime pubsub is built up-front so the publishing decorator can reference
9393+ // it. The subscribe endpoint uses the same instance.
9494+ let realtimePubsub: PubSub | null = null;
9595+ if (config.realtime && spacesCtx) {
9696+ realtimePubsub =
9797+ options.realtime?.pubsub ?? config.realtime.pubsub ?? new InMemoryPubSub({
9898+ queueBound: config.realtime.queueBound,
9999+ });
100100+ const communityAdapter = config.community ? new CommunityAdapter(spacesDb) : null;
101101+ const isCommunityDid = communityAdapter
102102+ ? cachedIsCommunityDid(communityAdapter)
103103+ : undefined;
104104+ spacesCtx = {
105105+ ...spacesCtx,
106106+ adapter: wrapWithPublishing(spacesCtx.adapter, realtimePubsub, { isCommunityDid }),
107107+ };
108108+ }
109109+85110 registerAdminRoutes(app, db, config);
86111 registerCollectionRoutes(app, db, config, spacesCtx);
87112 registerFeedRoutes(app, db, config);
···103128 );
104129 }
105130131131+ if (config.realtime && spacesCtx && realtimePubsub) {
132132+ const authMiddleware =
133133+ options.realtime?.authMiddleware ??
134134+ options.spaces?.authMiddleware ??
135135+ createServiceAuthMiddleware(spacesCtx.verifier);
136136+ const communityAdapter = config.community ? new CommunityAdapter(spacesDb) : null;
137137+ registerRealtimeRoutes(app, config, spacesCtx.adapter, communityAdapter, {
138138+ authMiddleware,
139139+ pubsub: realtimePubsub,
140140+ });
141141+ }
142142+106143 return app;
107144}
145145+146146+function cachedIsCommunityDid(
147147+ community: CommunityAdapter
148148+): (did: string) => Promise<boolean> {
149149+ const TTL = 60_000;
150150+ const cache = new Map<string, { value: boolean; expires: number }>();
151151+ return async (did: string) => {
152152+ const now = Date.now();
153153+ const hit = cache.get(did);
154154+ if (hit && hit.expires > now) return hit.value;
155155+ const row = await community.getCommunity(did);
156156+ const value = row != null;
157157+ cache.set(did, { value, expires: now + TTL });
158158+ return value;
159159+ };
160160+}
+3
src/core/types.ts
···158158 /** Community module configuration. When set, the service exposes community XRPCs
159159 * for managing community-owned spaces and tiered access levels. Requires `spaces`. */
160160 community?: import("./community/types").CommunityConfig;
161161+ /** Realtime module configuration. When set, the service exposes ticket + SSE/WS
162162+ * subscribe XRPCs, and wraps the spaces adapter to publish events after writes. */
163163+ realtime?: import("./realtime/types").RealtimeConfig;
161164 /** Customize the auto-generated `<namespace>.permissionSet` lexicon. */
162165 permissionSet?: PermissionSetConfig;
163166}
+53
src/generate.ts
···5151 return null;
5252}
53535454+function findRealtimeTemplatesDir(rootDir: string): string | null {
5555+ const candidates = [
5656+ join(rootDir, "realtime-lexicon-templates"),
5757+ join(rootDir, "node_modules/@atmo-dev/contrail/realtime-lexicon-templates"),
5858+ ];
5959+ for (const p of candidates) {
6060+ if (existsSync(p)) return p;
6161+ }
6262+ return null;
6363+}
6464+5465/** Yield all JSON files under a directory (recursive). */
5566function* walkJson(dir: string): Generator<string> {
5667 for (const entry of readdirSync(dir, { withFileTypes: true })) {
···753764 for (const [k, v] of Object.entries(obj)) {
754765 if (k === "ref" && typeof v === "string" && v.startsWith("tools.atmo.community")) {
755766 out[k] = v.replace(/^tools\.atmo\.community/, `${ns}.community`);
767767+ } else if (k === "id" && typeof v === "string" && templateIdRe.test(v)) {
768768+ out[k] = idReplace(v);
769769+ } else {
770770+ out[k] = rewriteRefs(v);
771771+ }
772772+ }
773773+ return out;
774774+ }
775775+ return obj;
776776+ };
777777+778778+ for (const file of walkJson(templatesDir)) {
779779+ const doc = JSON.parse(readFileSync(file, "utf-8"));
780780+ if (typeof doc.id !== "string" || !templateIdRe.test(doc.id)) continue;
781781+ const newId = idReplace(doc.id);
782782+ const rewritten = rewriteRefs({ ...doc, id: newId });
783783+ writeLexicon(newId, rewritten);
784784+ }
785785+ }
786786+ }
787787+788788+ // --- Realtime: instantiate library templates under <ns>.realtime.* ---
789789+790790+ if (config.realtime) {
791791+ log("Generating realtime endpoints...");
792792+ const templatesDir = findRealtimeTemplatesDir(rootDir);
793793+ if (!templatesDir) {
794794+ log(" (realtime templates not found — skipping)");
795795+ } else {
796796+ const templateIdRe = /^tools\.atmo\.realtime(\.[A-Za-z0-9.]+)?$/;
797797+ const idReplace = (id: string) =>
798798+ id.startsWith("tools.atmo.realtime")
799799+ ? id.replace(/^tools\.atmo\.realtime/, `${ns}.realtime`)
800800+ : id;
801801+802802+ const rewriteRefs = (obj: any): any => {
803803+ if (Array.isArray(obj)) return obj.map(rewriteRefs);
804804+ if (obj && typeof obj === "object") {
805805+ const out: any = {};
806806+ for (const [k, v] of Object.entries(obj)) {
807807+ if (k === "ref" && typeof v === "string" && v.startsWith("tools.atmo.realtime")) {
808808+ out[k] = v.replace(/^tools\.atmo\.realtime/, `${ns}.realtime`);
756809 } else if (k === "id" && typeof v === "string" && templateIdRe.test(v)) {
757810 out[k] = idReplace(v);
758811 } else {