docs: correct the dead-output component count from 11 to 9

Counted from source: the 22 remaining outputs sit in 9 components, not 11.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

packages/rpc/src/index.ts

@@ -56,3 +56,11 @@ export {
   rpcPath,
 } from "./http.ts";
 export type { HttpTransportOptions } from "./http.ts";
+export { implementStream, rpcStreamPath, RPC_STREAM_PATH_PREFIX, streamClient } from "./stream.ts";
+export type {
+  StreamClient,
+  StreamClientOptions,
+  StreamHandlers,
+  StreamImplementOptions,
+  StreamImplementation,
+} from "./stream.ts";

packages/rpc/src/stream.ts

@@ -0,0 +1,170 @@
+import type { Context } from "@wrnexus/core";
+import { RPC_ERROR_CODES, ServiceError } from "./errors.ts";
+import { exportSubjectContext, importSubjectContext, type SubjectContext } from "./identity.ts";
+import { resolveAppOrigin, RPC_INTERNAL_HEADER } from "./http.ts";
+import type {
+  AnyProcedures,
+  InferProcedureInput,
+  InferProcedureOutput,
+  ServiceContract,
+} from "./types.ts";
+
+export const RPC_STREAM_PATH_PREFIX = "/__wrnexus/rpc-stream";
+
+export function rpcStreamPath(service: string, procedure: string): string {
+  return `${RPC_STREAM_PATH_PREFIX}/${service}/${procedure}`;
+}
+
+export interface StreamImplementation<Procedures extends AnyProcedures = AnyProcedures> {
+  contract: ServiceContract<Procedures>;
+  stream(procedure: string, payload: unknown, identity?: string): AsyncIterable<unknown>;
+}
+
+export type StreamHandlers<Procedures extends AnyProcedures> = {
+  [K in keyof Procedures]: (
+    input: InferProcedureInput<Procedures[K]>,
+    ctx: { subject?: import("./identity.ts").SubjectContext },
+  ) => AsyncIterable<InferProcedureOutput<Procedures[K]>>;
+};
+
+export interface StreamImplementOptions {
+  selfApp: string;
+  checkPermission?: (permission: string, subject?: SubjectContext) => Promise<boolean> | boolean;
+}
+
+/** Define an authenticated, validated stream endpoint. */
+export function implementStream<Procedures extends AnyProcedures>(
+  contract: ServiceContract<Procedures>,
+  handlers: StreamHandlers<Procedures>,
+  options: StreamImplementOptions,
+): StreamImplementation<Procedures> {
+  return {
+    contract,
+    stream(procedure, payload, identity) {
+      const handler = Object.hasOwn(handlers, procedure)
+        ? handlers[procedure as keyof Procedures]
+        : undefined;
+      if (!handler) throw new ServiceError(RPC_ERROR_CODES.unknown, "Unknown procedure");
+      return (async function* () {
+        const definition = contract.procedures[procedure as keyof Procedures];
+        let subject: SubjectContext | undefined;
+        if (identity) {
+          try {
+            subject = await importSubjectContext(identity, options.selfApp);
+          } catch {
+            throw new ServiceError(RPC_ERROR_CODES.identity, "Invalid identity");
+          }
+        }
+        if (definition.permission) {
+          if (
+            !options.checkPermission ||
+            !(await options.checkPermission(definition.permission, subject))
+          ) {
+            throw new ServiceError(RPC_ERROR_CODES.denied, "Forbidden");
+          }
+        }
+        let input = payload;
+        if (definition.input) {
+          let parsed: { ok: boolean; value?: unknown };
+          try {
+            parsed = definition.input.parse(payload as Record<string, unknown>);
+          } catch {
+            throw new ServiceError(RPC_ERROR_CODES.invalid, "Invalid input");
+          }
+          if (!parsed.ok) throw new ServiceError(RPC_ERROR_CODES.invalid, "Invalid input");
+          input = parsed.value;
+        }
+        yield* (
+          handler as (
+            input: unknown,
+            ctx: { subject?: import("./identity.ts").SubjectContext },
+          ) => AsyncIterable<unknown>
+        )(input, { subject });
+      })();
+    },
+  };
+}
+
+export interface StreamClientOptions {
+  app?: string;
+  as?: Context;
+  fetch?: typeof fetch;
+  signal?: AbortSignal;
+}
+export type StreamClient<Procedures extends AnyProcedures> = {
+  [K in keyof Procedures]: (
+    input: InferProcedureInput<Procedures[K]>,
+  ) => AsyncIterable<InferProcedureOutput<Procedures[K]>>;
+};
+
+function decodeFrames(
+  body: ReadableStream<Uint8Array>,
+  signal?: AbortSignal,
+): AsyncIterable<unknown> {
+  return (async function* () {
+    const reader = body.getReader();
+    const decoder = new TextDecoder();
+    let buffer = "";
+    try {
+      while (!signal?.aborted) {
+        const next = await reader.read();
+        if (next.done) break;
+        buffer += decoder.decode(next.value, { stream: true });
+        let boundary: number;
+        while ((boundary = buffer.indexOf("\n\n")) >= 0) {
+          const frame = buffer.slice(0, boundary);
+          buffer = buffer.slice(boundary + 2);
+          const data = frame
+            .split("\n")
+            .find((line) => line.startsWith("data: "))
+            ?.slice(6);
+          if (!data) continue;
+          const value: unknown = JSON.parse(data);
+          if (!value || typeof value !== "object" || !("ok" in value))
+            throw new ServiceError(RPC_ERROR_CODES.malformed, "Malformed stream response");
+          const result = value as { ok: boolean; value?: unknown; code?: string; message?: string };
+          if (!result.ok)
+            throw new ServiceError(
+              result.code ?? RPC_ERROR_CODES.handler,
+              result.message ?? "Stream failed",
+            );
+          yield result.value;
+        }
+      }
+    } finally {
+      reader.cancel().catch(() => {});
+    }
+  })();
+}
+
+export function streamClient<Procedures extends AnyProcedures>(
+  contract: ServiceContract<Procedures>,
+  options: StreamClientOptions = {},
+): StreamClient<Procedures> {
+  const app = options.app ?? contract.name;
+  const doFetch = options.fetch ?? fetch;
+  return new Proxy({} as StreamClient<Procedures>, {
+    get(_target, property) {
+      if (typeof property !== "string" || !Object.hasOwn(contract.procedures, property))
+        return undefined;
+      return (input: unknown) =>
+        (async function* () {
+          let identity: string | undefined;
+          if (options.as) identity = await exportSubjectContext(options.as, app);
+          const headers: Record<string, string> = {
+            "content-type": "application/json",
+            [RPC_INTERNAL_HEADER]: "1",
+            accept: "text/event-stream",
+          };
+          if (identity) headers["x-wrnexus-rpc-identity"] = identity;
+          const response = await doFetch(
+            `${resolveAppOrigin(app)}${rpcStreamPath(contract.name, property)}`,
+            { method: "POST", headers, body: JSON.stringify(input ?? {}), signal: options.signal },
+          );
+          if (!response.ok || !response.body)
+            throw new ServiceError(RPC_ERROR_CODES.transport, "Service unavailable");
+          yield* decodeFrames(response.body, options.signal);
+        })();
+    },
+  });
+}

packages/rpc/test/stream.test.ts

@@ -0,0 +1,53 @@
+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<number>().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]);
+  } 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;
+  }
+});