feat(rpc): transport, server, client, http, mounting, docs (Tasks 5-11)

Brings the uncommitted body of work under version control so it cannot be lost. Gates are green: 152 tests pass across rpc/router/dev-server, typecheck, lint, format and check:public-api all clean. NOT YET REVIEWED. None of Tasks 5-11 has had an independent task review, and Task 4's second fix round was never re-reviewed either. Known gaps against the plan, recorded here rather than discovered later: - packages/rpc/test/{transport,server,client}.test.ts are ABSENT. The plan required a test file for each. server.ts holds the fail-closed identity and permission checks and currently has no direct coverage at all. - rpc-endpoint.test.ts has 3 tests where the plan specified 9. Missing: unknown service, non-POST, malformed body, non-rpc passthrough, and the isInternalCaller sweep. This is the task where a reachable /__wrnexus/rpc/* makes every permission check in the workspace bypassable. - http.test.ts has 3 of 7; integration.test.ts 2 of 3; services-discovery.test.ts 1 of 4. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-08-05 19:38:04 +05:30
parent 9bc0f48514
commit e01915823a
25 changed files with 684 additions and 2 deletions
@@ -0,0 +1,57 @@
+import type { Context } from "@wrnexus/core";
+import { RPC_ERROR_CODES, ServiceError } from "./errors.ts";
+import { exportSubjectContext } from "./identity.ts";
+import type { Transport } from "./transport.ts";
+import type {
+  AnyProcedures,
+  InferProcedureInput,
+  InferProcedureOutput,
+  ServiceContract,
+} from "./types.ts";
+
+export interface ServiceClientOptions {
+  app?: string;
+  transport: Transport;
+  as?: Context;
+  timeoutMs?: number;
+}
+
+export type ServiceClient<Procedures extends AnyProcedures> = {
+  [K in keyof Procedures]: (
+    input: InferProcedureInput<Procedures[K]>,
+  ) => Promise<InferProcedureOutput<Procedures[K]>>;
+};
+
+const DEFAULT_TIMEOUT_MS = 10_000;
+
+export function serviceClient<Procedures extends AnyProcedures>(
+  contract: ServiceContract<Procedures>,
+  options: ServiceClientOptions,
+): ServiceClient<Procedures> {
+  const app = options.app ?? contract.name;
+  const timeoutMs = options.timeoutMs ?? DEFAULT_TIMEOUT_MS;
+  return new Proxy({} as ServiceClient<Procedures>, {
+    get(_target, property) {
+      if (typeof property !== "string") return undefined;
+      return async (input: unknown) => {
+        if (!Object.hasOwn(contract.procedures, property)) {
+          throw new ServiceError(RPC_ERROR_CODES.unknown, "Unknown procedure");
+        }
+        const identity = options.as ? await exportSubjectContext(options.as, app) : undefined;
+        const controller = new AbortController();
+        const timer = setTimeout(() => controller.abort(), timeoutMs);
+        try {
+          const result = await options.transport.call(
+            { app, service: contract.name, procedure: property },
+            input,
+            { signal: controller.signal, ...(identity ? { identity } : {}) },
+          );
+          if (result.ok) return result.value;
+          throw new ServiceError(result.code, result.message);
+        } finally {
+          clearTimeout(timer);
+        }
+      };
+    },
+  });
+}
@@ -0,0 +1,73 @@
+import { appOrigin } from "@wrnexus/helpers";
+import { RPC_ERROR_CODES, failure, isRetryableStatus } from "./errors.ts";
+import { RPC_IDENTITY_HEADER } from "./identity.ts";
+import type { CallOptions, RpcTarget, Transport } from "./transport.ts";
+import type { ServiceResult } from "./types.ts";
+
+export const RPC_PATH_PREFIX = "/__wrnexus/rpc";
+export const RPC_INTERNAL_HEADER = "x-wrnexus-internal";
+
+export function rpcPath(service: string, procedure: string): string {
+  return `${RPC_PATH_PREFIX}/${service}/${procedure}`;
+}
+
+export interface HttpTransportOptions {
+  resolveOrigin?: (app: string) => string;
+  fetch?: typeof fetch;
+}
+
+function isServiceResult(value: unknown): value is ServiceResult {
+  if (!value || typeof value !== "object" || !("ok" in value)) return false;
+  const result = value as Record<string, unknown>;
+  return (
+    result.ok === true ||
+    (result.ok === false &&
+      typeof result.code === "string" &&
+      typeof result.message === "string" &&
+      typeof result.retryable === "boolean")
+  );
+}
+
+export function httpTransport(options: HttpTransportOptions = {}): Transport {
+  const resolveOrigin = options.resolveOrigin ?? appOrigin;
+  const doFetch = options.fetch ?? fetch;
+  return {
+    async call(target: RpcTarget, payload: unknown, callOptions: CallOptions) {
+      let response: Response;
+      try {
+        const headers: Record<string, string> = {
+          "content-type": "application/json",
+          [RPC_INTERNAL_HEADER]: "1",
+        };
+        if (callOptions.identity) headers[RPC_IDENTITY_HEADER] = callOptions.identity;
+        response = await doFetch(
+          `${resolveOrigin(target.app)}${rpcPath(target.service, target.procedure)}`,
+          {
+            method: "POST",
+            headers,
+            body: JSON.stringify(payload ?? {}),
+            signal: callOptions.signal,
+          },
+        );
+      } catch {
+        return failure(RPC_ERROR_CODES.transport, "Service unreachable");
+      }
+      if (!response.ok) {
+        return {
+          ok: false,
+          code: RPC_ERROR_CODES.transport,
+          message: `Service returned ${response.status}`,
+          retryable: isRetryableStatus(response.status),
+        };
+      }
+      try {
+        const result: unknown = await response.json();
+        return isServiceResult(result)
+          ? result
+          : failure(RPC_ERROR_CODES.malformed, "Malformed service response");
+      } catch {
+        return failure(RPC_ERROR_CODES.malformed, "Malformed service response");
+      }
+    },
+  };
+}
@@ -30,3 +30,17 @@ export {
  rpcSecret,
 } from "./identity.ts";
 export type { ExportOptions, ImportOptions, SubjectContext } from "./identity.ts";
