tech-log-frontend/src/adapters/browser-rpc/transport.ts

import type {
  BrowserRpcKind,
  BrowserRpcOperationV3,
  BrowserRpcProtocol,
  BrowserRpcProviderProfile,
  BrowserRpcRuntimeBindingIdentity,
  BrowserRpcTransportFailureCode,
} from "../../contracts/browser-rpc.ts";

export type BrowserRpcTransportFailure = Readonly<{
  code: BrowserRpcTransportFailureCode;
  retryAfterMs?: number;
}>;

export type BrowserRpcTransportCall = Readonly<{
  operation: BrowserRpcOperationV3;
  profile: BrowserRpcProviderProfile;
  request: unknown;
  encodedRequestBytes: number;
  attempt: number;
  timeoutMs: number;
  signal: AbortSignal;
  idempotencyKey?: string;
}>;

export type BrowserRpcUnaryTransportResult =
  | Readonly<{
      ok: true;
      message: unknown;
      encodedBytes: number;
    }>
  | Readonly<{
      ok: false;
      failure: BrowserRpcTransportFailure;
    }>;

export type BrowserRpcStreamFrame =
  | Readonly<{
      kind: "MESSAGE";
      message: unknown;
      encodedBytes: number;
    }>
  | Readonly<{
      kind: "TERMINAL";
      ok: true;
    }>
  | Readonly<{
      kind: "TERMINAL";
      ok: false;
      failure: BrowserRpcTransportFailure;
    }>;

/**
 * RPC-RR-01. A server stream is a physical resource, not just a sequence.
 *
 * A bare `AsyncIterable` gives the runtime no way to cancel the underlying
 * stream or to learn when it actually closed: `iterator.return()` is a request
 * a non-cooperative implementation may ignore. The runtime could then time out,
 * report the call finished, and admit a second stream for the same operation
 * while the first was still running against the server.
 *
 * The lease separates the three concerns the runtime needs:
 *
 * - `frames` is the sequence,
 * - `cancel(reason)` is a synchronous request to stop the physical stream,
 * - `waitClosed()` settles only once that stream is really closed,
 * - `streamId` names the physical stream so two leases are never confused.
 */
export type BrowserRpcServerStreamLease = Readonly<{
  streamId: string;
  frames: AsyncIterable<BrowserRpcStreamFrame>;
  cancel(reason: string): void;
  waitClosed(): Promise<void>;
}>;

export type BrowserRpcTransport = BrowserRpcRuntimeBindingIdentity &
  Readonly<{
    invokeUnary?(
      call: BrowserRpcTransportCall,
    ): Promise<BrowserRpcUnaryTransportResult>;
    openServerStream?(
      call: BrowserRpcTransportCall,
    ): BrowserRpcServerStreamLease;
  }>;

const STREAM_ID = /^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$/u;

/**
 * RPC-RR-01. Decodes a lease from own data descriptors before the runtime
 * registers it, so an accessor cannot hand the registry one object and the
 * cancellation path another.
 */
export function decodeServerStreamLease(
  value: unknown,
): BrowserRpcServerStreamLease | null {
  if (value === null || typeof value !== "object") return null;
  let streamId: unknown;
  let frames: unknown;
  let cancel: unknown;
  let waitClosed: unknown;
  try {
    if (Object.getOwnPropertySymbols(value).length > 0) return null;
    const names = Object.getOwnPropertyNames(value).sort();
    const expected = ["cancel", "frames", "streamId", "waitClosed"];
    if (
      names.length !== expected.length ||
      names.some((name, index) => name !== expected[index])
    ) {
      return null;
    }
    for (const name of names) {
      const descriptor = Object.getOwnPropertyDescriptor(value, name);
      if (!descriptor || !("value" in descriptor)) return null;
    }
    streamId = Object.getOwnPropertyDescriptor(value, "streamId")?.value;
    frames = Object.getOwnPropertyDescriptor(value, "frames")?.value;
    cancel = Object.getOwnPropertyDescriptor(value, "cancel")?.value;
    waitClosed = Object.getOwnPropertyDescriptor(value, "waitClosed")?.value;
  } catch {
    return null;
  }
  // RPC-02. The async-iterator lookup is a read of foreign state like any
  // other, so it happens inside the decoder's own boundary. Performing it after
  // the `try` let a throwing `Symbol.asyncIterator` getter escape this
  // function as a native `TypeError`, breaking the decoder's totality.
  let openFrames: unknown;
  try {
    if (
      typeof streamId !== "string" ||
      !STREAM_ID.test(streamId) ||
      frames === null ||
      typeof frames !== "object" ||
      typeof cancel !== "function" ||
      typeof waitClosed !== "function"
    ) {
      return null;
    }
    openFrames = (frames as AsyncIterable<unknown>)[Symbol.asyncIterator];
    if (typeof openFrames !== "function") return null;
  } catch {
    return null;
  }
  const iterate = (openFrames as () => AsyncIterator<BrowserRpcStreamFrame>)
    .bind(frames);
  return Object.freeze({
    streamId,
    frames: Object.freeze({
      [Symbol.asyncIterator]: iterate,
    }) as AsyncIterable<BrowserRpcStreamFrame>,
    cancel: (cancel as (reason: string) => void).bind(value),
    waitClosed: (waitClosed as () => Promise<void>).bind(value),
  });
}

export function defineBrowserRpcTransport(
  transport: BrowserRpcTransport,
): BrowserRpcTransport {
  if (
    !transport.runtimeProfileId ||
    !transport.providerId ||
    !isProtocol(transport.protocol) ||
    !isRpcKind(transport.rpcKind) ||
    (transport.rpcKind === "UNARY" &&
      (typeof transport.invokeUnary !== "function" ||
        transport.openServerStream !== undefined)) ||
    (transport.rpcKind === "SERVER_STREAM" &&
      (typeof transport.openServerStream !== "function" ||
        transport.invokeUnary !== undefined))
  ) {
    throw new TypeError("Browser RPC transport is invalid.");
  }
  return Object.freeze({ ...transport });
}

function isProtocol(value: string): value is BrowserRpcProtocol {
  return value === "CONNECT_HTTP" || value === "GRPC_WEB";
}

function isRpcKind(value: string): value is BrowserRpcKind {
  return value === "UNARY" || value === "SERVER_STREAM";
}