tech-log-frontend/tests/unit/browser-rpc/browser-rpc-runtime.test.ts

import { describe, expect, it } from "vitest";

import {
  createBrowserRpcRuntime,
  createUnavailableBrowserRpcTransport,
  defineBrowserRpcTransport,
  type BrowserRpcObservation,
  type BrowserRpcServerStreamLease,
  type BrowserRpcStreamFrame,
  type BrowserRpcTransport,
} from "../../../src/adapters/browser-rpc/index.ts";
import type { Result } from "../../../src/application/result.ts";
import type { AppFailure } from "../../../src/contracts/errors.ts";
import {
  MAPPERS,
  SCHEMA_CODECS,
  STREAM_ENCODER,
  UNARY_ENCODER,
  isResourceView,
  streamOperation,
  streamProfile,
  unaryOperation,
  unaryProfile,
  type ResourceView,
} from "./fixture.ts";

describe("Browser RPC provider-neutral runtime", () => {
  it("validates, encodes, maps and admits only the typed unary result", async () => {
    const observations: BrowserRpcObservation[] = [];
    const transport = defineBrowserRpcTransport({
      runtimeProfileId: "CONNECT_REFERENCE_UNARY",
      providerId: "REFERENCE_RPC",
      protocol: "CONNECT_HTTP",
      rpcKind: "UNARY",
      async invokeUnary(call) {
        expect(call.request).toEqual({ resourceId: "resource-1" });
        expect(call.timeoutMs).toBeGreaterThan(0);
        return Object.freeze({
          ok: true,
          message: Object.freeze({ id: "resource-1", name: "Resource one" }),
          encodedBytes: 48,
        });
      },
    });
    const port = unaryRuntime(transport, {
      observe(value) {
        observations.push(value);
      },
    }).bindUnary("GET_RPC_RESOURCE", isResourceView);

    await expect(port.execute({ resourceId: "resource-1" })).resolves.toEqual({
      ok: true,
      value: { id: "resource-1", label: "Resource one" },
    });
    expect(observations).toEqual([
      {
        operationId: "GET_RPC_RESOURCE",
        protocol: "CONNECT_HTTP",
        runtimeProfileId: "CONNECT_REFERENCE_UNARY",
        rpcKind: "UNARY",
        outcome: "SUCCESS",
        attemptCount: 1,
        messageCount: 1,
      },
    ]);
  });

  it("fails before transport on invalid input or unexpected idempotency metadata", async () => {
    let calls = 0;
    const transport = defineBrowserRpcTransport({
      runtimeProfileId: "CONNECT_REFERENCE_UNARY",
      providerId: "REFERENCE_RPC",
      protocol: "CONNECT_HTTP",
      rpcKind: "UNARY",
      async invokeUnary() {
        calls += 1;
        return {
          ok: false,
          failure: { code: "UNAVAILABLE" },
        };
      },
    });
    const port = unaryRuntime(transport).bindUnary(
      "GET_RPC_RESOURCE",
      isResourceView,
    );

    const invalid = await port.execute({ resourceId: 42 });
    expect(invalid).toMatchObject({
      ok: false,
      error: {
        kind: "VALIDATION_REJECTED",
        code: "RPC_REQUEST_SCHEMA_INVALID",
      },
    });
    const metadata = await port.execute(
      { resourceId: "resource-1" },
      { idempotencyKey: "caller-key-is-not-allowed" },
    );
    expect(metadata).toMatchObject({
      ok: false,
      error: {
        kind: "VALIDATION_REJECTED",
        code: "RPC_IDEMPOTENCY_KEY_INVALID",
      },
    });
    expect(calls).toBe(0);
  });

  it("keeps the frontend retry owner bounded by replay policy and one total deadline", async () => {
    let calls = 0;
    const profile = unaryProfile({
      retryProfileId: "RPC_RETRY_TWO",
      retryOwner: "FRONTEND_ADAPTER",
      maxAttempts: 2,
      backoffMs: [0],
      retryableFailures: ["UNAVAILABLE"],
      maxRetryAfterMs: 100,
    });
    const operation = unaryOperation({
      retryProfileId: "RPC_RETRY_TWO",
    });
    const transport = defineBrowserRpcTransport({
      runtimeProfileId: profile.runtimeProfileId,
      providerId: profile.providerId,
      protocol: profile.protocol,
      rpcKind: profile.rpcKind,
      async invokeUnary() {
        calls += 1;
        if (calls === 1) {
          return {
            ok: false,
            failure: { code: "UNAVAILABLE" },
          };
        }
        return {
          ok: true,
          message: { id: "resource-2", name: "Retried resource" },
          encodedBytes: 32,
        };
      },
    });
    const runtime = createBrowserRpcRuntime({
      operations: { GET_RPC_RESOURCE: operation },
      profiles: { CONNECT_REFERENCE_UNARY: profile },
      schemaCodecs: SCHEMA_CODECS,
      mappers: MAPPERS,
      requestEncoders: {
        RpcResourceRequestEncoder: UNARY_ENCODER,
      },
      transports: { CONNECT_REFERENCE_UNARY: transport },
    });

    await expect(
      runtime
        .bindUnary("GET_RPC_RESOURCE", isResourceView)
        .execute({ resourceId: "resource-2" }),
    ).resolves.toMatchObject({
      ok: true,
      value: { id: "resource-2", label: "Retried resource" },
    });
    expect(calls).toBe(2);
  });

  it("drops a late unary result after its scope generation changes", async () => {
    const transport = defineBrowserRpcTransport({
      runtimeProfileId: "CONNECT_REFERENCE_UNARY",
      providerId: "REFERENCE_RPC",
      protocol: "CONNECT_HTTP",
      rpcKind: "UNARY",
      async invokeUnary() {
        return {
          ok: true,
          message: { id: "late", name: "Late resource" },
          encodedBytes: 24,
        };
      },
    });
    const runtime = createBrowserRpcRuntime({
      ...baseDependencies(transport),
      generationFence: {
        capture: () => 1,
        isCurrent: () => false,
      },
    });

    await expect(
      runtime
        .bindUnary("GET_RPC_RESOURCE", isResourceView)
        .execute({ resourceId: "late" }),
    ).resolves.toMatchObject({
      ok: false,
      error: {
        kind: "SCOPE_GENERATION_CHANGED",
        code: "RPC_SCOPE_GENERATION_CHANGED",
      },
    });
  });

  it("uses an explicit unavailable adapter without network fallback", async () => {
    const transport = createUnavailableBrowserRpcTransport({
      runtimeProfileId: "CONNECT_REFERENCE_UNARY",
      providerId: "REFERENCE_RPC",
      protocol: "CONNECT_HTTP",
      rpcKind: "UNARY",
    });
    const result = await unaryRuntime(transport)
      .bindUnary("GET_RPC_RESOURCE", isResourceView)
      .execute({ resourceId: "resource-1" });

    expect(result).toMatchObject({
      ok: false,
      error: {
        kind: "SERVER_FAILURE",
        code: "RPC_UNAVAILABLE",
      },
    });
  });

  it("classifies valid oversized transport metadata as a response limit", async () => {
    const transport = defineBrowserRpcTransport({
      runtimeProfileId: "CONNECT_REFERENCE_UNARY",
      providerId: "REFERENCE_RPC",
      protocol: "CONNECT_HTTP",
      rpcKind: "UNARY",
      async invokeUnary() {
        return {
          ok: true,
          message: { id: "large", name: "Large resource" },
          encodedBytes: 4_097,
        };
      },
    });

    await expect(
      unaryRuntime(transport)
        .bindUnary("GET_RPC_RESOURCE", isResourceView)
        .execute({ resourceId: "large" }),
    ).resolves.toMatchObject({
      ok: false,
      error: {
        kind: "RESPONSE_BODY_LIMIT",
        code: "RPC_RESPONSE_MESSAGE_LIMIT",
      },
    });
  });

  it("rejects transports that expose the wrong call shape", () => {
    expect(() =>
      defineBrowserRpcTransport({
        runtimeProfileId: "CONNECT_REFERENCE_UNARY",
        providerId: "REFERENCE_RPC",
        protocol: "CONNECT_HTTP",
        rpcKind: "UNARY",
        async invokeUnary() {
          return {
            ok: false,
            failure: { code: "UNAVAILABLE" },
          };
        },
        openServerStream() {
          return serverStreamLease(
            (async function* () {
              yield Object.freeze({ kind: "TERMINAL" as const, ok: true as const });
            })(),
          );
        },
      }),
    ).toThrow("transport is invalid");
  });

  it("commits mapped stream messages only before one valid terminal envelope", async () => {
    const transport = streamTransport(async function* () {
      yield message("resource-1", "One", 24);
      yield message("resource-2", "Two", 24);
      yield Object.freeze({ kind: "TERMINAL", ok: true });
    });
    const observations: BrowserRpcObservation[] = [];
    const stream = streamRuntime(transport, {
      observe(value) {
        observations.push(value);
      },
    })
      .bindServerStream("WATCH_RPC_RESOURCES", isResourceView)
      .open({ resourceId: "scope-1" });

    await expect(collect(stream)).resolves.toEqual([
      {
        ok: true,
        value: { id: "resource-1", label: "One" },
      },
      {
        ok: true,
        value: { id: "resource-2", label: "Two" },
      },
    ]);
    expect(observations).toEqual([
      {
        operationId: "WATCH_RPC_RESOURCES",
        protocol: "CONNECT_HTTP",
        runtimeProfileId: "CONNECT_REFERENCE_STREAM",
        rpcKind: "SERVER_STREAM",
        outcome: "SUCCESS",
        attemptCount: 1,
        messageCount: 2,
      },
    ]);
  });

  it("rejects EOF without terminal and data after terminal", async () => {
    const missingTerminal = streamTransport(async function* () {
      yield message("resource-1", "One", 24);
    });
    const missingResults = await collect(
      streamRuntime(missingTerminal)
        .bindServerStream("WATCH_RPC_RESOURCES", isResourceView)
        .open({ resourceId: "scope-1" }),
    );
    expect(missingResults.at(-1)).toMatchObject({
      ok: false,
      error: {
        kind: "API_CONTRACT_MISMATCH",
        code: "RPC_PROTOCOL_MISMATCH",
      },
    });

    const afterTerminal = streamTransport(async function* () {
      yield Object.freeze({ kind: "TERMINAL", ok: true });
      yield message("resource-2", "Two", 24);
    });
    await expect(
      collect(
        streamRuntime(afterTerminal)
          .bindServerStream("WATCH_RPC_RESOURCES", isResourceView)
          .open({ resourceId: "scope-1" }),
      ),
    ).resolves.toMatchObject([
      {
        ok: false,
        error: {
          kind: "API_CONTRACT_MISMATCH",
          code: "RPC_PROTOCOL_MISMATCH",
        },
      },
    ]);
  });

  it("aborts the transport when stream message limits are exceeded", async () => {
    let cleaned = false;
    const transport = streamTransport(async function* (signal) {
      try {
        yield message("resource-1", "One", 24);
        yield message("resource-2", "Two", 24);
        yield Object.freeze({ kind: "TERMINAL", ok: true });
      } finally {
        cleaned = signal.aborted;
      }
    });
    const operation = streamOperation({ maxResponseMessages: 1 });
    const results = await collect(
      streamRuntime(transport, undefined, operation)
        .bindServerStream("WATCH_RPC_RESOURCES", isResourceView)
        .open({ resourceId: "scope-1" }),
    );

    expect(results.at(-1)).toMatchObject({
      ok: false,
      error: {
        kind: "RESPONSE_BODY_LIMIT",
        code: "RPC_STREAM_MESSAGE_LIMIT",
      },
    });
    expect(cleaned).toBe(true);
  });
});

