1
14import { afterAll, beforeAll, describe, expect, it } from "vitest";
15import {
16 buildRealtimeFixture,
17 connectClient,
18 mintToken,
19 probeUpgrade,
20 type RealtimeFixture,
21} from "./rt-helpers";
22import type { DomainEvent } from "@factory/shared";
23import type { PageUpdatedSource } from "@factory/realtime/src/page-updated-source";
24
25
26function makeFakeSource() {
27 let sink: ((event: DomainEvent) => void) | null = null;
28 const source: PageUpdatedSource = {
29 start(): void {
30
31 },
32 async stop(): Promise<void> {
33 sink = null;
34 },
35 };
36 return {
37 source,
38 factory: (onPageUpdated: (event: DomainEvent) => void): PageUpdatedSource => {
39 sink = onPageUpdated;
40 return source;
41 },
42
43 deliver(pageId: string, version?: number): void {
44 expect(sink).not.toBeNull();
45 sink!({
46 id: `evt_${Math.random().toString(36).slice(2)}`,
47 name: "page.updated",
48 payload: version !== undefined ? { pageId, version } : { pageId },
49 ts: new Date().toISOString(),
50 });
51 },
52 };
53}
54
55
56async function connectControl(fx: RealtimeFixture, sub = "u-ctl", workspaces: string[] = ["w1"]) {
57 const token = mintToken({ sub, workspaces });
58 const url = `${fx.wsUrl}/realtime?token=${encodeURIComponent(token)}`;
59 return connectClient(fx, { url });
60}
61
62describe("TASK-0092 6b — the roomless control socket + block:changed fan-out", () => {
63 let fx: RealtimeFixture;
64 let src: ReturnType<typeof makeFakeSource>;
65
66 beforeAll(async () => {
67 src = makeFakeSource();
68 fx = await buildRealtimeFixture({
69 pages: new Set(["p-ctl", "p-other"]),
70 workspaces: { "p-ctl": "w1", "p-other": "w1" },
71 pageUpdatedSource: src.factory,
72 });
73 });
74
75 afterAll(async () => {
76 await fx.stop();
77 });
78
79 it("admits a token-only (roomless) upgrade and sends a control hello", async () => {
80 const c = await connectControl(fx);
81 try {
82 const hello = await c.nextFrame("hello");
83 expect(hello.type).toBe("hello");
84 expect(typeof hello.connId).toBe("string");
85 } finally {
86 c.close();
87 }
88 });
89
90 it("fans a block:changed refetch-hint to the page's control channel on a page.updated", async () => {
91 const c = await connectControl(fx);
92 try {
93 await c.nextFrame("hello");
94 c.sendText({ type: "subscribe", channel: "page:p-ctl" });
95 await fx.settle(50);
96
97 src.deliver("p-ctl", 7);
98
99 const hint = await c.nextFrame("block:changed");
100 expect(hint).toEqual({ type: "block:changed", pageId: "p-ctl", version: 7 });
101 } finally {
102 c.close();
103 }
104 });
105
106 it("sends the hint WITHOUT a version when the page.updated has none (a REST write)", async () => {
107 const c = await connectControl(fx);
108 try {
109 await c.nextFrame("hello");
110 c.sendText({ type: "subscribe", channel: "page:p-ctl" });
111 await fx.settle(50);
112
113 src.deliver("p-ctl");
114
115 const hint = await c.nextFrame("block:changed");
116 expect(hint).toEqual({ type: "block:changed", pageId: "p-ctl" });
117 } finally {
118 c.close();
119 }
120 });
121
122 it("only fans to the subscribed channel (page A's update does not reach page B's subscriber)", async () => {
123 const other = await connectControl(fx);
124 try {
125 await other.nextFrame("hello");
126 other.sendText({ type: "subscribe", channel: "page:p-other" });
127 await fx.settle(50);
128
129 src.deliver("p-ctl", 3);
130
131
132 await expect(other.nextFrame("block:changed", 400)).rejects.toThrow();
133 } finally {
134 other.close();
135 }
136 });
137
138 it("a page:<id> channel and a workspace:<id> channel are both accepted (workspace rides no frames yet)", async () => {
139 const c = await connectControl(fx);
140 try {
141 await c.nextFrame("hello");
142 c.sendText({ type: "subscribe", channel: "page:p-ctl" });
143 c.sendText({ type: "subscribe", channel: "workspace:w1" });
144 await fx.settle(50);
145
146
147
148 src.deliver("p-ctl", 1);
149 const hint = await c.nextFrame("block:changed");
150 expect(hint.pageId).toBe("p-ctl");
151 expect(c.isOpen()).toBe(true);
152 } finally {
153 c.close();
154 }
155 });
156
157 it("unsubscribe stops the fan-out for that channel", async () => {
158 const c = await connectControl(fx);
159 try {
160 await c.nextFrame("hello");
161 c.sendText({ type: "subscribe", channel: "page:p-ctl" });
162 await fx.settle(50);
163 c.sendText({ type: "unsubscribe", channel: "page:p-ctl" });
164 await fx.settle(50);
165
166 src.deliver("p-ctl", 5);
167 await expect(c.nextFrame("block:changed", 400)).rejects.toThrow();
168 } finally {
169 c.close();
170 }
171 });
172
173 it("a control socket's sync.request is answered with an empty sync.state (no doc)", async () => {
174 const c = await connectControl(fx);
175 try {
176 await c.nextFrame("hello");
177 c.sendText({ type: "sync.request" });
178 const st = await c.nextFrameWhere((f) => f.type === "sync.state" && f.done === true);
179 expect(st.type).toBe("sync.state");
180 } finally {
181 c.close();
182 }
183 });
184
185 it("a malformed room key is STILL rejected (400) — only a MISSING room is a control socket", async () => {
186 const token = mintToken({ sub: "u-badroom", workspaces: ["w1"] });
187 const r = await probeUpgrade(fx, { token, room: "nonsense" });
188 expect(r.status).toBe(400);
189 });
190});
191
