diff --git a/packages/ui/src/sync/__tests__/event-pipeline.test.js b/packages/ui/src/sync/__tests__/event-pipeline.test.js new file mode 100644 index 00000000..78fc50a0 --- /dev/null +++ b/packages/ui/src/sync/__tests__/event-pipeline.test.js @@ -0,0 +1,195 @@ +import { afterEach, describe, expect, it } from 'bun:test'; +import { createEventPipeline } from '../event-pipeline'; + +const originalDocument = globalThis.document; +const originalWindow = globalThis.window; + +function installDomStubs() { + globalThis.document = { + visibilityState: 'visible', + addEventListener() {}, + removeEventListener() {}, + }; + + globalThis.window = { + addEventListener() {}, + removeEventListener() {}, + }; +} + +afterEach(() => { + globalThis.document = originalDocument; + globalThis.window = originalWindow; +}); + +function createSdkWithSingleEvent(event, hold) { + return { + global: { + event: async () => ({ + stream: (async function* () { + yield event; + await hold; + })(), + }), + }, + }; +} + +describe('createEventPipeline', () => { + it('falls back to payload.properties.directory when the SDK event omits top-level directory', async () => { + installDomStubs(); + + let releaseStream; + const hold = new Promise((resolve) => { + releaseStream = resolve; + }); + + const received = []; + const sdk = createSdkWithSingleEvent({ + payload: { + type: 'session.status', + properties: { + directory: 'C:/Users/daveotero/localdev/openchamber', + sessionID: 'session-1', + status: { type: 'busy' }, + }, + }, + }, hold); + + const delivered = new Promise((resolve) => { + const { cleanup } = createEventPipeline({ + sdk, + onEvent: (directory, payload) => { + received.push({ directory, payload }); + cleanup(); + releaseStream(); + resolve(); + }, + }); + }); + + await delivered; + + expect(received).toHaveLength(1); + expect(received[0].directory).toBe('C:/Users/daveotero/localdev/openchamber'); + expect(received[0].payload.type).toBe('session.status'); + }); + + it('prefers the explicit top-level event directory when present', async () => { + installDomStubs(); + + let releaseStream; + const hold = new Promise((resolve) => { + releaseStream = resolve; + }); + + const received = []; + const sdk = createSdkWithSingleEvent({ + directory: 'C:/top-level', + payload: { + type: 'session.status', + properties: { + directory: 'C:/nested', + sessionID: 'session-2', + status: { type: 'busy' }, + }, + }, + }, hold); + + const delivered = new Promise((resolve) => { + const { cleanup } = createEventPipeline({ + sdk, + onEvent: (directory, payload) => { + received.push({ directory, payload }); + cleanup(); + releaseStream(); + resolve(); + }, + }); + }); + + await delivered; + + expect(received).toHaveLength(1); + expect(received[0].directory).toBe('C:/top-level'); + expect(received[0].payload.type).toBe('session.status'); + }); + + it('uses payload.properties.directory when the top-level directory is an empty string', async () => { + installDomStubs(); + + let releaseStream; + const hold = new Promise((resolve) => { + releaseStream = resolve; + }); + + const received = []; + const sdk = createSdkWithSingleEvent({ + directory: '', + payload: { + type: 'message.part.updated', + properties: { + directory: 'C:/fallback-dir', + part: { + id: 'part-1', + type: 'text', + messageID: 'message-1', + }, + }, + }, + }, hold); + + const delivered = new Promise((resolve) => { + const { cleanup } = createEventPipeline({ + sdk, + onEvent: (directory, payload) => { + received.push({ directory, payload }); + cleanup(); + releaseStream(); + resolve(); + }, + }); + }); + + await delivered; + + expect(received).toHaveLength(1); + expect(received[0].directory).toBe('C:/fallback-dir'); + expect(received[0].payload.type).toBe('message.part.updated'); + }); + + it('keeps truly global events on the global channel when no directory is present anywhere', async () => { + installDomStubs(); + + let releaseStream; + const hold = new Promise((resolve) => { + releaseStream = resolve; + }); + + const received = []; + const sdk = createSdkWithSingleEvent({ + payload: { + type: 'server.connected', + properties: {}, + }, + }, hold); + + const delivered = new Promise((resolve) => { + const { cleanup } = createEventPipeline({ + sdk, + onEvent: (directory, payload) => { + received.push({ directory, payload }); + cleanup(); + releaseStream(); + resolve(); + }, + }); + }); + + await delivered; + + expect(received).toHaveLength(1); + expect(received[0].directory).toBe('global'); + expect(received[0].payload.type).toBe('server.connected'); + }); +}); diff --git a/packages/ui/src/sync/event-pipeline.ts b/packages/ui/src/sync/event-pipeline.ts index 6485863d..977dee0c 100644 --- a/packages/ui/src/sync/event-pipeline.ts +++ b/packages/ui/src/sync/event-pipeline.ts @@ -41,6 +41,25 @@ export type EventPipelineInput = { onReconnect?: () => void } +function resolveEventDirectory(event: unknown, payload: Event): string { + const directDirectory = + typeof event === "object" && event !== null && typeof (event as { directory?: unknown }).directory === "string" + ? (event as { directory: string }).directory + : null + + if (directDirectory && directDirectory.length > 0) { + return directDirectory + } + + const properties = + typeof payload.properties === "object" && payload.properties !== null + ? (payload.properties as Record) + : null + const propertyDirectory = typeof properties?.directory === "string" ? properties.directory : null + + return propertyDirectory && propertyDirectory.length > 0 ? propertyDirectory : "global" +} + export function createEventPipeline(input: EventPipelineInput) { const { sdk, onEvent, onReconnect } = input const abort = new AbortController() @@ -167,11 +186,11 @@ export function createEventPipeline(input: EventPipelineInput) { for await (const event of events.stream) { resetHeartbeat() streamErrorLogged = false - const directory = (event as { directory?: string }).directory ?? "global" const payload = (event as { payload?: Event }).payload ?? (event as unknown as Event) if (!payload || typeof payload !== "object" || typeof (payload as { type?: unknown }).type !== "string") { continue } + const directory = resolveEventDirectory(event, payload) const k = key(directory, payload) if (k) { const i = coalesced.get(k)