diff --git a/packages/chat/src/workbench-events.ts b/packages/chat/src/workbench-events.ts index eb838adb6..65c3ce59b 100644 --- a/packages/chat/src/workbench-events.ts +++ b/packages/chat/src/workbench-events.ts @@ -13,13 +13,21 @@ // stream immediately rather than leaving it registered until // `stream.onAbort` eventually runs — a dangling subscriber between // disconnect and abort is a zombie: every event published in that -// window still attempts a write. In practice Hono's own -// `StreamingApi.write` swallows writer errors internally and never -// rejects, so `stream.onAbort` (wired by the route) is the only -// signal that actually fires for a disconnected client today; this -// path exists for any write that does throw (a custom `stream` -// passed in tests, or a future Hono release that stops swallowing) -// and is otherwise inert rather than load-bearing (CL-7197). +// window still attempts a write. Hono's own `StreamingApi.write` +// swallows writer errors internally and never rejects, so a +// `writeSSE` that "succeeds" is not proof the bytes landed: after +// every write we check whether the underlying writer is errored +// (`desiredSize === null`) or the stream aborted/closed, and we +// bound the write itself with a timeout so a stalled TCP sink +// cannot pin the delivery queue forever (CL-7246). A write that +// does throw is still handled the same way. Stall teardown aborts +// the underlying writer rather than awaiting Hono `close()`, which +// queues behind the same in-flight write and never drops the +// socket. A clean end (revoked access) still `close()`s — aborting a +// healthy Hono stream errors the body with `undefined` instead of a +// normal SSE disconnect. `stream.onAbort` (wired by the route) remains the +// disconnect path for a client that goes away between writes; a +// clean abort already torn down that way is not a new incident. // 2. Access must be re-checked on every delivered event, not only at // connect time. The route resolves access once before opening the // stream, but a share or a share member's row can be revoked at @@ -62,6 +70,70 @@ const DEFAULT_KEEPALIVE_INTERVAL_MS = 25_000; // started. const MAX_QUEUED_DELIVERIES = 200; +// Hono's `StreamingApi.write` awaits `writer.write` inside a bare +// `catch {}`, so a failed or stalled sink never rejects `writeSSE`. +// Bound every write so a client that has stopped reading cannot pin +// the delivery queue; well below the keepalive interval, since a +// write of a small SSE frame that has not flushed in this long is +// already a dead connection, not a slow one. +const DEFAULT_WRITE_TIMEOUT_MS = 10_000; + +function underlyingWriterIsErrored(stream: SSEStreamingApi): boolean { + const writer = Reflect.get(stream, "writer"); + if (typeof writer !== "object" || writer === null) return false; + if (!("desiredSize" in writer)) return false; + return writer.desiredSize === null; +} + +function abortUnderlyingWriter(stream: SSEStreamingApi): boolean { + const writer = Reflect.get(stream, "writer"); + if (typeof writer !== "object" || writer === null) return false; + const abort = Reflect.get(writer, "abort"); + if (typeof abort !== "function") return false; + try { + void Promise.resolve(abort.call(writer)).catch(() => undefined); + return true; + } catch { + // report-error-ignore: aborting an already-closed writer is not an incident + return true; + } +} + +function closeStream(stream: SSEStreamingApi): void { + void stream.close().catch(() => undefined); +} + +function dropHungStream(stream: SSEStreamingApi): void { + if (abortUnderlyingWriter(stream)) return; + closeStream(stream); +} + +async function writeSSEObservingFailure( + stream: SSEStreamingApi, + message: { event?: string; data: string }, + timeoutMs: number, +): Promise { + let timeoutId: ReturnType | undefined; + const write = stream.writeSSE(message).then(() => { + if (stream.aborted || stream.closed || underlyingWriterIsErrored(stream)) { + throw new Error("workbench SSE write failed"); + } + }); + const stalled = new Promise((_, reject) => { + timeoutId = setTimeout(() => { + reject(new Error("workbench SSE write stalled")); + }, timeoutMs); + }); + try { + await Promise.race([write, stalled]); + } finally { + if (timeoutId !== undefined) clearTimeout(timeoutId); + // If the timeout won, the in-flight write may still reject later; + // swallow that so it cannot surface as an unhandled rejection. + void write.catch(() => undefined); + } +} + export type WorkbenchSubscriber = (event: ChatWorkbenchEvent) => void; export interface WorkbenchSubscriberRegistry { @@ -173,11 +245,17 @@ export interface WorkbenchStreamBridge { * member stops receiving events on the very next one published. A * `writeSSE` that throws (the client is already gone) unsubscribes * that source immediately, closes the stream, and reports the - * failure — Hono's own `write` swallows writer errors today, so in - * practice `stream.onAbort` is what actually catches a disconnected - * client, but this path covers any write that does throw rather than - * silently degrading. A periodic keepalive keeps idle connections - * alive behind proxies that time out on silence. + * failure. Hono's own `write` swallows writer errors today, so a + * resolved `writeSSE` is inspected for an errored writer, an aborted + * or closed stream, or a write that never settles within + * `writeTimeoutMs` — any of those is treated as the same failure. A + * stalled write aborts the underlying writer so the client's + * EventSource sees a disconnect; Hono `close()` is not awaited + * because it queues behind the in-flight write. A client abort that + * already ran `teardown` via `stream.onAbort` is not reported as a + * new incident, but still aborts the in-flight writer so Hono + * `close()` does not wait on it. A periodic keepalive keeps idle + * connections alive behind proxies that time out on silence. */ export function bridgeWorkbenchStream(input: { registry: WorkbenchSubscriberRegistry; @@ -202,6 +280,8 @@ export function bridgeWorkbenchStream(input: { }; /** Overrides `DEFAULT_KEEPALIVE_INTERVAL_MS`; exists for tests. */ keepaliveIntervalMs?: number; + /** Overrides `DEFAULT_WRITE_TIMEOUT_MS`; exists for tests. */ + writeTimeoutMs?: number; }): WorkbenchStreamBridge { let tornDown = false; let resolveClosed: () => void = () => undefined; @@ -238,7 +318,7 @@ export function bridgeWorkbenchStream(input: { if (keepaliveTimer !== undefined) clearInterval(keepaliveTimer); resolveClosed(); }; - const closeStream = () => input.stream.close().catch(() => undefined); + const writeTimeoutMs = input.writeTimeoutMs ?? DEFAULT_WRITE_TIMEOUT_MS; // Every write to this stream — the presence snapshot, each delivered // event, and the keepalive ping — is chained onto this one promise so @@ -257,7 +337,7 @@ export function bridgeWorkbenchStream(input: { roomId: input.workbenchId, }); teardown(); - void closeStream(); + dropHungStream(input.stream); return; } queuedCount += 1; @@ -283,26 +363,32 @@ export function bridgeWorkbenchStream(input: { roomId: input.workbenchId, }); teardown(); - await closeStream(); + closeStream(input.stream); return; } if (!authorized) { teardown(); - await closeStream(); + closeStream(input.stream); return; } try { - await input.stream.writeSSE({ - event: event.type, - data: JSON.stringify(event.data), - }); + await writeSSEObservingFailure( + input.stream, + { + event: event.type, + data: JSON.stringify(event.data), + }, + writeTimeoutMs, + ); } catch (error) { - reportError(error, { - operation: "chat.workbenchStream.write", - roomId: input.workbenchId, - }); + if (!(tornDown || input.stream.aborted)) { + reportError(error, { + operation: "chat.workbenchStream.write", + roomId: input.workbenchId, + }); + } teardown(); - await closeStream(); + dropHungStream(input.stream); } }; @@ -317,17 +403,23 @@ export function bridgeWorkbenchStream(input: { }); enqueue(async () => { try { - await input.stream.writeSSE({ - event: "chat.presence.snapshot", - data: JSON.stringify(snapshot), - }); + await writeSSEObservingFailure( + input.stream, + { + event: "chat.presence.snapshot", + data: JSON.stringify(snapshot), + }, + writeTimeoutMs, + ); } catch (error) { - reportError(error, { - operation: "chat.workbenchStream.presenceSnapshot", - roomId: input.workbenchId, - }); + if (!(tornDown || input.stream.aborted)) { + reportError(error, { + operation: "chat.workbenchStream.presenceSnapshot", + roomId: input.workbenchId, + }); + } teardown(); - await closeStream(); + dropHungStream(input.stream); } }); } @@ -376,14 +468,20 @@ export function bridgeWorkbenchStream(input: { if (tornDown || queuedCount > 0) return; enqueue(async () => { try { - await input.stream.writeSSE({ event: "keepalive", data: "" }); + await writeSSEObservingFailure( + input.stream, + { event: "keepalive", data: "" }, + writeTimeoutMs, + ); } catch (error) { - reportError(error, { - operation: "chat.workbenchStream.keepalive", - roomId: input.workbenchId, - }); + if (!(tornDown || input.stream.aborted)) { + reportError(error, { + operation: "chat.workbenchStream.keepalive", + roomId: input.workbenchId, + }); + } teardown(); - await closeStream(); + dropHungStream(input.stream); } }); }, input.keepaliveIntervalMs ?? DEFAULT_KEEPALIVE_INTERVAL_MS); diff --git a/packages/chat/test/workbench-events.test.ts b/packages/chat/test/workbench-events.test.ts index 574aee2e3..997a09d47 100644 --- a/packages/chat/test/workbench-events.test.ts +++ b/packages/chat/test/workbench-events.test.ts @@ -226,6 +226,182 @@ describe("bridgeWorkbenchStream", () => { expect(closeCount).toBe(1); }); + test("a writeSSE that resolves while the underlying writer is errored (Hono swallows writer failures) removes that subscriber and closes the stream", async () => { + const registry = createWorkbenchSubscriberRegistry(); + let writeCount = 0; + let closeCount = 0; + const stream = Object.assign( + fakeStream( + () => { + writeCount += 1; + return Promise.resolve(); + }, + () => { + closeCount += 1; + return Promise.resolve(); + }, + ), + { writer: { desiredSize: null } }, + ); + const report = spyOn(errorSink, "reportError").mockReturnValue("ref_test"); + + bridgeWorkbenchStream({ + registry, + platform: noopPlatformEvents(), + workbenchId: "chan_1", + stream, + authorize: alwaysAuthorized, + }); + + registry.publish("chan_1", { type: "chat.typing", data: {} }); + await flush(); + + registry.publish("chan_1", { type: "chat.typing", data: {} }); + await flush(); + + // Hono's StreamingApi.write catches the writer's rejection and + // resolves anyway, so the throw-path test above never fires on a + // real stream. The errored writer (`desiredSize === null`) is the + // signal that the write actually failed (CL-7246). + expect(writeCount).toBe(1); + expect(closeCount).toBe(1); + expect(report).toHaveBeenCalledTimes(1); + expect(report.mock.calls[0]?.[0]).toEqual( + expect.objectContaining({ message: "workbench SSE write failed" }), + ); + expect(report.mock.calls[0]?.[1]).toMatchObject({ + operation: "chat.workbenchStream.write", + roomId: "chan_1", + }); + }); + + test("a writeSSE that never resolves is treated as stalled, reported, and closes the stream", async () => { + const registry = createWorkbenchSubscriberRegistry(); + let writeCount = 0; + let closeCount = 0; + const stream = fakeStream( + () => { + writeCount += 1; + return new Promise(() => undefined); + }, + () => { + closeCount += 1; + return Promise.resolve(); + }, + ); + const report = spyOn(errorSink, "reportError").mockReturnValue("ref_test"); + + bridgeWorkbenchStream({ + registry, + platform: noopPlatformEvents(), + workbenchId: "chan_1", + stream, + authorize: alwaysAuthorized, + writeTimeoutMs: 20, + }); + + registry.publish("chan_1", { type: "chat.typing", data: {} }); + await new Promise((resolve) => setTimeout(resolve, 50)); + + registry.publish("chan_1", { type: "chat.typing", data: {} }); + await flush(); + + expect(writeCount).toBe(1); + expect(closeCount).toBe(1); + expect(report).toHaveBeenCalledTimes(1); + expect(report.mock.calls[0]?.[0]).toEqual( + expect.objectContaining({ message: "workbench SSE write stalled" }), + ); + expect(report.mock.calls[0]?.[1]).toMatchObject({ + operation: "chat.workbenchStream.write", + roomId: "chan_1", + }); + }); + + test("a stalled write aborts the writer instead of awaiting a hung close()", async () => { + const registry = createWorkbenchSubscriberRegistry(); + let abortCount = 0; + let closeCount = 0; + const stream = Object.assign( + fakeStream( + () => new Promise(() => undefined), + () => { + closeCount += 1; + return new Promise(() => undefined); + }, + ), + { + writer: { + desiredSize: 1, + abort: () => { + abortCount += 1; + }, + }, + }, + ); + const report = spyOn(errorSink, "reportError").mockReturnValue("ref_test"); + + const { closed } = bridgeWorkbenchStream({ + registry, + platform: noopPlatformEvents(), + workbenchId: "chan_1", + stream, + authorize: alwaysAuthorized, + writeTimeoutMs: 20, + }); + + registry.publish("chan_1", { type: "chat.typing", data: {} }); + + await Promise.race([ + closed, + new Promise((_, reject) => + setTimeout(() => reject(new Error("hung on close()")), 200), + ), + ]); + + expect(abortCount).toBe(1); + expect(closeCount).toBe(0); + expect(report).toHaveBeenCalledTimes(1); + expect(report.mock.calls[0]?.[0]).toEqual( + expect.objectContaining({ message: "workbench SSE write stalled" }), + ); + }); + + test("a client abort mid-write does not reportError after onAbort teardown", async () => { + const registry = createWorkbenchSubscriberRegistry(); + let abortCount = 0; + const stream = Object.assign( + fakeStream(() => new Promise(() => undefined)), + { aborted: false }, + ); + Reflect.set(stream, "writer", { + desiredSize: 1, + abort: () => { + abortCount += 1; + }, + }); + const report = spyOn(errorSink, "reportError").mockReturnValue("ref_test"); + + const { teardown } = bridgeWorkbenchStream({ + registry, + platform: noopPlatformEvents(), + workbenchId: "chan_1", + stream, + authorize: alwaysAuthorized, + writeTimeoutMs: 20, + }); + + registry.publish("chan_1", { type: "chat.typing", data: {} }); + await flush(); + + stream.aborted = true; + teardown(); + await new Promise((resolve) => setTimeout(resolve, 50)); + + expect(report).not.toHaveBeenCalled(); + expect(abortCount).toBe(1); + }); + test("a platform subscribe that throws still opens the stream, registry-only, and reports the failure", async () => { const registry = createWorkbenchSubscriberRegistry(); const writes: unknown[] = []; @@ -310,6 +486,46 @@ describe("bridgeWorkbenchStream", () => { expect(closeCount).toBe(1); }); + test("authorize going false closes a real Hono-shaped stream instead of aborting the writer", async () => { + const registry = createWorkbenchSubscriberRegistry(); + let abortCount = 0; + let closeCount = 0; + const stream = Object.assign( + fakeStream( + () => Promise.resolve(), + () => { + closeCount += 1; + return Promise.resolve(); + }, + ), + { + abort: () => { + abortCount += 1; + }, + writer: { + desiredSize: 1, + abort: () => { + abortCount += 1; + }, + }, + }, + ); + + bridgeWorkbenchStream({ + registry, + platform: noopPlatformEvents(), + workbenchId: "chan_1", + stream, + authorize: () => Promise.resolve(false), + }); + + registry.publish("chan_1", { type: "chat.typing", data: {} }); + await flush(); + + expect(closeCount).toBe(1); + expect(abortCount).toBe(0); + }); + test("an authorize call that rejects is caught, reported, and closes the stream", async () => { const registry = createWorkbenchSubscriberRegistry(); let closeCount = 0; diff --git a/packages/chat/test/workbench-share-routes.test.ts b/packages/chat/test/workbench-share-routes.test.ts index 5450da9fd..e9841f104 100644 --- a/packages/chat/test/workbench-share-routes.test.ts +++ b/packages/chat/test/workbench-share-routes.test.ts @@ -403,7 +403,8 @@ describe("shared workbench projection", () => { .read() .then((result) => result.done ? undefined : decoder.decode(result.value), - ), + ) + .catch(() => undefined), new Promise((resolve) => setTimeout(() => resolve(undefined), timeoutMs), ), diff --git a/packages/presence/src/routes.ts b/packages/presence/src/routes.ts index da1179eb8..cf77273c5 100644 --- a/packages/presence/src/routes.ts +++ b/packages/presence/src/routes.ts @@ -29,6 +29,7 @@ import { PresenceHeartbeatBody, PresenceJoinBody, } from "./schema"; +import { bindPresenceStream } from "./sse-stream"; const DEFAULT_HEARTBEAT_TIMEOUT_MS = 45_000; @@ -297,34 +298,13 @@ export function createPresenceRoutes( const key = { tenantId: tenant.id, surface }; return streamSSE(c, async (stream) => { - let unsubscribePresence: () => void = () => undefined; - let unsubscribeDoc: () => void = () => undefined; - let unsubscribeSnapshots: () => void = () => undefined; - const teardown = () => { - unsubscribePresence(); - unsubscribeDoc(); - unsubscribeSnapshots(); - }; - unsubscribePresence = registry.subscribe(key, (states) => { - stream - .writeSSE({ event: "presence.state", data: JSON.stringify(states) }) - .catch(teardown); - }); - unsubscribeDoc = registry.subscribeDocUpdates(key, (update) => { - stream - .writeSSE({ - event: "doc.update", - data: JSON.stringify({ update: encodeBase64(update) }), - }) - .catch(teardown); - }); - unsubscribeSnapshots = registry.subscribeSnapshots(key, (info) => { - stream - .writeSSE({ event: "doc.saved", data: JSON.stringify(info) }) - .catch(teardown); + const { teardown, closed } = bindPresenceStream({ + stream, + registry, + key, }); stream.onAbort(teardown); - await new Promise(() => undefined); + await closed; }); }); diff --git a/packages/presence/src/sse-stream.test.ts b/packages/presence/src/sse-stream.test.ts new file mode 100644 index 000000000..24be9ec83 --- /dev/null +++ b/packages/presence/src/sse-stream.test.ts @@ -0,0 +1,186 @@ +import { afterEach, describe, expect, spyOn, test } from "bun:test"; +import * as errorSink from "@corbits/error-sink"; +import { bindPresenceStream } from "./sse-stream"; +import type { + PresenceDocSnapshotListener, + PresenceDocUpdateListener, + PresenceRoomKey, + PresenceRoomListener, +} from "./room-registry"; + +const KEY: PresenceRoomKey = { tenantId: "tnt_a", surface: "channel:chn_1" }; + +async function flush(times = 20): Promise { + for (let i = 0; i < times; i++) { + await Promise.resolve(); + } +} + +function fakeStream( + writeSSE: (message: unknown) => Promise, + close: () => Promise = () => Promise.resolve(), +) { + return { writeSSE, close } as unknown as Parameters< + typeof bindPresenceStream + >[0]["stream"]; +} + +function fakeRegistry() { + let unsubPresence = 0; + let unsubDoc = 0; + let unsubSnapshots = 0; + return { + subscribe(_key: PresenceRoomKey, listener: PresenceRoomListener) { + listener([]); + return () => { + unsubPresence += 1; + }; + }, + subscribeDocUpdates( + _key: PresenceRoomKey, + _listener: PresenceDocUpdateListener, + ) { + return () => { + unsubDoc += 1; + }; + }, + subscribeSnapshots( + _key: PresenceRoomKey, + _listener: PresenceDocSnapshotListener, + ) { + return () => { + unsubSnapshots += 1; + }; + }, + unsubCounts() { + return { unsubPresence, unsubDoc, unsubSnapshots }; + }, + }; +} + +describe("bindPresenceStream", () => { + afterEach(() => { + ( + errorSink.reportError as unknown as { mockRestore?: () => void } + ).mockRestore?.(); + }); + + test("forwards a presence snapshot onto the stream as an SSE write", async () => { + const writes: unknown[] = []; + const stream = fakeStream((message) => { + writes.push(message); + return Promise.resolve(); + }); + bindPresenceStream({ + stream, + registry: fakeRegistry(), + key: KEY, + }); + await flush(); + expect(writes).toHaveLength(1); + expect(writes[0]).toMatchObject({ event: "presence.state" }); + }); + + test("a stalled write aborts the writer instead of awaiting a hung close()", async () => { + let abortCount = 0; + let closeCount = 0; + const stream = Object.assign( + fakeStream( + () => new Promise(() => undefined), + () => { + closeCount += 1; + return new Promise(() => undefined); + }, + ), + { + writer: { + desiredSize: 1, + abort: () => { + abortCount += 1; + }, + }, + }, + ); + const report = spyOn(errorSink, "reportError").mockReturnValue("ref_test"); + const registry = fakeRegistry(); + + const { closed } = bindPresenceStream({ + stream, + registry, + key: KEY, + writeTimeoutMs: 20, + }); + + await Promise.race([ + closed, + new Promise((_, reject) => + setTimeout(() => reject(new Error("hung on close()")), 200), + ), + ]); + + expect(abortCount).toBe(1); + expect(closeCount).toBe(0); + expect(report).toHaveBeenCalledTimes(1); + expect(report.mock.calls[0]?.[0]).toEqual( + expect.objectContaining({ message: "presence SSE write stalled" }), + ); + expect(report.mock.calls[0]?.[1]).toMatchObject({ + operation: "presence.stream.write", + tenantId: "tnt_a", + extra: { surface: "channel:chn_1" }, + }); + expect(registry.unsubCounts()).toEqual({ + unsubPresence: 1, + unsubDoc: 1, + unsubSnapshots: 1, + }); + }); + + test("a client abort mid-write does not reportError after onAbort teardown", async () => { + let abortCount = 0; + const stream = Object.assign( + fakeStream(() => new Promise(() => undefined)), + { aborted: false }, + ); + Reflect.set(stream, "writer", { + desiredSize: 1, + abort: () => { + abortCount += 1; + }, + }); + const report = spyOn(errorSink, "reportError").mockReturnValue("ref_test"); + + const { teardown } = bindPresenceStream({ + stream, + registry: fakeRegistry(), + key: KEY, + writeTimeoutMs: 20, + }); + await Promise.resolve(); + + stream.aborted = true; + teardown(); + await new Promise((resolve) => setTimeout(resolve, 50)); + + expect(report).not.toHaveBeenCalled(); + expect(abortCount).toBe(1); + }); + + test("an errored writer after a swallowed write reports and tears down", async () => { + const stream = Object.assign( + fakeStream(() => Promise.resolve()), + { writer: { desiredSize: null } }, + ); + const report = spyOn(errorSink, "reportError").mockReturnValue("ref_test"); + const registry = fakeRegistry(); + + bindPresenceStream({ stream, registry, key: KEY }); + await flush(); + + expect(report).toHaveBeenCalledTimes(1); + expect(report.mock.calls[0]?.[0]).toEqual( + expect.objectContaining({ message: "presence SSE write failed" }), + ); + expect(registry.unsubCounts().unsubPresence).toBe(1); + }); +}); diff --git a/packages/presence/src/sse-stream.ts b/packages/presence/src/sse-stream.ts new file mode 100644 index 000000000..2554e9c7c --- /dev/null +++ b/packages/presence/src/sse-stream.ts @@ -0,0 +1,147 @@ +// SSE write observation for the presence room stream (CL-7246 / CL-7212). +// Hono's `StreamingApi.write` awaits `writer.write` inside a bare +// `catch {}`, so a failed or stalled sink never rejects `writeSSE`. +// Bound every write, inspect the writer after it "succeeds", and on +// stall abort the writer rather than awaiting `close()` — that queues +// behind the same in-flight write and never drops the socket. +import type { SSEStreamingApi } from "hono/streaming"; +import { reportError } from "@corbits/error-sink"; + +import { encodeBase64 } from "./base64"; +import type { PresenceRoomKey, PresenceRoomRegistry } from "./room-registry"; + +const DEFAULT_WRITE_TIMEOUT_MS = 10_000; + +function underlyingWriterIsErrored(stream: SSEStreamingApi): boolean { + const writer = Reflect.get(stream, "writer"); + if (typeof writer !== "object" || writer === null) return false; + if (!("desiredSize" in writer)) return false; + return writer.desiredSize === null; +} + +function abortUnderlyingWriter(stream: SSEStreamingApi): boolean { + const writer = Reflect.get(stream, "writer"); + if (typeof writer !== "object" || writer === null) return false; + const abort = Reflect.get(writer, "abort"); + if (typeof abort !== "function") return false; + try { + void Promise.resolve(abort.call(writer)).catch(() => undefined); + return true; + } catch { + // report-error-ignore: aborting an already-closed writer is not an incident + return true; + } +} + +function dropStream(stream: SSEStreamingApi): void { + const abortedWriter = abortUnderlyingWriter(stream); + if (typeof stream.abort === "function") { + stream.abort(); + return; + } + if (abortedWriter) return; + void stream.close().catch(() => undefined); +} + +async function writeSSEObservingFailure( + stream: SSEStreamingApi, + message: { event?: string; data: string }, + timeoutMs: number, +): Promise { + let timeoutId: ReturnType | undefined; + const write = stream.writeSSE(message).then(() => { + if (stream.aborted || stream.closed || underlyingWriterIsErrored(stream)) { + throw new Error("presence SSE write failed"); + } + }); + const stalled = new Promise((_, reject) => { + timeoutId = setTimeout(() => { + reject(new Error("presence SSE write stalled")); + }, timeoutMs); + }); + try { + await Promise.race([write, stalled]); + } finally { + if (timeoutId !== undefined) clearTimeout(timeoutId); + void write.catch(() => undefined); + } +} + +export interface PresenceStreamBridge { + teardown: () => void; + closed: Promise; +} + +export function bindPresenceStream(input: { + stream: SSEStreamingApi; + registry: Pick< + PresenceRoomRegistry, + "subscribe" | "subscribeDocUpdates" | "subscribeSnapshots" + >; + key: PresenceRoomKey; + writeTimeoutMs?: number; +}): PresenceStreamBridge { + let tornDown = false; + let resolveClosed: () => void = () => undefined; + const closed = new Promise((resolve) => { + resolveClosed = resolve; + }); + + let unsubscribePresence: () => void = () => undefined; + let unsubscribeDoc: () => void = () => undefined; + let unsubscribeSnapshots: () => void = () => undefined; + + const teardown = () => { + if (tornDown) return; + tornDown = true; + unsubscribePresence(); + unsubscribeDoc(); + unsubscribeSnapshots(); + resolveClosed(); + }; + + const writeTimeoutMs = input.writeTimeoutMs ?? DEFAULT_WRITE_TIMEOUT_MS; + + const writeOrTearDown = (message: { event?: string; data: string }) => { + if (tornDown) return; + void (async () => { + try { + await writeSSEObservingFailure(input.stream, message, writeTimeoutMs); + } catch (error) { + if (!(tornDown || input.stream.aborted)) { + reportError(error, { + operation: "presence.stream.write", + tenantId: input.key.tenantId, + extra: { surface: input.key.surface }, + }); + } + teardown(); + dropStream(input.stream); + } + })(); + }; + + unsubscribePresence = input.registry.subscribe(input.key, (states) => { + writeOrTearDown({ + event: "presence.state", + data: JSON.stringify(states), + }); + }); + unsubscribeDoc = input.registry.subscribeDocUpdates(input.key, (update) => { + writeOrTearDown({ + event: "doc.update", + data: JSON.stringify({ update: encodeBase64(update) }), + }); + }); + unsubscribeSnapshots = input.registry.subscribeSnapshots( + input.key, + (info) => { + writeOrTearDown({ + event: "doc.saved", + data: JSON.stringify(info), + }); + }, + ); + + return { teardown, closed }; +}