WRNexusJS/packages/rpc/src/client.ts

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) {
      // Every declared procedure is a string key; anything else (including
      // `then`/`catch`/`finally`) is not one of ours. Returning a function for
      // those makes `await client` or `return client` from an async function
      // read the proxy as thenable — the runtime then calls `then(resolve,
      // reject)`, which throws "Unknown procedure". Returning undefined lets
      // the caller be treated as a plain (non-thenable) object instead.
      if (typeof property !== "string" || !Object.hasOwn(contract.procedures, property)) {
        return undefined;
      }
      return async (input: unknown) => {
        let identity: string | undefined;
        if (options.as) {
          try {
            identity = await exportSubjectContext(options.as, app);
          } catch (error) {
            // A missing/misconfigured WRNEXUS_RPC_SECRET otherwise rejects with
            // a bare Error, so a caller matching on ServiceError treats
            // misconfiguration as a crash instead of a handled RPC failure.
            // The operator-facing message names no secret value, so it is safe
            // to preserve.
            const message = error instanceof Error ? error.message : String(error);
            throw new ServiceError(RPC_ERROR_CODES.identity, message);
          }
        }
        const controller = new AbortController();
        const timer = setTimeout(() => controller.abort(), timeoutMs);
        const timeout = new Promise<never>((_, reject) => {
          controller.signal.addEventListener("abort", () => {
            reject(new ServiceError(RPC_ERROR_CODES.transport, "Call timed out"));
          });
        });
        const callPromise = options.transport.call(
          { app, service: contract.name, procedure: property },
          input,
          {
            signal: controller.signal,
            idempotent: contract.procedures[property as keyof Procedures].idempotent === true,
            ...(identity ? { identity } : {}),
          },
        );
        try {
          const result = await Promise.race([callPromise, timeout]);
          if (result.ok) return result.value;
          throw new ServiceError(result.code, result.message, result.retryable);
        } finally {
          clearTimeout(timer);
          // If the timeout won the race, the transport call may still settle
          // later — a transport that ignores the abort signal keeps running.
          // Nothing awaits it again, so swallow a late rejection here rather
          // than let it surface as an unhandled promise rejection.
          callPromise.catch(() => {});
        }
      };
    },
  });
}