1
40import { expect, request as pwRequest } from "@playwright/test";
41import * as Y from "yjs";
42import WebSocket from "ws";
43import * as fs from "node:fs";
44import * as net from "node:net";
45import { spawn, execFileSync } from "node:child_process";
46import type { Browser, Page, BrowserContext, APIRequestContext } from "@playwright/test";
47import type { AuthState, Workspace, PageDto } from "../fixtures";
48import { DEMO_EMAIL, DEMO_PASSWORD } from "../fixtures";
49import {
50 addMemberViaSeam,
51 createBlock,
52 createDatabase,
53 createPage,
54 createWorkspace,
55 register,
56 login,
57 seamServiceToken,
58 getBlocks,
59} from "../utils/api";
60
61const API_BASE = process.env.E2E_API_URL ?? "http://localhost:4000";
62const WS_BASE = process.env.E2E_WS_URL ?? "ws://localhost:4001";
63
64export const CONVERGE_MS = 3_000;
65
66
67
68type Cookie = {
69 name: string;
70 value: string;
71 domain: string;
72 path: string;
73 expires: number;
74 httpOnly: boolean;
75 secure: boolean;
76 sameSite: "Strict" | "Lax" | "None";
77};
78
79function cookiesFromLogin(jar: Cookie[]): AuthState["cookies"] {
80 return jar.map((c) => ({
81 name: c.name,
82 value: c.value,
83 domain: c.domain,
84 path: c.path,
85 expires: c.expires,
86 httpOnly: c.httpOnly,
87 secure: c.secure,
88 sameSite: c.sameSite,
89 }));
90}
91
92
93interface OwnerTab {
94 context: BrowserContext;
95 page: Page;
96 close: () => Promise<void>;
97}
98
99async function contextWithLogin(
100 browser: Browser,
101 email: string,
102 password: string,
103): Promise<OwnerTab> {
104 const plain = await browser.newContext();
105 const res = await plain.request.post("/api/v1/auth/login", {
106 data: { email, password },
107 });
108 if (res.status() !== 200) {
109 throw new Error(`contextWithLogin(${email}): login failed (${res.status()})`);
110 }
111 const jar = await plain.cookies();
112 await plain.close();
113 const context = await browser.newContext({
114 storageState: { cookies: cookiesFromLogin(jar), origins: [] },
115 });
116 const page = await context.newPage();
117 return { context, page, close: () => context.close() };
118}
119
120
128export interface NewUser {
129 user: { email: string; password: string };
130 context: BrowserContext;
131 page: Page;
132 api: APIRequestContext;
133
134 token: string;
135 close: () => Promise<void>;
136}
137
138export async function newUser(
139 browser: Browser,
140 name: string,
141): Promise<{ user: { email: string; password: string } }> {
142 const email = `e2e-rt-${Date.now()}-${Math.random().toString(36).slice(2, 8)}@factory.local`;
143 const password = `e2e-${Date.now().toString(36)}`;
144 const plain = await browser.newContext();
145 await register(plain.request, email, password, name);
146 await plain.close();
147 return { user: { email, password } };
148}
149
150
156export async function authedUser(
157 browser: Browser,
158 user: { email: string; password: string },
159): Promise<NewUser> {
160 const plain = await browser.newContext();
161 const logged = await login(plain.request, user.email, user.password);
162 const token = logged.accessToken ?? "";
163 const jar = await plain.cookies();
164 await plain.close();
165 const context = await browser.newContext({
166 storageState: { cookies: cookiesFromLogin(jar), origins: [] },
167 });
168 const page = await context.newPage();
169 return {
170 user,
171 context,
172 page,
173 api: context.request,
174 token,
175 close: () => context.close(),
176 };
177}
178
179
180export async function ownerTab(browser: Browser) {
181 return contextWithLogin(browser, DEMO_EMAIL, DEMO_PASSWORD);
182}
183
184
185
186export interface RtPageSeed {
187 ws: Workspace;
188 page: PageDto;
189 url: string;
190 owner: OwnerTab;
191 a: NewUser;
192 b: NewUser;
193 closeAll: () => Promise<void>;
194}
195
196
209export async function seedPage(
210 browser: Browser,
211 api: APIRequestContext,
212 blockCount = 3,
213): Promise<RtPageSeed> {
214 const ws = await createWorkspace(api, `RT ${Date.now()}`);
215 const page = await createPage(api, ws.id, { title: "Realtime page" });
216 const seeds = ["alpha", "beta", "gamma", "delta", "epsilon", "zeta"];
217 for (let i = 0; i < blockCount; i += 1) {
218 await createBlock(api, page.id, {
219 type: "paragraph",
220 content: { text: seeds[i] ?? `seed-${i}` },
221 position: `${i + 1}.0`,
222 });
223 }
224 const owner = await ownerTab(browser);
225 const aReg = await newUser(browser, "Alice");
226 const bReg = await newUser(browser, "Bob");
227
228
229 const token = seamServiceToken();
230 await addMemberViaSeam(token, ws.id, aReg.user.email, "ADMIN");
231 await addMemberViaSeam(token, ws.id, bReg.user.email, "ADMIN");
232
233 const a = await authedUser(browser, aReg.user);
234 const b = await authedUser(browser, bReg.user);
235 const url = `/w/${ws.id}/page/${page.id}`;
236 return {
237 ws,
238 page,
239 url,
240 owner,
241 a,
242 b,
243 closeAll: async () => {
244 await owner.close().catch(() => {});
245 await a.close().catch(() => {});
246 await b.close().catch(() => {});
247 },
248 };
249}
250
251
252
253
254export function blockText(page: Page, nth: number): Promise<string> {
255 return page.locator('[data-testid^="block-"]').nth(nth).innerText().then((t) => t.trim());
256}
257
258
260export async function typeInto(page: Page, nth: number, text: string): Promise<void> {
261 const block = page.locator('[data-testid^="block-"]').nth(nth);
262 await block.click();
263 await page.keyboard.press("End");
264 await page.keyboard.type(text);
265}
266
267
268export async function expectConverged(
269 a: Page,
270 b: Page,
271 checks: { nth: number; contains: string }[],
272 timeoutMs = CONVERGE_MS * 4,
273): Promise<void> {
274 for (const p of [a, b]) {
275 for (const c of checks) {
276 await expect
277 .poll(() => blockText(p, c.nth), {
278 timeout: timeoutMs,
279 message: `block ${c.nth} to contain ${c.contains}`,
280 })
281 .toContain(c.contains);
282 }
283 }
284}
285
286
287export function titleInput(page: Page) {
288 return page.locator('input[aria-label="Page title"]');
289}
290
291
298export async function expectAtRestBlock(
299 api: APIRequestContext,
300 pageId: string,
301 nth: number,
302 text: string,
303): Promise<void> {
304 const blocks = await getBlocks(api, pageId);
305 const got = blocks[nth]?.content as { text?: unknown } | unknown[] | undefined;
306 const gotText =
307 got !== null && typeof got === "object" && !Array.isArray(got) && "text" in got
308 ? String((got as { text?: unknown }).text ?? "")
309 : Array.isArray(got)
310 ? got.join("")
311 : String(got ?? "");
312 expect(
313 gotText,
314 `at-rest block ${nth} (API truth) — expected "${text}", got "${gotText}"`,
315 ).toBe(text);
316}
317
318
319
320export interface NetworkCut {
321 restore: () => Promise<void>;
322}
323
324
341export async function cutNetwork(context: BrowserContext): Promise<NetworkCut> {
342 await context.setOffline(true);
343 return {
344 restore: async () => {
345 await context.setOffline(false);
346 },
347 };
348}
349
350
351
352export interface RawClient {
353 ws: WebSocket;
354
355 frames: () => Record<string, unknown>[];
356
357 binaries: () => Uint8Array[];
358 nextFrame: (type: string, timeoutMs?: number) => Promise<Record<string, unknown>>;
359 frameCount: (type: string) => number;
360 errorFrames: () => Record<string, unknown>[];
361
362 lastAckSeq: () => number | null;
363 sendText: (obj: unknown) => void;
364
366 sendUpdate: (doc: Y.Doc, text: string) => void;
367 close: (code?: number) => void;
368 onClose: (timeoutMs?: number) => Promise<{ code: number; reason: string }>;
369 closed: () => { code: number; reason: string } | null;
370}
371
372
378export async function rawClient(
379 token: string,
380 room: string,
381 wsBase: string = WS_BASE,
382): Promise<RawClient> {
383 const url = `${wsBase.replace(/\/+$/, "")}/realtime?token=${encodeURIComponent(
384 token,
385 )}&room=${encodeURIComponent(room)}`;
386 const ws = new WebSocket(url);
387 const textQueue: Record<string, unknown>[] = [];
388 const binQueue: Uint8Array[] = [];
389 let closeInfo: { code: number; reason: string } | null = null;
390 const closeWaiters: {
391 resolve: (v: { code: number; reason: string }) => void;
392 timer: ReturnType<typeof setTimeout>;
393 }[] = [];
394
395 ws.on("message", (data: Buffer, isBinary: boolean) => {
396 if (isBinary) {
397
398 binQueue.push(new Uint8Array(data.subarray(1)));
399 return;
400 }
401 try {
402 textQueue.push(JSON.parse(data.toString("utf8")) as Record<string, unknown>);
403 } catch {
404
405 }
406 });
407 ws.on("close", (code, reason) => {
408 closeInfo = { code, reason: reason.toString() };
409 for (const w of closeWaiters.splice(0)) {
410 clearTimeout(w.timer);
411 w.resolve(closeInfo!);
412 }
413 });
414
415 await new Promise<void>((resolve, reject) => {
416 const t = setTimeout(
417 () => reject(new Error(`rawClient connect timeout (${room})`)),
418 8_000,
419 );
420 ws.once("open", () => {
421 clearTimeout(t);
422 resolve();
423 });
424 ws.once("error", (err) => {
425 clearTimeout(t);
426 reject(err instanceof Error ? err : new Error(String(err)));
427 });
428 });
429
430 return {
431 ws,
432 frames: () => [...textQueue],
433 binaries: () => [...binQueue],
434 nextFrame: (type, timeoutMs = 8_000) =>
435 new Promise((resolve, reject) => {
436 const i = textQueue.findIndex((f) => f.type === type);
437 if (i !== -1) {
438 resolve(textQueue.splice(i, 1)[0]!);
439 return;
440 }
441 const timer = setTimeout(
442 () =>
443 reject(
444 new Error(
445 `timeout waiting for "${type}" (got ${textQueue.length} frames)`,
446 ),
447 ),
448 timeoutMs,
449 );
450 const onMsg = (data: Buffer, isBinary: boolean) => {
451 if (isBinary) return;
452 try {
453 const f = JSON.parse(data.toString("utf8")) as Record<string, unknown>;
454 if (f.type !== type) return;
455 const j = textQueue.indexOf(f);
456 if (j !== -1) textQueue.splice(j, 1);
457 ws.off("message", onMsg);
458 clearTimeout(timer);
459 resolve(f);
460 } catch {
461
462 }
463 };
464 ws.on("message", onMsg);
465 }),
466 frameCount: (type) => textQueue.filter((f) => f.type === type).length,
467 errorFrames: () => textQueue.filter((f) => f.type === "error"),
468 lastAckSeq: () => {
469 for (let i = textQueue.length - 1; i >= 0; i -= 1) {
470 const f = textQueue[i]!;
471 if (f.type === "ack" && typeof f.seq === "number") return f.seq;
472 }
473 return null;
474 },
475 sendText: (obj) => ws.send(JSON.stringify(obj)),
476 sendUpdate: (doc, text) => {
477 const yText = doc.getText("t");
478
479
480
481
482 const sv = Y.encodeStateVector(doc);
483 yText.insert(0, text);
484 const delta = Y.encodeStateAsUpdate(doc, sv);
485 const out = new Uint8Array(delta.byteLength + 1);
486 out[0] = 0x01;
487 out.set(delta, 1);
488 ws.send(out);
489 },
490 close: (code = 1000) => ws.close(code),
491 onClose: (timeoutMs = 8_000) =>
492 new Promise((resolve, reject) => {
493 if (closeInfo !== null) {
494 resolve(closeInfo);
495 return;
496 }
497 const w = {
498 resolve,
499 timer: setTimeout(
500 () => reject(new Error("timeout waiting for close (socket still open)")),
501 timeoutMs,
502 ),
503 };
504 closeWaiters.push(w);
505 }),
506 closed: () => closeInfo,
507 };
508}
509
510
511
512
527export async function restartRealtimeServer(): Promise<void> {
528 const pid = pidListeningOn(4001);
529 if (pid === null) {
530 throw new Error("restartRealtimeServer: nothing is listening on :4001 (stack not up?)");
531 }
532 const info = procInfo(pid);
533 if (info === null) {
534 throw new Error(`restartRealtimeServer: cannot read /proc/${pid} (non-Linux runner?)`);
535 }
536
537
538 for (const p of [...descendantsOf(pid), pid].sort((x, y) => y - x)) {
539 try {
540 process.kill(p, "SIGTERM");
541 } catch {
542
543 }
544 }
545 await waitForPortFree(4001, 30_000);
546
547
548 const child = spawn(info.argv[0]!, info.argv.slice(1), {
549 cwd: info.cwd,
550 env: info.env,
551 stdio: "ignore",
552 detached: true,
553 });
554 child.unref();
555 await waitForPort(4001, 90_000);
556 await waitForPort(4002, 90_000);
557}
558
559
560function pidListeningOn(port: number): number | null {
561 let out: string;
562 try {
563 out = execFileSync("ss", ["-tlnp"], { encoding: "utf8" });
564 } catch {
565 return null;
566 }
567 for (const line of out.split("\n")) {
568
569 const portMatch = new RegExp(`:\\b${port}\\b`).test(line);
570 if (!portMatch) continue;
571 const pid = /pid=(\d+)/.exec(line)?.[1];
572 if (pid !== undefined) return Number(pid);
573 }
574 return null;
575}
576
577
578function descendantsOf(pid: number): number[] {
579 const children = new Map<number, number[]>();
580 const pids = fs
581 .readdirSync("/proc")
582 .filter((n) => /^\d+$/.test(n))
583 .map(Number);
584 for (const p of pids) {
585 try {
586
587 const stat = fs.readFileSync(`/proc/${p}/stat`, "utf8");
588 const close = stat.lastIndexOf(")");
589 const fields = stat.slice(close + 2).split(" ");
590 const ppid = Number(fields[1]);
591 if (!Number.isFinite(ppid)) continue;
592 const arr = children.get(ppid) ?? [];
593 arr.push(p);
594 children.set(ppid, arr);
595 } catch {
596
597 }
598 }
599 const out: number[] = [];
600 const stack = [...(children.get(pid) ?? [])];
601 while (stack.length > 0) {
602 const p = stack.pop()!;
603 out.push(p);
604 for (const c of children.get(p) ?? []) stack.push(c);
605 }
606 return out;
607}
608
609interface ProcInfo {
610 argv: string[];
611 env: Record<string, string>;
612 cwd: string;
613}
614
615
616function procInfo(pid: number): ProcInfo | null {
617 try {
618 const argv = fs
619 .readFileSync(`/proc/${pid}/cmdline`, "utf8")
620 .split("\0")
621 .filter((s) => s.length > 0);
622 const cwd = fs.readlinkSync(`/proc/${pid}/cwd`);
623 const raw = fs.readFileSync(`/proc/${pid}/environ`, "utf8");
624 const env: Record<string, string> = {};
625 for (const entry of raw.split("\0")) {
626 const eq = entry.indexOf("=");
627 if (eq > 0) env[entry.slice(0, eq)] = entry.slice(eq + 1);
628 }
629 if (argv.length === 0) return null;
630 return { argv, env, cwd };
631 } catch {
632 return null;
633 }
634}
635
636export async function waitForPort(port: number, timeoutMs: number): Promise<void> {
637 const deadline = Date.now() + timeoutMs;
638
639 while (true) {
640 const ok = await new Promise<boolean>((resolve) => {
641 const s = net.connect(port, "127.0.0.1");
642 s.once("connect", () => {
643 s.destroy();
644 resolve(true);
645 });
646 s.once("error", () => resolve(false));
647 });
648 if (ok) return;
649 if (Date.now() >= deadline) throw new Error(`port ${port} not accepting within ${timeoutMs}ms`);
650 await new Promise((r) => setTimeout(r, 500));
651 }
652}
653
654async function waitForPortFree(port: number, timeoutMs: number): Promise<void> {
655 const deadline = Date.now() + timeoutMs;
656
657 while (true) {
658 const ok = await new Promise<boolean>((resolve) => {
659 const s = net.connect(port, "127.0.0.1");
660 s.once("connect", () => {
661 s.destroy();
662 resolve(true);
663 });
664 s.once("error", () => resolve(false));
665 });
666 if (!ok) return;
667 if (Date.now() >= deadline) throw new Error(`port ${port} still bound after ${timeoutMs}ms`);
668 await new Promise((r) => setTimeout(r, 300));
669 }
670}
671
672
673
674
675export async function apiReachable(): Promise<boolean> {
676 try {
677 const req = await pwRequest.newContext({ baseURL: API_BASE, timeout: 5_000 });
678 const res = await req.get("/api/v1/health").catch(() => null);
679 await req.dispose();
680 return res !== null;
681 } catch {
682 return false;
683 }
684}
685
686
687export async function seedDatabase(
688 api: APIRequestContext,
689 wsId: string,
690 title = "RT Records",
691) {
692 return createDatabase(api, wsId, {
693 title,
694 pageId: null,
695 properties: [{ name: "Name", type: "title" }, { name: "Note", type: "text" }],
696 });
697}
698