function unaryRuntime(
  transport: BrowserRpcTransport,
  observations?: Readonly<{
    observe(value: BrowserRpcObservation): void;
  }>,
) {
  return createBrowserRpcRuntime({
    ...baseDependencies(transport),
    observations,
  });
}

function baseDependencies(transport: BrowserRpcTransport) {
  return {
    operations: { GET_RPC_RESOURCE: unaryOperation() },
    profiles: { CONNECT_REFERENCE_UNARY: unaryProfile() },
    schemaCodecs: SCHEMA_CODECS,
    mappers: MAPPERS,
    requestEncoders: {
      RpcResourceRequestEncoder: UNARY_ENCODER,
    },
    transports: { CONNECT_REFERENCE_UNARY: transport },
  } as const;
}

function streamRuntime(
  transport: BrowserRpcTransport,
  observations?: Readonly<{
    observe(value: BrowserRpcObservation): void;
  }>,
  operation = streamOperation(),
) {
  return createBrowserRpcRuntime({
    operations: { WATCH_RPC_RESOURCES: operation },
    profiles: { CONNECT_REFERENCE_STREAM: streamProfile() },
    schemaCodecs: SCHEMA_CODECS,
    mappers: MAPPERS,
    requestEncoders: {
      RpcResourceStreamRequestEncoder: STREAM_ENCODER,
    },
    transports: { CONNECT_REFERENCE_STREAM: transport },
    observations,
  });
}

