diff --git a/packages/cultmesh-ts/test/interop/cultmesh-quic-peer.ts b/packages/cultmesh-ts/test/interop/cultmesh-quic-peer.ts index 909cdd86..b3f06412 100644 --- a/packages/cultmesh-ts/test/interop/cultmesh-quic-peer.ts +++ b/packages/cultmesh-ts/test/interop/cultmesh-quic-peer.ts @@ -1,15 +1,41 @@ -// The TypeScript half of the QUIC realtime interop lane: a one-shot process -// that dials an advertised `cultmesh-state+quic://` endpoint with the native -// connector, receives `--expect` frames from a peer (normally the C# managed -// provider started by `cultnet-interop.test.ts`'s -// "CultMesh QUIC realtime" test), and prints them as one JSON line. It never -// imports `cultmesh-ts` itself (the harness drives processes, and the -// dependency direction is `cultmesh-ts` -> `cultnet-ts`, never the reverse); -// this script reaches the connector through its own package's `src/`. +// The TypeScript half of the QUIC realtime interop lane. Two modes: +// +// `dial` (Cut 4): a one-shot process that dials an advertised +// `cultmesh-state+quic://` endpoint with the native connector, receives +// `--expect` frames from a peer (normally the C# managed provider started by +// `cultnet-interop.test.ts`'s "CultMesh QUIC realtime" test), and prints them +// as one JSON line. +// +// `serve` (Cut 5): the StreamPixels role. Opens a `CultMeshQuicRealtimeProvider` +// listener, prints `{status: "ready", endpoint}` (the same shape the C# +// `quic-realtime-serve` peer prints), waits for a first peer before +// broadcasting (mirroring `QuicRealtimeServeAsync`'s +// `WaitForFirstConnectionAsync`, since a broadcast to nobody would otherwise +// silently no-op), then broadcasts `--frames` frames at `--delivery` with +// `--interval-ms` between them, watching stdin for EOF as its own shutdown +// signal like the C# peer's `WatchStdinCloseAsync`. +// +// Neither mode imports `cultmesh-ts` itself (the harness drives processes, +// and the dependency direction is `cultmesh-ts` -> `cultnet-ts`, never the +// reverse); this script reaches the provider/connector through its own +// package's `src/`. +import { readFileSync } from "node:fs"; +import { join } from "node:path"; import process from "node:process"; -import { CultMeshQuicRealtimeConnector, type CultMeshRealtimeTarget } from "../../src/realtime-quic"; +import { + CultMeshQuicRealtimeConnector, + CultMeshQuicRealtimeProvider, + type CultMeshRealtimeTarget, +} from "../../src/realtime-quic"; +import type { CultMeshRealtimeDelivery, CultMeshRealtimeFrame } from "../../src/realtime-wire"; + +// `__dirname` at runtime is `dist-test/test/interop`; fixtures are binary and +// are never compiled/copied there, so they are read from their source +// location, three levels up. The same self-signed test credential the unit +// tests use (`test/fixtures/README.md`): not a production credential. +const FIXTURE_P12 = join(__dirname, "..", "..", "..", "test", "fixtures", "quic-test.p12"); function parseArgs(argv: readonly string[]): Map { const args = new Map(); @@ -47,9 +73,77 @@ function writeLog(event: string, payload: Record): void { process.stderr.write(`${JSON.stringify({ event, ...payload })}\n`); } +function watchStdinCloseAsync(): Promise { + return new Promise((resolve) => { + process.stdin.resume(); + process.stdin.once("end", resolve); + }); +} + +async function serveAsync(args: Map): Promise { + const frames = optionalIntArg(args, "frames", 5); + const deliveryArg = args.get("delivery") ?? "latest-only"; + if (deliveryArg !== "latest-only" && deliveryArg !== "reliable-ordered") { + throw new Error(`Unsupported --delivery '${deliveryArg}'.`); + } + const delivery = deliveryArg as CultMeshRealtimeDelivery; + const intervalMs = optionalIntArg(args, "interval-ms", 50); + const port = optionalIntArg(args, "port", 0); + + const provider = await CultMeshQuicRealtimeProvider.listen({ + host: "127.0.0.1", + port, + serverCertificate: { pkcs12: readFileSync(FIXTURE_P12), password: "" }, + }); + writeJsonLine({ status: "ready", endpoint: provider.advertisedEndpoint }); + + try { + // A broadcast to zero peers is not an error, but it sends nothing: + // matching the C# `quic-realtime-serve` peer, wait for a first + // connection, bounded, so a peer that never shows up still exits. + const connectDeadline = Date.now() + 30_000; + while (provider.connectionCount === 0) { + if (Date.now() >= connectDeadline) throw new Error("quic-realtime-serve: no client connected within 30000ms."); + await new Promise((resolve) => setTimeout(resolve, 20)); + } + + const stdinClosed = watchStdinCloseAsync(); + let stopped = false; + void stdinClosed.then(() => { + stopped = true; + }); + for (let sequence = 1; sequence <= frames && !stopped; sequence += 1) { + const frame: CultMeshRealtimeFrame = { + channelId: "interop.quic-realtime", + schemaId: "gamecult.interop.quic_realtime_frame.v0", + bodyId: "interop:quic-realtime:frame", + producerEpoch: 1n, + sequence: BigInt(sequence), + delivery, + payload: Buffer.from(`frame-${sequence}`, "utf8"), + }; + await provider.broadcast(frame); + writeLog("quic-realtime-serve", { sent: sequence }); + if (intervalMs > 0) await new Promise((resolve) => setTimeout(resolve, intervalMs)); + } + } finally { + provider.dispose(); + // Mirrors the C# peer returning from `QuicRealtimeServeAsync`: once + // broadcasting is done (or stdin closed early), let the process exit on + // its own instead of keeping the event loop alive on a stdin that may + // never close. The harness still ends/kills this process explicitly in + // its own cleanup; this only lets a plain manual run exit cleanly too. + process.stdin.pause(); + } +} + async function main(): Promise { const [mode, ...rest] = process.argv.slice(2); - if (mode !== "dial") throw new Error("Expected mode: dial"); + if (mode === "serve") { + await serveAsync(parseArgs(rest)); + return; + } + if (mode !== "dial") throw new Error("Expected mode: dial | serve"); const args = parseArgs(rest); const endpoint = requireArg(args, "endpoint"); const expect = optionalIntArg(args, "expect", 1);