diff options
| author | Adam Malczewski <[email protected]> | 2026-06-27 01:09:39 +0900 |
|---|---|---|
| committer | Adam Malczewski <[email protected]> | 2026-06-27 01:09:39 +0900 |
| commit | 61e45e60d699ed1ca46f94a8f181c92a940317c6 (patch) | |
| tree | 2892d9773c5a8e367e1e58cdb1e88d9c6ad3fe6d /packages/message-queue | |
| parent | 63c7e64532e85e0bbdd6d9ac6825d8f86be98e7a (diff) | |
| parent | 727c98c9dae516a2070eb950410314380a20c974 (diff) | |
| download | dispatch-61e45e60d699ed1ca46f94a8f181c92a940317c6.tar.gz dispatch-61e45e60d699ed1ca46f94a8f181c92a940317c6.zip | |
Merge branch 'feature/indent-change' into dev
Diffstat (limited to 'packages/message-queue')
| -rw-r--r-- | packages/message-queue/package.json | 24 | ||||
| -rw-r--r-- | packages/message-queue/src/extension.ts | 106 | ||||
| -rw-r--r-- | packages/message-queue/src/index.ts | 26 | ||||
| -rw-r--r-- | packages/message-queue/src/pure.test.ts | 238 | ||||
| -rw-r--r-- | packages/message-queue/src/pure.ts | 74 | ||||
| -rw-r--r-- | packages/message-queue/src/service.test.ts | 148 | ||||
| -rw-r--r-- | packages/message-queue/src/service.ts | 94 | ||||
| -rw-r--r-- | packages/message-queue/tsconfig.json | 18 |
8 files changed, 364 insertions, 364 deletions
diff --git a/packages/message-queue/package.json b/packages/message-queue/package.json index a8fcc4b..75df09e 100644 --- a/packages/message-queue/package.json +++ b/packages/message-queue/package.json @@ -1,14 +1,14 @@ { - "name": "@dispatch/message-queue", - "version": "0.0.0", - "type": "module", - "private": true, - "main": "dist/index.js", - "types": "dist/index.d.ts", - "dependencies": { - "@dispatch/kernel": "workspace:*", - "@dispatch/surface-registry": "workspace:*", - "@dispatch/ui-contract": "workspace:*", - "@dispatch/wire": "workspace:*" - } + "name": "@dispatch/message-queue", + "version": "0.0.0", + "type": "module", + "private": true, + "main": "dist/index.js", + "types": "dist/index.d.ts", + "dependencies": { + "@dispatch/kernel": "workspace:*", + "@dispatch/surface-registry": "workspace:*", + "@dispatch/ui-contract": "workspace:*", + "@dispatch/wire": "workspace:*" + } } diff --git a/packages/message-queue/src/extension.ts b/packages/message-queue/src/extension.ts index 26d19ca..29979be 100644 --- a/packages/message-queue/src/extension.ts +++ b/packages/message-queue/src/extension.ts @@ -12,71 +12,71 @@ import { buildQueueSpec, MESSAGE_QUEUE_SURFACE_ID } from "./pure.js"; import { createMessageQueueService, messageQueueHandle } from "./service.js"; export const manifest: Manifest = { - id: "message-queue", - name: "Message Queue", - version: "0.0.0", - apiVersion: "^0.1.0", - trust: "bundled", - activation: "eager", - dependsOn: ["surface-registry"], - capabilities: {}, - contributes: { - services: ["message-queue"], - }, + id: "message-queue", + name: "Message Queue", + version: "0.0.0", + apiVersion: "^0.1.0", + trust: "bundled", + activation: "eager", + dependsOn: ["surface-registry"], + capabilities: {}, + contributes: { + services: ["message-queue"], + }, }; export function activate(host: HostAPI): void { - const registry = host.getService(surfaceRegistryHandle); + const registry = host.getService(surfaceRegistryHandle); - const subscribers = new Set<() => void>(); + const subscribers = new Set<() => void>(); - const service = createMessageQueueService({ - id: () => crypto.randomUUID(), - now: () => Date.now(), - notify: () => { - for (const sub of subscribers) { - sub(); - } - }, - logger: host.logger, - }); + const service = createMessageQueueService({ + id: () => crypto.randomUUID(), + now: () => Date.now(), + notify: () => { + for (const sub of subscribers) { + sub(); + } + }, + logger: host.logger, + }); - host.provideService(messageQueueHandle, service); + host.provideService(messageQueueHandle, service); - function getSpec(context?: SurfaceContext): SurfaceSpec { - const convId = context?.conversationId; - const messages = convId === undefined ? [] : service.getQueue(convId); - return buildQueueSpec(messages); - } + function getSpec(context?: SurfaceContext): SurfaceSpec { + const convId = context?.conversationId; + const messages = convId === undefined ? [] : service.getQueue(convId); + return buildQueueSpec(messages); + } - function invoke(_actionId: string, _payload?: unknown, _context?: SurfaceContext): void { - // The message-queue surface is read-only: a client renders the queue and - // the session-orchestrator drains it. No client-facing actions. - } + function invoke(_actionId: string, _payload?: unknown, _context?: SurfaceContext): void { + // The message-queue surface is read-only: a client renders the queue and + // the session-orchestrator drains it. No client-facing actions. + } - const provider: SurfaceProvider = { - catalogEntry: { - id: MESSAGE_QUEUE_SURFACE_ID, - region: "side", - title: "Message Queue", - scope: "conversation", - }, - getSpec, - invoke, - subscribe(onChange) { - subscribers.add(onChange); - return () => { - subscribers.delete(onChange); - }; - }, - }; + const provider: SurfaceProvider = { + catalogEntry: { + id: MESSAGE_QUEUE_SURFACE_ID, + region: "side", + title: "Message Queue", + scope: "conversation", + }, + getSpec, + invoke, + subscribe(onChange) { + subscribers.add(onChange); + return () => { + subscribers.delete(onChange); + }; + }, + }; - registry.register(provider); + registry.register(provider); - host.logger.info("message-queue: registered"); + host.logger.info("message-queue: registered"); } export const extension: Extension = { - manifest, - activate, + manifest, + activate, }; diff --git a/packages/message-queue/src/index.ts b/packages/message-queue/src/index.ts index 79bbb25..11467e1 100644 --- a/packages/message-queue/src/index.ts +++ b/packages/message-queue/src/index.ts @@ -9,19 +9,19 @@ export type { QueuedMessage, QueuePayload } from "@dispatch/wire"; export { extension, manifest } from "./extension.js"; export { - buildQueueSpec, - combine, - drain, - enqueue, - getQueue, - MESSAGE_QUEUE_RENDERER_ID, - MESSAGE_QUEUE_SURFACE_ID, - type MessageQueueState, - type QueueDeps, + buildQueueSpec, + combine, + drain, + enqueue, + getQueue, + MESSAGE_QUEUE_RENDERER_ID, + MESSAGE_QUEUE_SURFACE_ID, + type MessageQueueState, + type QueueDeps, } from "./pure.js"; export { - createMessageQueueService, - type MessageQueueDeps, - type MessageQueueService, - messageQueueHandle, + createMessageQueueService, + type MessageQueueDeps, + type MessageQueueService, + messageQueueHandle, } from "./service.js"; diff --git a/packages/message-queue/src/pure.test.ts b/packages/message-queue/src/pure.test.ts index ff1a08f..3fd6039 100644 --- a/packages/message-queue/src/pure.test.ts +++ b/packages/message-queue/src/pure.test.ts @@ -1,144 +1,144 @@ import type { QueuedMessage } from "@dispatch/wire"; import { describe, expect, it } from "vitest"; import { - buildQueueSpec, - combine, - drain, - enqueue, - getQueue, - MESSAGE_QUEUE_RENDERER_ID, - MESSAGE_QUEUE_SURFACE_ID, - type MessageQueueState, - type QueueDeps, + buildQueueSpec, + combine, + drain, + enqueue, + getQueue, + MESSAGE_QUEUE_RENDERER_ID, + MESSAGE_QUEUE_SURFACE_ID, + type MessageQueueState, + type QueueDeps, } from "./pure.js"; function makeDeps(): QueueDeps { - let idCounter = 0; - let now = 1_000; - return { - id: () => `id-${++idCounter}`, - now: () => (now += 10), - }; + let idCounter = 0; + let now = 1_000; + return { + id: () => `id-${++idCounter}`, + now: () => (now += 10), + }; } describe("enqueue", () => { - it("enqueue appends + returns snapshot with the new message (id unique, queuedAt set)", () => { - const state: MessageQueueState = new Map(); - const deps = makeDeps(); - - const first = enqueue(state, "c1", "first", deps); - expect(first).toHaveLength(1); - const firstMsg = first[0]; - if (firstMsg === undefined) throw new Error("expected a message"); - expect(firstMsg.id).toBe("id-1"); - expect(firstMsg.text).toBe("first"); - expect(firstMsg.queuedAt).toBe(1_010); - - const second = enqueue(state, "c1", "second", deps); - expect(second).toHaveLength(2); - const secondMsg = second[1]; - if (secondMsg === undefined) throw new Error("expected a message"); - expect(secondMsg.id).toBe("id-2"); // unique per message - expect(secondMsg.text).toBe("second"); - expect(secondMsg.queuedAt).toBe(1_020); - - // a separate conversation does not share state - const other = enqueue(state, "c2", "other", deps); - expect(other).toHaveLength(1); - expect(getQueue(state, "c2")).toHaveLength(1); - }); + it("enqueue appends + returns snapshot with the new message (id unique, queuedAt set)", () => { + const state: MessageQueueState = new Map(); + const deps = makeDeps(); + + const first = enqueue(state, "c1", "first", deps); + expect(first).toHaveLength(1); + const firstMsg = first[0]; + if (firstMsg === undefined) throw new Error("expected a message"); + expect(firstMsg.id).toBe("id-1"); + expect(firstMsg.text).toBe("first"); + expect(firstMsg.queuedAt).toBe(1_010); + + const second = enqueue(state, "c1", "second", deps); + expect(second).toHaveLength(2); + const secondMsg = second[1]; + if (secondMsg === undefined) throw new Error("expected a message"); + expect(secondMsg.id).toBe("id-2"); // unique per message + expect(secondMsg.text).toBe("second"); + expect(secondMsg.queuedAt).toBe(1_020); + + // a separate conversation does not share state + const other = enqueue(state, "c2", "other", deps); + expect(other).toHaveLength(1); + expect(getQueue(state, "c2")).toHaveLength(1); + }); }); describe("getQueue", () => { - it("getQueue returns current snapshot; empty array when none", () => { - const state: MessageQueueState = new Map(); - const deps = makeDeps(); + it("getQueue returns current snapshot; empty array when none", () => { + const state: MessageQueueState = new Map(); + const deps = makeDeps(); - expect(getQueue(state, "unknown")).toEqual([]); + expect(getQueue(state, "unknown")).toEqual([]); - enqueue(state, "c1", "x", deps); - enqueue(state, "c1", "y", deps); - const snap = getQueue(state, "c1"); - expect(snap.map((m) => m.text)).toEqual(["x", "y"]); + enqueue(state, "c1", "x", deps); + enqueue(state, "c1", "y", deps); + const snap = getQueue(state, "c1"); + expect(snap.map((m) => m.text)).toEqual(["x", "y"]); - // returns a COPY — mutating it does not affect live state - snap.push({ id: "evil", text: "mutate", queuedAt: 0 }); - expect(getQueue(state, "c1")).toHaveLength(2); - }); + // returns a COPY — mutating it does not affect live state + snap.push({ id: "evil", text: "mutate", queuedAt: 0 }); + expect(getQueue(state, "c1")).toHaveLength(2); + }); }); describe("drain", () => { - it("drain returns all + clears; second drain returns []", () => { - const state: MessageQueueState = new Map(); - const deps = makeDeps(); - enqueue(state, "c1", "a", deps); - enqueue(state, "c1", "b", deps); - - const drained = drain(state, "c1"); - expect(drained.map((m) => m.text)).toEqual(["a", "b"]); - - // cleared - expect(getQueue(state, "c1")).toEqual([]); - expect(drain(state, "c1")).toEqual([]); - }); - - it("drain on an empty queue returns [] and is a no-op", () => { - const state: MessageQueueState = new Map(); - - // never had a queue - expect(drain(state, "c1")).toEqual([]); - expect(getQueue(state, "c1")).toEqual([]); - - // and after a drain that emptied it, draining again is a no-op - const deps = makeDeps(); - enqueue(state, "c1", "once", deps); - drain(state, "c1"); - expect(drain(state, "c1")).toEqual([]); - expect(getQueue(state, "c1")).toEqual([]); - }); + it("drain returns all + clears; second drain returns []", () => { + const state: MessageQueueState = new Map(); + const deps = makeDeps(); + enqueue(state, "c1", "a", deps); + enqueue(state, "c1", "b", deps); + + const drained = drain(state, "c1"); + expect(drained.map((m) => m.text)).toEqual(["a", "b"]); + + // cleared + expect(getQueue(state, "c1")).toEqual([]); + expect(drain(state, "c1")).toEqual([]); + }); + + it("drain on an empty queue returns [] and is a no-op", () => { + const state: MessageQueueState = new Map(); + + // never had a queue + expect(drain(state, "c1")).toEqual([]); + expect(getQueue(state, "c1")).toEqual([]); + + // and after a drain that emptied it, draining again is a no-op + const deps = makeDeps(); + enqueue(state, "c1", "once", deps); + drain(state, "c1"); + expect(drain(state, "c1")).toEqual([]); + expect(getQueue(state, "c1")).toEqual([]); + }); }); describe("combine", () => { - it("combine joins texts with blank-line separator", () => { - const msgs: QueuedMessage[] = [ - { id: "1", text: "alpha", queuedAt: 1 }, - { id: "2", text: "beta", queuedAt: 2 }, - ]; - expect(combine(msgs)).toBe("alpha\n\nbeta"); - expect(combine([])).toBe(""); - expect(combine([{ id: "1", text: "solo", queuedAt: 1 }])).toBe("solo"); - }); + it("combine joins texts with blank-line separator", () => { + const msgs: QueuedMessage[] = [ + { id: "1", text: "alpha", queuedAt: 1 }, + { id: "2", text: "beta", queuedAt: 2 }, + ]; + expect(combine(msgs)).toBe("alpha\n\nbeta"); + expect(combine([])).toBe(""); + expect(combine([{ id: "1", text: "solo", queuedAt: 1 }])).toBe("solo"); + }); }); describe("buildQueueSpec", () => { - it("renders a custom field with the queue snapshot", () => { - const msgs: QueuedMessage[] = [ - { id: "1", text: "a", queuedAt: 1 }, - { id: "2", text: "b", queuedAt: 2 }, - ]; - const spec = buildQueueSpec(msgs); - expect(spec.id).toBe(MESSAGE_QUEUE_SURFACE_ID); - expect(spec.region).toBe("side"); - expect(spec.title).toBe("Message Queue"); - expect(spec.fields).toHaveLength(1); - - const field = spec.fields[0]; - if (field === undefined || field.kind !== "custom") { - throw new Error("expected a custom field"); - } - expect(field.rendererId).toBe(MESSAGE_QUEUE_RENDERER_ID); - const payload = field.payload as { messages: readonly QueuedMessage[] }; - expect(payload.messages).toHaveLength(2); - expect(payload.messages.map((m) => m.text)).toEqual(["a", "b"]); - }); - - it("renders an empty list for an empty queue", () => { - const spec = buildQueueSpec([]); - const field = spec.fields[0]; - if (field === undefined || field.kind !== "custom") { - throw new Error("expected a custom field"); - } - const payload = field.payload as { messages: readonly QueuedMessage[] }; - expect(payload.messages).toEqual([]); - }); + it("renders a custom field with the queue snapshot", () => { + const msgs: QueuedMessage[] = [ + { id: "1", text: "a", queuedAt: 1 }, + { id: "2", text: "b", queuedAt: 2 }, + ]; + const spec = buildQueueSpec(msgs); + expect(spec.id).toBe(MESSAGE_QUEUE_SURFACE_ID); + expect(spec.region).toBe("side"); + expect(spec.title).toBe("Message Queue"); + expect(spec.fields).toHaveLength(1); + + const field = spec.fields[0]; + if (field === undefined || field.kind !== "custom") { + throw new Error("expected a custom field"); + } + expect(field.rendererId).toBe(MESSAGE_QUEUE_RENDERER_ID); + const payload = field.payload as { messages: readonly QueuedMessage[] }; + expect(payload.messages).toHaveLength(2); + expect(payload.messages.map((m) => m.text)).toEqual(["a", "b"]); + }); + + it("renders an empty list for an empty queue", () => { + const spec = buildQueueSpec([]); + const field = spec.fields[0]; + if (field === undefined || field.kind !== "custom") { + throw new Error("expected a custom field"); + } + const payload = field.payload as { messages: readonly QueuedMessage[] }; + expect(payload.messages).toEqual([]); + }); }); diff --git a/packages/message-queue/src/pure.ts b/packages/message-queue/src/pure.ts index 834018b..981e005 100644 --- a/packages/message-queue/src/pure.ts +++ b/packages/message-queue/src/pure.ts @@ -17,10 +17,10 @@ export type MessageQueueState = Map<string, QueuedMessage[]>; /** Injected effectful factories kept out of the pure core. */ export interface QueueDeps { - /** Stable (client-visible) id factory for UI keying + dedup. */ - readonly id: () => string; - /** Clock returning epoch-ms for `queuedAt`. */ - readonly now: () => number; + /** Stable (client-visible) id factory for UI keying + dedup. */ + readonly id: () => string; + /** Clock returning epoch-ms for `queuedAt`. */ + readonly now: () => number; } /** Surface id this extension contributes (also the manifest + catalog id). */ @@ -34,19 +34,19 @@ export const MESSAGE_QUEUE_RENDERER_ID = "message-queue"; * live state through the returned value. */ export function enqueue( - state: MessageQueueState, - conversationId: string, - text: string, - deps: QueueDeps, + state: MessageQueueState, + conversationId: string, + text: string, + deps: QueueDeps, ): QueuedMessage[] { - const message: QueuedMessage = { id: deps.id(), text, queuedAt: deps.now() }; - const existing = state.get(conversationId); - if (existing === undefined) { - state.set(conversationId, [message]); - } else { - existing.push(message); - } - return getQueue(state, conversationId); + const message: QueuedMessage = { id: deps.id(), text, queuedAt: deps.now() }; + const existing = state.get(conversationId); + if (existing === undefined) { + state.set(conversationId, [message]); + } else { + existing.push(message); + } + return getQueue(state, conversationId); } /** @@ -54,9 +54,9 @@ export function enqueue( * if the conversation has no queue / is unknown. */ export function getQueue(state: MessageQueueState, conversationId: string): QueuedMessage[] { - const existing = state.get(conversationId); - if (existing === undefined) return []; - return [...existing]; + const existing = state.get(conversationId); + if (existing === undefined) return []; + return [...existing]; } /** @@ -67,11 +67,11 @@ export function getQueue(state: MessageQueueState, conversationId: string): Queu * ChatMessage. */ export function drain(state: MessageQueueState, conversationId: string): QueuedMessage[] { - const existing = state.get(conversationId); - if (existing === undefined || existing.length === 0) return []; - const drained = [...existing]; - state.delete(conversationId); - return drained; + const existing = state.get(conversationId); + if (existing === undefined || existing.length === 0) return []; + const drained = [...existing]; + state.delete(conversationId); + return drained; } /** @@ -80,7 +80,7 @@ export function drain(state: MessageQueueState, conversationId: string): QueuedM * ChatMessage from this. */ export function combine(messages: readonly QueuedMessage[]): string { - return messages.map((m) => m.text).join("\n\n"); + return messages.map((m) => m.text).join("\n\n"); } /** @@ -90,16 +90,16 @@ export function combine(messages: readonly QueuedMessage[]): string { * I/O; the surface-registry re-fetches this on every notify. */ export function buildQueueSpec(messages: readonly QueuedMessage[]): SurfaceSpec { - const payload: QueuePayload = { messages }; - const field: CustomField = { - kind: "custom", - rendererId: MESSAGE_QUEUE_RENDERER_ID, - payload, - }; - return { - id: MESSAGE_QUEUE_SURFACE_ID, - region: "side", - title: "Message Queue", - fields: [field], - }; + const payload: QueuePayload = { messages }; + const field: CustomField = { + kind: "custom", + rendererId: MESSAGE_QUEUE_RENDERER_ID, + payload, + }; + return { + id: MESSAGE_QUEUE_SURFACE_ID, + region: "side", + title: "Message Queue", + fields: [field], + }; } diff --git a/packages/message-queue/src/service.test.ts b/packages/message-queue/src/service.test.ts index 04cde91..aa59dd3 100644 --- a/packages/message-queue/src/service.test.ts +++ b/packages/message-queue/src/service.test.ts @@ -5,97 +5,97 @@ import { buildQueueSpec, combine } from "./pure.js"; import { createMessageQueueService, type MessageQueueDeps } from "./service.js"; interface Recorder extends MessageQueueDeps { - readonly calls: { value: number }; + readonly calls: { value: number }; } function makeDeps(): Recorder { - let idCounter = 0; - let now = 1_000; - const calls = { value: 0 }; - return { - id: () => `q-${++idCounter}`, - now: () => (now += 10), - notify: () => { - calls.value += 1; - }, - calls, - }; + let idCounter = 0; + let now = 1_000; + const calls = { value: 0 }; + return { + id: () => `q-${++idCounter}`, + now: () => (now += 10), + notify: () => { + calls.value += 1; + }, + calls, + }; } function queueMessages(spec: SurfaceSpec): readonly QueuedMessage[] { - const field = spec.fields[0]; - if (field === undefined || field.kind !== "custom") { - throw new Error("expected a custom field"); - } - return (field.payload as QueuePayload).messages; + const field = spec.fields[0]; + if (field === undefined || field.kind !== "custom") { + throw new Error("expected a custom field"); + } + return (field.payload as QueuePayload).messages; } describe("message-queue service", () => { - it("enqueue pushes a surface update (snapshot grew)", () => { - const deps = makeDeps(); - const svc = createMessageQueueService(deps); - expect(deps.calls.value).toBe(0); + it("enqueue pushes a surface update (snapshot grew)", () => { + const deps = makeDeps(); + const svc = createMessageQueueService(deps); + expect(deps.calls.value).toBe(0); - const snapshot = svc.enqueue("c1", "hello"); - expect(deps.calls.value).toBe(1); // surface update pushed - expect(snapshot).toHaveLength(1); - const first = snapshot[0]; - if (first === undefined) throw new Error("expected a message"); - expect(first.text).toBe("hello"); - expect(typeof first.id).toBe("string"); - expect(first.queuedAt).toBe(1_010); + const snapshot = svc.enqueue("c1", "hello"); + expect(deps.calls.value).toBe(1); // surface update pushed + expect(snapshot).toHaveLength(1); + const first = snapshot[0]; + if (first === undefined) throw new Error("expected a message"); + expect(first.text).toBe("hello"); + expect(typeof first.id).toBe("string"); + expect(first.queuedAt).toBe(1_010); - // the surface spec reflects the grown snapshot - expect(queueMessages(buildQueueSpec(svc.getQueue("c1")))).toHaveLength(1); - }); + // the surface spec reflects the grown snapshot + expect(queueMessages(buildQueueSpec(svc.getQueue("c1")))).toHaveLength(1); + }); - it("drain pushes a surface update (empty)", () => { - const deps = makeDeps(); - const svc = createMessageQueueService(deps); - svc.enqueue("c1", "a"); - svc.enqueue("c1", "b"); - expect(deps.calls.value).toBe(2); // two enqueues notified + it("drain pushes a surface update (empty)", () => { + const deps = makeDeps(); + const svc = createMessageQueueService(deps); + svc.enqueue("c1", "a"); + svc.enqueue("c1", "b"); + expect(deps.calls.value).toBe(2); // two enqueues notified - const drained = svc.drain("c1"); - expect(deps.calls.value).toBe(3); // drain pushed a surface update - expect(drained).toHaveLength(2); - expect(drained.map((m) => m.text)).toEqual(["a", "b"]); + const drained = svc.drain("c1"); + expect(deps.calls.value).toBe(3); // drain pushed a surface update + expect(drained).toHaveLength(2); + expect(drained.map((m) => m.text)).toEqual(["a", "b"]); - // cleared + surface now shows empty - expect(svc.getQueue("c1")).toEqual([]); - expect(queueMessages(buildQueueSpec(svc.getQueue("c1")))).toEqual([]); - }); + // cleared + surface now shows empty + expect(svc.getQueue("c1")).toEqual([]); + expect(queueMessages(buildQueueSpec(svc.getQueue("c1")))).toEqual([]); + }); - it("drain on empty does NOT push a surface update (already empty)", () => { - const deps = makeDeps(); - const svc = createMessageQueueService(deps); + it("drain on empty does NOT push a surface update (already empty)", () => { + const deps = makeDeps(); + const svc = createMessageQueueService(deps); - expect(svc.drain("c1")).toEqual([]); - expect(deps.calls.value).toBe(0); // no notify — no change + expect(svc.drain("c1")).toEqual([]); + expect(deps.calls.value).toBe(0); // no notify — no change - // after a real drain empties it, a further drain is a no-op (no notify) - svc.enqueue("c1", "x"); - expect(deps.calls.value).toBe(1); - svc.drain("c1"); - expect(deps.calls.value).toBe(2); - svc.drain("c1"); - expect(deps.calls.value).toBe(2); // unchanged — queue was already empty - }); + // after a real drain empties it, a further drain is a no-op (no notify) + svc.enqueue("c1", "x"); + expect(deps.calls.value).toBe(1); + svc.drain("c1"); + expect(deps.calls.value).toBe(2); + svc.drain("c1"); + expect(deps.calls.value).toBe(2); // unchanged — queue was already empty + }); - it("getQueue returns current snapshot; empty for unknown conversation", () => { - const deps = makeDeps(); - const svc = createMessageQueueService(deps); - expect(svc.getQueue("nope")).toEqual([]); - svc.enqueue("c1", "hi"); - expect(svc.getQueue("c1")).toHaveLength(1); - }); + it("getQueue returns current snapshot; empty for unknown conversation", () => { + const deps = makeDeps(); + const svc = createMessageQueueService(deps); + expect(svc.getQueue("nope")).toEqual([]); + svc.enqueue("c1", "hi"); + expect(svc.getQueue("c1")).toHaveLength(1); + }); - it("drain returns the raw QueuedMessage[] the orchestrator combines", () => { - const deps = makeDeps(); - const svc = createMessageQueueService(deps); - svc.enqueue("c1", "alpha"); - svc.enqueue("c1", "beta"); - const drained = svc.drain("c1"); - expect(combine(drained)).toBe("alpha\n\nbeta"); - }); + it("drain returns the raw QueuedMessage[] the orchestrator combines", () => { + const deps = makeDeps(); + const svc = createMessageQueueService(deps); + svc.enqueue("c1", "alpha"); + svc.enqueue("c1", "beta"); + const drained = svc.drain("c1"); + expect(combine(drained)).toBe("alpha\n\nbeta"); + }); }); diff --git a/packages/message-queue/src/service.ts b/packages/message-queue/src/service.ts index 91d0fc5..97e270d 100644 --- a/packages/message-queue/src/service.ts +++ b/packages/message-queue/src/service.ts @@ -18,17 +18,17 @@ import { drain as drainQueue, enqueue as enqueueMessage, getQueue as readQueue } * `host.getService(messageQueueHandle)`. */ export interface MessageQueueService { - /** Append a message; return the CURRENT queue snapshot (post-append). */ - enqueue(conversationId: string, text: string): QueuedMessage[]; - /** Current queue snapshot (empty array if none / unknown conversation). */ - getQueue(conversationId: string): QueuedMessage[]; - /** - * Drain: return all queued messages, CLEAR the queue, and push a surface - * update (empty). Returns the raw `QueuedMessage[]` (NOT a ChatMessage — - * the caller combines + builds the ChatMessage). Empty array if the queue - * was empty (and then NO surface update is pushed — no change). - */ - drain(conversationId: string): QueuedMessage[]; + /** Append a message; return the CURRENT queue snapshot (post-append). */ + enqueue(conversationId: string, text: string): QueuedMessage[]; + /** Current queue snapshot (empty array if none / unknown conversation). */ + getQueue(conversationId: string): QueuedMessage[]; + /** + * Drain: return all queued messages, CLEAR the queue, and push a surface + * update (empty). Returns the raw `QueuedMessage[]` (NOT a ChatMessage — + * the caller combines + builds the ChatMessage). Empty array if the queue + * was empty (and then NO surface update is pushed — no change). + */ + drain(conversationId: string): QueuedMessage[]; } /** @@ -36,19 +36,19 @@ export interface MessageQueueService { * session-orchestrator imports to reach the queue — no string-keyed lookup. */ export const messageQueueHandle: ServiceHandle<MessageQueueService> = - defineService<MessageQueueService>("message-queue"); + defineService<MessageQueueService>("message-queue"); /** Deps for the service shell — the pure core's deps plus the surface notifier. */ export interface MessageQueueDeps extends QueueDeps { - /** - * Notify the surface-registry that the queue changed, so the transport - * re-fetches + pushes a full new SurfaceUpdate. Called ONLY on a real - * change (enqueue → grew; drain-non-empty → emptied); a no-op drain of an - * already-empty queue does NOT notify (no change → no surface push). - */ - readonly notify: () => void; - /** Injected host logger (optional in tests). Logs counts/lengths at debug, never bodies. */ - readonly logger?: Logger; + /** + * Notify the surface-registry that the queue changed, so the transport + * re-fetches + pushes a full new SurfaceUpdate. Called ONLY on a real + * change (enqueue → grew; drain-non-empty → emptied); a no-op drain of an + * already-empty queue does NOT notify (no change → no surface push). + */ + readonly notify: () => void; + /** Injected host logger (optional in tests). Logs counts/lengths at debug, never bodies. */ + readonly logger?: Logger; } /** @@ -58,31 +58,31 @@ export interface MessageQueueDeps extends QueueDeps { * in this closure, reachable only through the returned service. */ export function createMessageQueueService(deps: MessageQueueDeps): MessageQueueService { - const state: MessageQueueState = new Map(); + const state: MessageQueueState = new Map(); - return { - enqueue(conversationId, text) { - const snapshot = enqueueMessage(state, conversationId, text, deps); - deps.logger?.debug("message-queue: enqueued", { - conversationId, - queueLen: snapshot.length, - textLen: text.length, - }); - deps.notify(); - return snapshot; - }, - getQueue(conversationId) { - return readQueue(state, conversationId); - }, - drain(conversationId) { - const drained = drainQueue(state, conversationId); - if (drained.length === 0) return drained; - deps.logger?.debug("message-queue: drained", { - conversationId, - count: drained.length, - }); - deps.notify(); - return drained; - }, - }; + return { + enqueue(conversationId, text) { + const snapshot = enqueueMessage(state, conversationId, text, deps); + deps.logger?.debug("message-queue: enqueued", { + conversationId, + queueLen: snapshot.length, + textLen: text.length, + }); + deps.notify(); + return snapshot; + }, + getQueue(conversationId) { + return readQueue(state, conversationId); + }, + drain(conversationId) { + const drained = drainQueue(state, conversationId); + if (drained.length === 0) return drained; + deps.logger?.debug("message-queue: drained", { + conversationId, + count: drained.length, + }); + deps.notify(); + return drained; + }, + }; } diff --git a/packages/message-queue/tsconfig.json b/packages/message-queue/tsconfig.json index da5dfb7..1c0d44b 100644 --- a/packages/message-queue/tsconfig.json +++ b/packages/message-queue/tsconfig.json @@ -1,11 +1,11 @@ { - "extends": "../../tsconfig.base.json", - "compilerOptions": { "rootDir": "src", "outDir": "dist", "composite": true }, - "include": ["src/**/*.ts"], - "references": [ - { "path": "../kernel" }, - { "path": "../surface-registry" }, - { "path": "../ui-contract" }, - { "path": "../wire" } - ] + "extends": "../../tsconfig.base.json", + "compilerOptions": { "rootDir": "src", "outDir": "dist", "composite": true }, + "include": ["src/**/*.ts"], + "references": [ + { "path": "../kernel" }, + { "path": "../surface-registry" }, + { "path": "../ui-contract" }, + { "path": "../wire" } + ] } |