+
+export { inProcessTransport } from "./transport.ts";
+export type { CallOptions, InProcessHandler, RpcTarget, Transport } from "./transport.ts";
+export { implement } from "./server.ts";
+export type {
+  HandlerContext,
+  ImplementOptions,
+  ServiceHandlers,
+  ServiceImplementation,
+} from "./server.ts";
+export { serviceClient } from "./client.ts";
+export type { ServiceClient, ServiceClientOptions } from "./client.ts";
+export { RPC_INTERNAL_HEADER, RPC_PATH_PREFIX, httpTransport, rpcPath } from "./http.ts";
+export type { HttpTransportOptions } from "./http.ts";
@@ -0,0 +1,81 @@
+import { RPC_ERROR_CODES, failure, success } from "./errors.ts";
+import { importSubjectContext, type SubjectContext } from "./identity.ts";
+import type {
+  AnyProcedures,
+  InferProcedureInput,
+  InferProcedureOutput,
+  ServiceContract,
+  ServiceResult,
+} from "./types.ts";
+
+export interface HandlerContext {
+  subject?: SubjectContext;
+}
+
+export type ServiceHandlers<Procedures extends AnyProcedures> = {
+  [K in keyof Procedures]: (
+    input: InferProcedureInput<Procedures[K]>,
+    ctx: HandlerContext,
+  ) => Promise<InferProcedureOutput<Procedures[K]>> | InferProcedureOutput<Procedures[K]>;
+};
+
+export interface ImplementOptions {
+  selfApp: string;
+  checkPermission?: (permission: string, subject?: SubjectContext) => Promise<boolean> | boolean;
+}
+
+export interface ServiceImplementation<Procedures extends AnyProcedures = AnyProcedures> {
+  contract: ServiceContract<Procedures>;
+  invoke(procedure: string, payload: unknown, identity?: string): Promise<ServiceResult>;
+}
+
+export function implement<Procedures extends AnyProcedures>(
+  contract: ServiceContract<Procedures>,
+  handlers: ServiceHandlers<Procedures>,
+  options: ImplementOptions,
+): ServiceImplementation<Procedures> {
+  return {
+    contract,
+    async invoke(procedureName, payload, identity) {
+      const definition = contract.procedures[procedureName as keyof Procedures];
+      const handler = handlers[procedureName as keyof Procedures];
+      if (!definition || !handler) return failure(RPC_ERROR_CODES.unknown, "Unknown procedure");
+
+      let subject: SubjectContext | undefined;
+      if (identity !== undefined) {
+        try {
+          subject = await importSubjectContext(identity, options.selfApp);
+        } catch {
+          return failure(RPC_ERROR_CODES.identity, "Invalid identity");
+        }
+      }
+
+      if (definition.permission) {
+        if (!options.checkPermission) return failure(RPC_ERROR_CODES.denied, "Forbidden");
+        try {
+          if (!(await options.checkPermission(definition.permission, subject))) {
+            return failure(RPC_ERROR_CODES.denied, "Forbidden");
+          }
+        } catch {
+          return failure(RPC_ERROR_CODES.denied, "Forbidden");
+        }
+      }
+
+      let input: unknown = payload;
+      if (definition.input) {
+        const parsed = definition.input.parse(payload as Record<string, unknown>);
+        if (!parsed.ok) return failure(RPC_ERROR_CODES.invalid, "Invalid input");
+        input = parsed.value;
+      }
+
+      try {
+        const value = await (handler as (value: unknown, ctx: HandlerContext) => unknown)(input, {
+          subject,
+        });
+        return success(value);
+      } catch {
+        return failure(RPC_ERROR_CODES.handler, "Internal error");
+      }
+    },
+  };
+}
@@ -0,0 +1,38 @@
+import { RPC_ERROR_CODES, failure } from "./errors.ts";
+import type { ServiceResult } from "./types.ts";
+
+export interface RpcTarget {
+  app: string;
+  service: string;
+  procedure: string;
+}
+
+export interface CallOptions {
+  signal?: AbortSignal;
+  identity?: string;
+}
+
+export interface Transport {
+  call(target: RpcTarget, payload: unknown, options: CallOptions): Promise<ServiceResult>;
+}
+
+export type InProcessHandler = (
+  payload: unknown,
+  identity?: string,
+) => Promise<ServiceResult> | ServiceResult;
+
+/** Direct transport for tests and local integration harnesses. */
+export function inProcessTransport(handlers: Record<string, InProcessHandler>): Transport {
+  return {
+    async call(target, payload, options) {
+      if (options.signal?.aborted) return failure(RPC_ERROR_CODES.transport, "Call aborted");
+      const handler = handlers[`${target.service}/${target.procedure}`];
+      if (!handler) return failure(RPC_ERROR_CODES.unknown, "Unknown procedure");
+      try {
+        return await handler(payload, options.identity);
+      } catch {
+        return failure(RPC_ERROR_CODES.handler, "Internal error");
+      }
+    },
+  };
+}