1
10import { describe, expect, it } from "vitest";
11
12import {
13 createDomainEventSource,
14 DOMAIN_EVENTS_STREAM,
15 type DomainEventStreamClient,
16} from "../src/event-source.js";
17import type { DomainEvent } from "@factory/shared";
18
19
23function makeFakeClient(blockMs = 3) {
24 const calls: (string | number)[][] = [];
25 const queue: (Array<[string, Array<[string, string[]]>]> | null)[] = [];
26 const client: DomainEventStreamClient = {
27 async xread(...args: (string | number)[]) {
28 calls.push(args);
29 if (queue.length === 0) {
30 await new Promise((r) => setTimeout(r, blockMs));
31 return [];
32 }
33 return queue.shift() ?? [];
34 },
35 };
36 return {
37 client,
38 calls,
39 prime(result: Array<[string, Array<[string, string[]]>]> | null): void {
40 queue.push(result);
41 },
42 };
43}
44
45const sleep = (ms: number): Promise<void> => new Promise((r) => setTimeout(r, ms));
46const settle = (): Promise<void> => sleep(40);
47
48const entry = (id: string, event: string, payload: Record<string, unknown>): [string, string[]] => [
49 id,
50 [
51 "event",
52 event,
53 "payload",
54 JSON.stringify(payload),
55 "ts",
56 "2026-09-27T00:00:00.000Z",
57 "id",
58 `evt_${id}`,
59 ],
60];
61
62describe("createDomainEventSource (TASK-0094)", () => {
63 it("reads the factory:events stream (the API bus's stream)", async () => {
64 const fake = makeFakeClient();
65 const source = createDomainEventSource(() => {}, { client: fake.client });
66 source.start();
67 await settle();
68 await source.stop();
69 expect(fake.calls.length).toBeGreaterThanOrEqual(1);
70 const first = fake.calls[0]!;
71
72 expect(first[0]).toBe("COUNT");
73 expect(first[2]).toBe("BLOCK");
74 expect(first[4]).toBe("STREAMS");
75 expect(first[5]).toBe(DOMAIN_EVENTS_STREAM);
76 });
77
78 it("delivers EVERY event name (the server's sink routes; the source does not filter)", async () => {
79 const fake = makeFakeClient();
80 const seen: DomainEvent[] = [];
81 const source = createDomainEventSource((e) => seen.push(e), { client: fake.client });
82 source.start();
83 fake.prime([
84 [
85 DOMAIN_EVENTS_STREAM,
86 [
87 entry("1", "page.updated", { pageId: "pg_1" }),
88 entry("2", "comment.created", { commentId: "c1", pageId: "pg_1" }),
89 entry("3", "notification.created", { notificationId: "n1", userId: "u1" }),
90 entry("4", "record.created", { recordId: "r1" }),
91 ],
92 ],
93 ]);
94 await settle();
95 await source.stop();
96 expect(seen.map((e) => e.name)).toEqual([
97 "page.updated",
98 "comment.created",
99 "notification.created",
100 "record.created",
101 ]);
102 expect((seen[1]!.payload as { commentId: string }).commentId).toBe("c1");
103 });
104
105 it("keeps the parsed payload + ts + id of the envelope", async () => {
106 const fake = makeFakeClient();
107 const seen: DomainEvent[] = [];
108 const source = createDomainEventSource((e) => seen.push(e), { client: fake.client });
109 source.start();
110 fake.prime([
111 [
112 DOMAIN_EVENTS_STREAM,
113 [
114 entry("7", "comment.created", {
115 commentId: "c9",
116 pageId: "pg_9",
117 actorId: "u9",
118 body: "hi",
119 }),
120 ],
121 ],
122 ]);
123 await settle();
124 await source.stop();
125 expect(seen).toHaveLength(1);
126 expect(seen[0]!.payload).toEqual({
127 commentId: "c9",
128 pageId: "pg_9",
129 actorId: "u9",
130 body: "hi",
131 });
132 expect(seen[0]!.ts).toBe("2026-09-27T00:00:00.000Z");
133 expect(seen[0]!.id).toBe("evt_7");
134 });
135
136 it("skips a MALFORMED payload (logged via onError, never crashes the loop)", async () => {
137 const fake = makeFakeClient();
138 const seen: DomainEvent[] = [];
139 const errors: unknown[] = [];
140 const source = createDomainEventSource((e) => seen.push(e), {
141 client: fake.client,
142 onError: (err) => errors.push(err),
143 });
144 source.start();
145 fake.prime([
146 [
147 DOMAIN_EVENTS_STREAM,
148 [["6", ["event", "comment.created", "payload", "{not-json", "ts", "t"]]],
149 ],
150 ]);
151 await settle();
152 fake.prime([
153 [DOMAIN_EVENTS_STREAM, [entry("8", "comment.created", { commentId: "ok", pageId: "p" })]],
154 ]);
155 await settle();
156 await source.stop();
157 expect(errors).toHaveLength(1);
158 expect(seen).toHaveLength(1);
159 expect((seen[0]!.payload as { commentId: string }).commentId).toBe("ok");
160 });
161
162 it("skips an entry WITHOUT an event name", async () => {
163 const fake = makeFakeClient();
164 const seen: DomainEvent[] = [];
165 const source = createDomainEventSource((e) => seen.push(e), { client: fake.client });
166 source.start();
167 fake.prime([[DOMAIN_EVENTS_STREAM, [["11", ["payload", "{}", "ts", "t"]]]]]);
168 await settle();
169 await source.stop();
170 expect(seen).toHaveLength(0);
171 });
172
173 it("advances the cursor past delivered entries (XREAD starts at the last id)", async () => {
174 const fake = makeFakeClient();
175 const source = createDomainEventSource(() => {}, { client: fake.client });
176 source.start();
177 fake.prime([
178 [DOMAIN_EVENTS_STREAM, [entry("42", "comment.created", { commentId: "c", pageId: "p" })]],
179 ]);
180 await settle();
181 await source.stop();
182 const reads = fake.calls;
183 expect(reads.length).toBeGreaterThanOrEqual(2);
184 const after = reads.find((c) => c[c.length - 1] === "42");
185 expect(after).toBeDefined();
186 expect(reads[0]![reads[0]!.length - 1]).toBe("$");
187 });
188
189 it("does nothing after stop() (the loop exits)", async () => {
190 const fake = makeFakeClient();
191 const seen: DomainEvent[] = [];
192 const source = createDomainEventSource((e) => seen.push(e), { client: fake.client });
193 source.start();
194 await source.stop();
195 fake.prime([
196 [DOMAIN_EVENTS_STREAM, [entry("9", "comment.created", { commentId: "c", pageId: "p" })]],
197 ]);
198 await settle();
199 expect(seen).toHaveLength(0);
200 });
201
202 it("a dead client (throwing xread) back-offs and does not crash the loop", async () => {
203 const errors: unknown[] = [];
204 const client: DomainEventStreamClient = {
205 async xread() {
206 throw new Error("redis down");
207 },
208 };
209 const seen: DomainEvent[] = [];
210 const source = createDomainEventSource((e) => seen.push(e), {
211 client,
212 onError: (err) => errors.push(err),
213 });
214 source.start();
215 await sleep(30);
216 await source.stop();
217 expect(errors.length).toBeGreaterThanOrEqual(1);
218 expect(seen).toHaveLength(0);
219 });
220});
221
