diff --git a/packages/cultmesh-ts/test/fixtures/quic-test-cert.der b/packages/cultmesh-ts/test/fixtures/quic-test-cert.der new file mode 100644 index 00000000..5cce6f94 Binary files /dev/null and b/packages/cultmesh-ts/test/fixtures/quic-test-cert.der differ diff --git a/packages/cultmesh-ts/test/fixtures/quic-test.p12 b/packages/cultmesh-ts/test/fixtures/quic-test.p12 new file mode 100644 index 00000000..e9d03ce6 Binary files /dev/null and b/packages/cultmesh-ts/test/fixtures/quic-test.p12 differ diff --git a/packages/cultmesh-ts/test/realtime-quic-consumer.test.ts b/packages/cultmesh-ts/test/realtime-quic-consumer.test.ts new file mode 100644 index 00000000..8e4128fb --- /dev/null +++ b/packages/cultmesh-ts/test/realtime-quic-consumer.test.ts @@ -0,0 +1,470 @@ +// The TypeScript QUIC realtime consumer, proven two ways: directly against a +// raw listener opened through the native runtime (no provider semantics yet, +// mirrors `CultMeshNativeQuicRealtimeTransportTests.cs`), and through +// `CultMeshQuicRealtimeSessionManager` with a static lookup source and routes +// signed with `node:crypto`, mirroring `CultMeshQuicRealtimeTransportTests.cs` +// and `CultMeshSessions.cs`'s realtime binding. +// +// Skips instead of failing when no native bridge is available; see +// `realtime-quic-native.test.ts` for why. + +import assert from "node:assert/strict"; +import { createHash, generateKeyPairSync, sign as cryptoSign, type KeyObject } from "node:crypto"; +import { readFileSync } from "node:fs"; +import { join } from "node:path"; +import test from "node:test"; + +import { + canonicalRoute, + type CultMeshAuthorityRouteView, + type CultMeshAuthorityTrustPolicy, + type CultMeshP256PublicKey, +} from "cultnet-ts"; +import type { CultMeshAuthorityRouteMessage, CultMeshVerseDescriptorMessage } from "cultnet-ts/contracts"; + +import { + CultMeshQuicNativeRuntime, + CULTMESH_QUIC_EVENT_LISTENER_NEW_CONNECTION, + CULTMESH_QUIC_STREAM_LATEST_ONLY, +} from "../src/realtime-quic-native"; +import { + CultMeshQuicRealtimeConnector, + CultMeshQuicRealtimeSessionManager, + CultMeshStaticRealtimeLookupSource, + CULTMESH_REALTIME_STATE_PROTOCOL_ID, + type CultMeshRealtimeCandidate, + type CultMeshRealtimeTarget, + type CultMeshRealtimeTransport, + type CultMeshRealtimeTransportConnector, +} from "../src/realtime-quic"; +import { encodeRealtimeFrame, type CultMeshRealtimeFrame } from "../src/realtime-wire"; + +const FIXTURE_P12 = join(__dirname, "fixtures", "quic-test.p12"); +const FIXTURE_DER = join(__dirname, "fixtures", "quic-test-cert.der"); + +function nativeBridgeAvailable(): boolean { + const dir = process.env.CULTMESH_QUIC_NATIVE_DIR; + if (!dir) return false; + try { + const bridge = process.platform === "win32" ? "gamecult_mesh_quic_native.dll" : "libgamecult_mesh_quic_native.so"; + readFileSync(join(dir, bridge)); + return true; + } catch { + return false; + } +} + +function fixturePinHex(): string { + return createHash("sha256").update(readFileSync(FIXTURE_DER)).digest("hex"); +} + +/** A bare loopback listener over the raw native runtime, no provider semantics. */ +class RawQuicListener { + private constructor( + private readonly runtime: CultMeshQuicNativeRuntime, + private readonly listenerId: bigint, + readonly boundPort: number, + ) {} + + static async open(): Promise { + const runtime = await CultMeshQuicNativeRuntime.open(); + const pkcs12 = readFileSync(FIXTURE_P12); + const { listenerId, boundPort } = runtime.listenerOpen("127.0.0.1", 0, pkcs12, ""); + return new RawQuicListener(runtime, listenerId, boundPort); + } + + /** Waits for one inbound connection and returns its native connection id. */ + async acceptOnce(): Promise { + return await new Promise((resolve) => { + this.runtime.onListenerEvent(this.listenerId, (event) => { + if (event.type === CULTMESH_QUIC_EVENT_LISTENER_NEW_CONNECTION) resolve(event.connectionId); + }); + }); + } + + /** Sends one frame to an accepted connection on a fresh stream of `kind`. */ + sendFrame(connectionId: bigint, frame: CultMeshRealtimeFrame, kind: number): void { + const streamId = this.runtime.streamOpen(connectionId, kind); + this.runtime.streamSendFrame(streamId, encodeRealtimeFrame(frame), true); + } + + async close(): Promise { + this.runtime.listenerClose(this.listenerId); + await this.runtime.release(); + } +} + +function testFrame(overrides: Partial = {}): CultMeshRealtimeFrame { + return { + channelId: "aetheria.entities", + schemaId: "eve.entity_soa.v1", + bodyId: "body:aetheria:entities", + producerEpoch: 7n, + sequence: 1n, + delivery: "latest-only", + payload: new Uint8Array([3, 4, 5]), + ...overrides, + }; +} + +test("connect with the correct pin succeeds and receives a frame", async (t) => { + if (!nativeBridgeAvailable()) return void t.skip("no native bridge available"); + const listener = await RawQuicListener.open(); + try { + const target: CultMeshRealtimeTarget = { verseId: "aetheria", authorityRuntimeId: "service:aetheria.daemon" }; + const candidate: CultMeshRealtimeCandidate = { + endpoint: `cultmesh-state+quic://127.0.0.1:${listener.boundPort}?cert-sha256=${fixturePinHex()}`, + authorityRuntimeId: target.authorityRuntimeId, + priority: 0, + generation: "gen-1", + }; + const connector = new CultMeshQuicRealtimeConnector(); + const accepted = listener.acceptOnce(); + const transport = await connector.connect(candidate, target); + try { + assert.equal(transport.transportId, "msquic-realtime"); + const connectionId = await accepted; + listener.sendFrame(connectionId, testFrame(), CULTMESH_QUIC_STREAM_LATEST_ONLY); + const received = await transport.receiveFrame(); + assert.equal(received.bodyId, "body:aetheria:entities"); + assert.equal(received.sequence, 1n); + } finally { + transport.dispose(); + } + } finally { + await listener.close(); + } +}); + +test("connect refuses a wrong advertised pin", async (t) => { + if (!nativeBridgeAvailable()) return void t.skip("no native bridge available"); + const listener = await RawQuicListener.open(); + try { + const target: CultMeshRealtimeTarget = { verseId: "aetheria", authorityRuntimeId: "service:aetheria.daemon" }; + const candidate: CultMeshRealtimeCandidate = { + endpoint: `cultmesh-state+quic://127.0.0.1:${listener.boundPort}?cert-sha256=${"0".repeat(64)}`, + authorityRuntimeId: target.authorityRuntimeId, + priority: 0, + generation: "gen-1", + }; + const connector = new CultMeshQuicRealtimeConnector(); + await assert.rejects(connector.connect(candidate, target), /certificate|rejected/i); + } finally { + await listener.close(); + } +}); + +test("connect refuses a missing certificate pin when no validator is supplied", async (t) => { + if (!nativeBridgeAvailable()) return void t.skip("no native bridge available"); + const listener = await RawQuicListener.open(); + try { + const target: CultMeshRealtimeTarget = { verseId: "aetheria", authorityRuntimeId: "service:aetheria.daemon" }; + const candidate: CultMeshRealtimeCandidate = { + endpoint: `cultmesh-state+quic://127.0.0.1:${listener.boundPort}`, + authorityRuntimeId: target.authorityRuntimeId, + priority: 0, + generation: "gen-1", + }; + const connector = new CultMeshQuicRealtimeConnector(); + await assert.rejects(connector.connect(candidate, target), /certificate|rejected/i); + } finally { + await listener.close(); + } +}); + +test("a caller-supplied validator observes the target identity and the DER bytes", async (t) => { + if (!nativeBridgeAvailable()) return void t.skip("no native bridge available"); + const listener = await RawQuicListener.open(); + try { + const target: CultMeshRealtimeTarget = { verseId: "aetheria", authorityRuntimeId: "service:aetheria.daemon" }; + const candidate: CultMeshRealtimeCandidate = { + endpoint: `cultmesh-state+quic://127.0.0.1:${listener.boundPort}`, + authorityRuntimeId: target.authorityRuntimeId, + priority: 0, + generation: "gen-1", + }; + let observedTarget: CultMeshRealtimeTarget | undefined; + let observedDerLength = 0; + const connector = new CultMeshQuicRealtimeConnector({ + validateProviderCertificate: (validatorTarget, der) => { + observedTarget = validatorTarget; + observedDerLength = der.byteLength; + return true; + }, + }); + const transport = await connector.connect(candidate, target); + transport.dispose(); + assert.deepEqual(observedTarget, target); + assert.ok(observedDerLength > 0); + } finally { + await listener.close(); + } +}); + +test("a late older generation is dropped by the receive-side filter", async (t) => { + if (!nativeBridgeAvailable()) return void t.skip("no native bridge available"); + const listener = await RawQuicListener.open(); + try { + const target: CultMeshRealtimeTarget = { verseId: "aetheria", authorityRuntimeId: "service:aetheria.daemon" }; + const candidate: CultMeshRealtimeCandidate = { + endpoint: `cultmesh-state+quic://127.0.0.1:${listener.boundPort}?cert-sha256=${fixturePinHex()}`, + authorityRuntimeId: target.authorityRuntimeId, + priority: 0, + generation: "gen-1", + }; + const connector = new CultMeshQuicRealtimeConnector(); + const accepted = listener.acceptOnce(); + const transport = await connector.connect(candidate, target); + try { + const connectionId = await accepted; + listener.sendFrame(connectionId, testFrame({ producerEpoch: 8n, sequence: 0n }), CULTMESH_QUIC_STREAM_LATEST_ONLY); + const latest = await transport.receiveFrame(); + assert.equal(latest.producerEpoch, 8n); + assert.equal(latest.sequence, 0n); + + listener.sendFrame(connectionId, testFrame({ producerEpoch: 7n, sequence: 999n }), CULTMESH_QUIC_STREAM_LATEST_ONLY); + const staleTimeout = new AbortController(); + const timer = setTimeout(() => staleTimeout.abort(), 300); + await assert.rejects(transport.receiveFrame(staleTimeout.signal)); + clearTimeout(timer); + } finally { + transport.dispose(); + } + } finally { + await listener.close(); + } +}); + +// --------------------------------------------------------------------------- +// Trust negatives, through the session manager with a static lookup source. +// --------------------------------------------------------------------------- + +function base64UrlToBase64(value: string): string { + let base64 = value.replace(/-/g, "+").replace(/_/g, "/"); + while (base64.length % 4 !== 0) base64 += "="; + return base64; +} + +function generateOdinRoot(): { privateKey: KeyObject; publicKey: CultMeshP256PublicKey } { + const { publicKey, privateKey } = generateKeyPairSync("ec", { namedCurve: "prime256v1" }); + const jwk = publicKey.export({ format: "jwk" }) as { x: string; y: string }; + return { + privateKey, + publicKey: { + keyId: "odin-root-1", + x: base64UrlToBase64(jwk.x), + y: base64UrlToBase64(jwk.y), + }, + }; +} + +interface SignedRouteOptions { + readonly verseId: string; + readonly authorityRuntimeId: string; + readonly endpoint: string; + readonly protocolIds: readonly string[]; + readonly privateKey: KeyObject; + readonly odinKeyId: string; + readonly priority?: number; + readonly generation?: string; + readonly issuedAtUnixMilliseconds?: number; + readonly expiresAtUnixMilliseconds?: number; + readonly providerKey?: CultMeshP256PublicKey; + readonly corruptSignature?: boolean; +} + +function signRoute(options: SignedRouteOptions): CultMeshAuthorityRouteMessage { + const priority = options.priority ?? 0; + const generation = options.generation ?? ""; + const issuedAt = options.issuedAtUnixMilliseconds ?? Date.now() - 60_000; + const expiresAt = options.expiresAtUnixMilliseconds ?? Date.now() + 3_600_000; + const providerKey = options.providerKey ?? { keyId: "provider-1", x: "unused", y: "unused" }; + const view: CultMeshAuthorityRouteView = { + verseId: options.verseId, + authorityRuntimeId: options.authorityRuntimeId, + endpoint: options.endpoint, + protocolIds: options.protocolIds, + priority, + generation, + certificate: { + providerKey, + odinKeyId: options.odinKeyId, + issuedAtUnixMilliseconds: issuedAt, + expiresAtUnixMilliseconds: expiresAt, + signature: "", + }, + }; + const payload = canonicalRoute(view); + const signatureBytes = options.corruptSignature + ? Buffer.alloc(64, 0x41) + : (cryptoSign("sha256", Buffer.from(payload), { key: options.privateKey, dsaEncoding: "ieee-p1363" }) as Buffer); + return { + authorityRuntimeId: options.authorityRuntimeId, + endpoint: options.endpoint, + protocolIds: [...options.protocolIds], + priority, + generation, + certificate: { + providerKeyId: providerKey.keyId, + providerPublicKeyX: providerKey.x, + providerPublicKeyY: providerKey.y, + odinKeyId: options.odinKeyId, + issuedAtUnixMilliseconds: issuedAt, + expiresAtUnixMilliseconds: expiresAt, + signature: signatureBytes.toString("base64"), + }, + }; +} + +function verseWithRoute(verseId: string, route: CultMeshAuthorityRouteMessage): CultMeshVerseDescriptorMessage { + return { + verseId, + displayName: verseId, + authorityModel: "operator-cluster", + compatibility: { transportVersion: "cultmesh.v1", rulesHash: "test", compatibleVerseIds: [], requiredPluginIds: [], optionalPluginIds: [] }, + discoveryEndpoints: [route.endpoint], + authorityRuntimeIds: [route.authorityRuntimeId], + authorityRoutes: [route], + }; +} + +const unreachableConnector: CultMeshRealtimeTransportConnector = { + connectorId: "unreachable", + priority: 0, + canConnect: () => true, + connect: async () => { + throw new Error("a trust-refused route must never reach a connector"); + }, +}; + +test("trust negative (a): an expired certificate is refused", async () => { + const target: CultMeshRealtimeTarget = { verseId: "aetheria", authorityRuntimeId: "service:aetheria.daemon" }; + const root = generateOdinRoot(); + const route = signRoute({ + verseId: target.verseId, + authorityRuntimeId: target.authorityRuntimeId, + endpoint: "cultmesh-state+quic://127.0.0.1:9443", + protocolIds: [CULTMESH_REALTIME_STATE_PROTOCOL_ID], + privateKey: root.privateKey, + odinKeyId: root.publicKey.keyId, + issuedAtUnixMilliseconds: Date.now() - 120_000, + expiresAtUnixMilliseconds: Date.now() - 60_000, + }); + const trust: CultMeshAuthorityTrustPolicy = { mode: "authenticated-remote", odinRoots: [root.publicKey] }; + const manager = new CultMeshQuicRealtimeSessionManager({ + lookupSource: new CultMeshStaticRealtimeLookupSource([verseWithRoute(target.verseId, route)]), + trust, + connectors: [unreachableConnector], + }); + await assert.rejects(manager.connect(target), /valid|expired/i); +}); + +test("trust negative (b): an unsigned remote route is refused", async () => { + const target: CultMeshRealtimeTarget = { verseId: "aetheria", authorityRuntimeId: "service:aetheria.daemon" }; + const route: CultMeshAuthorityRouteMessage = { + authorityRuntimeId: target.authorityRuntimeId, + endpoint: "cultmesh-state+quic://198.51.100.7:9443", + protocolIds: [CULTMESH_REALTIME_STATE_PROTOCOL_ID], + priority: 0, + generation: "gen-1", + }; + const trust: CultMeshAuthorityTrustPolicy = { mode: "authenticated-remote", odinRoots: [] }; + const manager = new CultMeshQuicRealtimeSessionManager({ + lookupSource: new CultMeshStaticRealtimeLookupSource([verseWithRoute(target.verseId, route)]), + trust, + connectors: [unreachableConnector], + }); + await assert.rejects(manager.connect(target), /Odin-signed/i); +}); + +test("trust negative (c): a route signed by an untrusted Odin root is refused", async () => { + const target: CultMeshRealtimeTarget = { verseId: "aetheria", authorityRuntimeId: "service:aetheria.daemon" }; + const signer = generateOdinRoot(); + const trustedRoot = generateOdinRoot(); + const route = signRoute({ + verseId: target.verseId, + authorityRuntimeId: target.authorityRuntimeId, + endpoint: "cultmesh-state+quic://127.0.0.1:9443", + protocolIds: [CULTMESH_REALTIME_STATE_PROTOCOL_ID], + privateKey: signer.privateKey, + odinKeyId: signer.publicKey.keyId, + }); + const trust: CultMeshAuthorityTrustPolicy = { mode: "authenticated-remote", odinRoots: [trustedRoot.publicKey] }; + const manager = new CultMeshQuicRealtimeSessionManager({ + lookupSource: new CultMeshStaticRealtimeLookupSource([verseWithRoute(target.verseId, route)]), + trust, + connectors: [unreachableConnector], + }); + await assert.rejects(manager.connect(target), /not trusted/i); +}); + +test("trust negative (d): a route that verifies but pins the wrong certificate is refused at the transport", async (t) => { + if (!nativeBridgeAvailable()) return void t.skip("no native bridge available"); + const listener = await RawQuicListener.open(); + try { + const target: CultMeshRealtimeTarget = { verseId: "aetheria", authorityRuntimeId: "service:aetheria.daemon" }; + const root = generateOdinRoot(); + const route = signRoute({ + verseId: target.verseId, + authorityRuntimeId: target.authorityRuntimeId, + endpoint: `cultmesh-state+quic://127.0.0.1:${listener.boundPort}?cert-sha256=${"0".repeat(64)}`, + protocolIds: [CULTMESH_REALTIME_STATE_PROTOCOL_ID], + privateKey: root.privateKey, + odinKeyId: root.publicKey.keyId, + }); + const trust: CultMeshAuthorityTrustPolicy = { mode: "authenticated-remote", odinRoots: [root.publicKey] }; + const manager = new CultMeshQuicRealtimeSessionManager({ + lookupSource: new CultMeshStaticRealtimeLookupSource([verseWithRoute(target.verseId, route)]), + trust, + connectors: [new CultMeshQuicRealtimeConnector()], + }); + await assert.rejects(manager.connect(target), /certificate|rejected/i); + } finally { + await listener.close(); + } +}); + +test("trust negative (e): a transport that fails isVerifiedFor is disposed and rejected", async () => { + const target: CultMeshRealtimeTarget = { verseId: "aetheria", authorityRuntimeId: "service:aetheria.daemon" }; + const root = generateOdinRoot(); + const route = signRoute({ + verseId: target.verseId, + authorityRuntimeId: target.authorityRuntimeId, + endpoint: "cultmesh-state+quic://127.0.0.1:9443", + protocolIds: [CULTMESH_REALTIME_STATE_PROTOCOL_ID], + privateKey: root.privateKey, + odinKeyId: root.publicKey.keyId, + generation: "route-generation-1", + }); + const trust: CultMeshAuthorityTrustPolicy = { mode: "authenticated-remote", odinRoots: [root.publicKey] }; + + let disposed = false; + const staleTransport: CultMeshRealtimeTransport = { + transportId: "msquic-realtime", + endpoint: route.endpoint, + async sendFrame() {}, + async receiveFrame() { + return testFrame(); + }, + dispose() { + disposed = true; + }, + // A replayed/stale generation: the transport proves a different route + // generation than the one the session manager just verified. + isVerifiedFor: () => false, + }; + const staleConnector: CultMeshRealtimeTransportConnector = { + connectorId: "stale", + priority: 0, + canConnect: () => true, + connect: async () => staleTransport, + }; + const manager = new CultMeshQuicRealtimeSessionManager({ + lookupSource: new CultMeshStaticRealtimeLookupSource([verseWithRoute(target.verseId, route)]), + trust, + connectors: [staleConnector], + }); + + await assert.rejects(manager.connect(target), /did not prove|No realtime state path/i); + assert.equal(disposed, true, "the session manager must dispose a transport that fails isVerifiedFor"); +}); diff --git a/packages/cultmesh-ts/test/realtime-quic-native.test.ts b/packages/cultmesh-ts/test/realtime-quic-native.test.ts new file mode 100644 index 00000000..660a83d9 --- /dev/null +++ b/packages/cultmesh-ts/test/realtime-quic-native.test.ts @@ -0,0 +1,148 @@ +// The Cut 3 FFI smoke, now run against `CultMeshQuicNativeRuntime` instead of +// raw koffi calls: open a runtime, open a listener with a real credential, +// connect a real loopback connection, exchange one frame's worth of raw +// bytes, and close everything down. This proves the binding's declared ABI +// matches the shipped bridge; frame semantics (delivery, generation) are +// `realtime-quic-consumer.test.ts`'s job. +// +// Skips instead of failing when no native bridge is available: set +// CULTMESH_QUIC_NATIVE_DIR to a directory `scripts/build-quic-native.ps1` or +// `scripts/build-quic-native.sh` built, or run from the committed +// `packages/cultmesh-ts/native/` tree once a later cut lands it. + +import assert from "node:assert/strict"; +import { readFileSync } from "node:fs"; +import { join } from "node:path"; +import test from "node:test"; + +import { + CultMeshQuicNativeRuntime, + CULTMESH_QUIC_EVENT_LISTENER_NEW_CONNECTION, + CULTMESH_QUIC_EVENT_CONNECTION_CERTIFICATE_RECEIVED, + CULTMESH_QUIC_EVENT_CONNECTION_CONNECTED, + CULTMESH_QUIC_EVENT_STREAM_STARTED, + CULTMESH_QUIC_EVENT_STREAM_FRAME, + CULTMESH_QUIC_EVENT_STREAM_SEND_COMPLETE, + CULTMESH_QUIC_STREAM_RELIABLE, +} from "../src/realtime-quic-native"; + +const FIXTURE_P12 = join(__dirname, "fixtures", "quic-test.p12"); + +function nativeBridgeAvailable(): boolean { + const dir = process.env.CULTMESH_QUIC_NATIVE_DIR; + if (!dir) return false; + try { + const bridge = process.platform === "win32" ? "gamecult_mesh_quic_native.dll" : "libgamecult_mesh_quic_native.so"; + const dependency = process.platform === "win32" ? "msquic.dll" : "libmsquic.so.2"; + readFileSync(join(dir, bridge)); + readFileSync(join(dir, dependency)); + return true; + } catch { + return false; + } +} + +async function waitFor(collect: (resolve: (value: T) => void) => void, timeoutMs = 5_000): Promise { + return await new Promise((resolve, reject) => { + const timer = setTimeout(() => reject(new Error("timed out waiting for a native QUIC event")), timeoutMs); + collect((value) => { + clearTimeout(timer); + resolve(value); + }); + }); +} + +test("CultMeshQuicNativeRuntime opens and closes with no listener or connection", async (t) => { + if (!nativeBridgeAvailable()) { + t.skip("CULTMESH_QUIC_NATIVE_DIR is not set to a built native bridge."); + return; + } + const runtime = await CultMeshQuicNativeRuntime.open(); + await runtime.release(); +}); + +test("CultMeshQuicNativeRuntime opens and closes a listener with a real PKCS12 credential", async (t) => { + if (!nativeBridgeAvailable()) { + t.skip("CULTMESH_QUIC_NATIVE_DIR is not set to a built native bridge."); + return; + } + const runtime = await CultMeshQuicNativeRuntime.open(); + try { + const pkcs12 = readFileSync(FIXTURE_P12); + const { listenerId, boundPort } = runtime.listenerOpen("127.0.0.1", 0, pkcs12, ""); + assert.ok(boundPort > 0); + runtime.listenerClose(listenerId); + } finally { + await runtime.release(); + } +}); + +test("a full loopback connection delivers one raw frame end to end", async (t) => { + if (!nativeBridgeAvailable()) { + t.skip("CULTMESH_QUIC_NATIVE_DIR is not set to a built native bridge."); + return; + } + const runtime = await CultMeshQuicNativeRuntime.open(); + try { + const pkcs12 = readFileSync(FIXTURE_P12); + const { listenerId, boundPort } = runtime.listenerOpen("127.0.0.1", 0, pkcs12, ""); + + const accepted = waitFor((resolve) => { + runtime.onListenerEvent(listenerId, (event) => { + if (event.type === CULTMESH_QUIC_EVENT_LISTENER_NEW_CONNECTION) resolve(event.connectionId); + }); + }); + + const clientConnectionId = runtime.connectionOpen("127.0.0.1", boundPort); + const clientConnected = waitFor((resolve) => { + runtime.onConnectionEvent(clientConnectionId, (event) => { + if (event.type === CULTMESH_QUIC_EVENT_CONNECTION_CERTIFICATE_RECEIVED) { + runtime.connectionCertificateComplete(clientConnectionId, true); + return; + } + if (event.type === CULTMESH_QUIC_EVENT_CONNECTION_CONNECTED) resolve(); + }); + }); + + const serverConnectionId = await accepted; + const serverFrame = waitFor((resolve) => { + runtime.onConnectionEvent(serverConnectionId, (event) => { + if (event.type === CULTMESH_QUIC_EVENT_STREAM_STARTED) { + assert.equal(event.streamKind, CULTMESH_QUIC_STREAM_RELIABLE); + return; + } + if (event.type === CULTMESH_QUIC_EVENT_STREAM_FRAME) resolve(event.payload); + }); + }); + + await clientConnected; + const streamId = runtime.streamOpen(clientConnectionId, CULTMESH_QUIC_STREAM_RELIABLE); + const sendComplete = waitFor((resolve) => { + runtime.onConnectionEvent(clientConnectionId, (event) => { + if (event.type === CULTMESH_QUIC_EVENT_STREAM_SEND_COMPLETE) resolve(); + }); + }); + runtime.streamSendFrame(streamId, new Uint8Array([1, 2, 3, 4]), true); + await sendComplete; + + const payload = await serverFrame; + assert.deepEqual(Array.from(payload), [1, 2, 3, 4]); + + runtime.connectionShutdown(clientConnectionId, 0n); + runtime.connectionShutdown(serverConnectionId, 0n); + runtime.listenerClose(listenerId); + } finally { + await runtime.release(); + } +}); + +test("negative grep: no koffi callbacks, .async only on nextEvent", () => { + const source = readFileSync(join(__dirname, "..", "src", "realtime-quic-native.ts"), "utf8"); + const asyncUses = [...source.matchAll(/\.async\(/g)]; + assert.ok(asyncUses.length > 0, "expected at least one .async( use for nextEvent"); + for (const match of asyncUses) { + const before = source.slice(Math.max(0, match.index! - 40), match.index!); + assert.match(before, /nextEvent/, "the only .async( call site must be nextEvent"); + } + assert.doesNotMatch(source, /koffi\.register/, "no koffi callbacks"); +});