function streamTransport(
  source: (
    signal: AbortSignal,
  ) => AsyncIterable<BrowserRpcStreamFrame>,
): BrowserRpcTransport {
  return defineBrowserRpcTransport({
    runtimeProfileId: "CONNECT_REFERENCE_STREAM",
    providerId: "REFERENCE_RPC",
    protocol: "CONNECT_HTTP",
    rpcKind: "SERVER_STREAM",
    openServerStream(call) {
      return serverStreamLease(source(call.signal));
    },
  });
}

/**
 * RPC-RR-01. Wraps a plain frame sequence in the lease the transport contract
 * requires. `cancel` resolves `waitClosed`, which is what a cooperative
 * transport does.
 */
let leaseSequence = 0;

function serverStreamLease(
  frames: AsyncIterable<BrowserRpcStreamFrame>,
  options: Readonly<{
    onCancel?: (reason: string) => void;
    closeOnCancel?: boolean;
  }> = {},
): BrowserRpcServerStreamLease {
  leaseSequence += 1;
  let release: (() => void) | undefined;
  const closed = new Promise<void>((resolve) => {
    release = resolve;
  });
  return Object.freeze({
    streamId: `test-stream-${leaseSequence}`,
    frames,
    cancel(reason: string) {
      options.onCancel?.(reason);
      if (options.closeOnCancel !== false) release?.();
    },
    waitClosed: () => closed,
  });
}

function message(
  id: string,
  name: string,
  encodedBytes: number,
): BrowserRpcStreamFrame {
  return Object.freeze({
    kind: "MESSAGE",
    message: Object.freeze({ id, name }),
    encodedBytes,
  });
}

async function collect<Value>(
  iterable: AsyncIterable<Result<Value, AppFailure>>,
): Promise<readonly Result<Value, AppFailure>[]> {
  const values: Result<Value, AppFailure>[] = [];
  for await (const value of iterable) values.push(value);
  return values;
}