import { expect, test } from "bun:test"; import { defineService, procedure } from "../src/contract.ts"; import { implementStream, streamClient } from "../src/stream.ts"; import { handleRpcRequest } from "../../dev-server/src/rpc-dispatch.ts"; const numbers = defineService({ name: "numbers", procedures: { count: procedure.output().build() }, }); test("private streaming RPC authenticates identity and yields SSE frames as an async iterable", async () => { const oldSecret = process.env.WRNEXUS_RPC_SECRET; const oldApp = process.env.WRNEXUS_APP_NAME; const oldOrigins = process.env.WRNEXUS_INTERNAL_ORIGINS; try { process.env.WRNEXUS_RPC_SECRET = "test-rpc-secret-at-least-32-chars-long"; process.env.WRNEXUS_APP_NAME = "web"; process.env.WRNEXUS_INTERNAL_ORIGINS = JSON.stringify({ numbers: "http://numbers.internal" }); const service = implementStream( numbers, { async *count(_input, context) { expect(context.subject?.subjectId).toBe("u1"); yield 1; yield 2; }, }, { selfApp: "numbers" }, ); const client = streamClient(numbers, { app: "numbers", as: { user: { id: "u1" }, locals: {} } as never, fetch: (async (input: RequestInfo | URL, init?: RequestInit) => { const request = new Request(input, init); return (await handleRpcRequest( request, new URL(request.url), new Map([["numbers", service]]), ))!; }) as typeof fetch, }); const values: number[] = []; for await (const value of client.count(undefined)) values.push(value); expect(values).toEqual([1, 2]); expect(service.metrics.snapshot()).toEqual({ started: 1, completed: 1, failed: 0, active: 0 }); } finally { if (oldSecret === undefined) delete process.env.WRNEXUS_RPC_SECRET; else process.env.WRNEXUS_RPC_SECRET = oldSecret; if (oldApp === undefined) delete process.env.WRNEXUS_APP_NAME; else process.env.WRNEXUS_APP_NAME = oldApp; if (oldOrigins === undefined) delete process.env.WRNEXUS_INTERNAL_ORIGINS; else process.env.WRNEXUS_INTERNAL_ORIGINS = oldOrigins; } });