tech-log-frontend/tests/unit/resumable-upload-fetch-transport.test.ts

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

import {
  createResumableUploadFetchJsonTransport,
  type ResumableUploadFetchTransportDependencies,
} from "../../src/adapters/browser-transfer/resumable-upload/fetch-json-transport.ts";
import type {
  ResumableUploadControlOperation,
  ResumableUploadJsonTransport,
} from "../../src/adapters/browser-transfer/resumable-upload/http-control-plane-adapter.ts";
import { createResumableUploadWebLock } from "../../src/adapters/browser-transfer/resumable-upload/upload-mutation-lock.ts";

const API_ORIGIN = "https://api.example";
const ENDPOINTS = Object.freeze({
  CREATE_SESSION: `${API_ORIGIN}/uploads/create`,
  GET_STATUS: `${API_ORIGIN}/uploads/status`,
  COMPLETE: `${API_ORIGIN}/uploads/complete`,
  ABORT: `${API_ORIGIN}/uploads/abort`,
});
const activeSignal = new AbortController().signal;

function responseAt(
  url: string,
  body: BodyInit | null,
  init: ResponseInit,
): Response {
  const response = new Response(body, init);
  Object.defineProperty(response, "url", {
    configurable: true,
    value: url,
  });
  return response;
}

function jsonResponseAt(
  url: string,
  value: unknown,
  status = 200,
  headers: Readonly<Record<string, string>> = {},
): Response {
  const body = JSON.stringify(value);
  return responseAt(url, body, {
    status,
    headers: {
      "content-type": "application/json; charset=utf-8",
      "content-length": String(new TextEncoder().encode(body).byteLength),
      ...headers,
    },
  });
}

function createTransport(
  fetcher: typeof fetch,
  overrides: Partial<ResumableUploadFetchTransportDependencies> = {},
): ResumableUploadJsonTransport {
  return createResumableUploadFetchJsonTransport({
    endpoints: ENDPOINTS,
    allowedOrigins: [API_ORIGIN],
    credentials: "same-origin",
    fetcher,
    timeoutMs: 100,
    maxRequestBytes: 4_096,
    maxResponseBytes: 4_096,
    maxRetryAfterMs: 3_000,
    ...overrides,
  });
}

describe("resumable upload fetch transport", () => {
  it.each([
    { label: "delta-seconds", header: "2", now: 1_000, expected: 2_000 },
    { label: "HTTP-date ahead", header: "Thu, 01 Jan 1970 00:00:03 GMT", now: 1_000, expected: 2_000 },
    { label: "HTTP-date behind (clock rollback)", header: "Thu, 01 Jan 1970 00:00:01 GMT", now: 9_000, expected: 0 },
  ])(
    "resolves Retry-After against the injected clock ($label)",
    async ({ header, now, expected }) => {
      // BT-UP-02. Both branches use the same captured `now`, so boundaries and
      // clock rollback are deterministic.
      const transport = createTransport(
        (async () =>
          responseAt(ENDPOINTS.GET_STATUS, null, {
            status: 429,
            headers: { "retry-after": header },
          })) as unknown as typeof fetch,
        { nowEpochMs: () => now },
      );

      const result = await transport.execute({
        operation: "GET_STATUS",
        body: { sessionId: "session_01" },
        signal: new AbortController().signal,
      });
      expect(result).toMatchObject({
        ok: false,
        error: { code: "UNAVAILABLE", retryAfterMs: expected },
      });
    },
  );

  it("ignores an invalid Retry-After date instead of guessing", async () => {
    const transport = createTransport(
      (async () =>
        responseAt(ENDPOINTS.GET_STATUS, null, {
          status: 429,
          headers: { "retry-after": "not-a-date" },
        })) as unknown as typeof fetch,
      { nowEpochMs: () => 1_000 },
    );

    const result = await transport.execute({
      operation: "GET_STATUS",
      body: { sessionId: "session_01" },
      signal: new AbortController().signal,
    });
    expect(result).toMatchObject({ ok: false });
    if (result.ok) return;
    expect(result.error.retryAfterMs).toBeUndefined();
  });

  it("uses a closed operation map and fixed production fetch policy", async () => {
    let receivedUrl = "";
    let receivedInit: RequestInit | undefined;
    const fetcher = vi.fn(
      async (
        input: RequestInfo | URL,
        init?: RequestInit,
      ): Promise<Response> => {
        receivedUrl = String(input);
        receivedInit = init;
        return jsonResponseAt(ENDPOINTS.GET_STATUS, { state: "ok" });
      },
    ) as unknown as typeof fetch;
    const transport = createTransport(fetcher, {
      requestHeaders: [{ name: "x-runtime-version", value: "v1" }],
    });

    const result = await transport.execute({
      operation: "GET_STATUS",
      body: { sessionId: "session_01" },
      signal: activeSignal,
    });

    expect(result).toEqual({
      ok: true,
      value: { state: "ok" },
    });
    expect(receivedUrl).toBe(ENDPOINTS.GET_STATUS);
    expect(receivedInit).toMatchObject({
      method: "POST",
      credentials: "same-origin",
      redirect: "error",
      referrerPolicy: "no-referrer",
      cache: "no-store",
      mode: "cors",
    });
    const headers = new Headers(receivedInit?.headers);
    expect(headers.get("accept")).toBe("application/json");
    expect(headers.get("content-type")).toBe(
      "application/json; charset=utf-8",
    );
    expect(headers.get("x-runtime-version")).toBe("v1");
  });

  it("rejects a runtime operation outside the allowlist before fetch", async () => {
    const fetcher = vi.fn();
    const transport = createTransport(
      fetcher as unknown as typeof fetch,
    );
    const execute = transport.execute as (
      input: Readonly<{
        operation: string;
        body: Readonly<Record<string, unknown>>;
        signal: AbortSignal;
      }>,
    ) => ReturnType<ResumableUploadJsonTransport["execute"]>;

    const result = await execute({
      operation: "DELETE_EVERYTHING",
      body: {},
      signal: activeSignal,
    });

    expect(result).toMatchObject({
      ok: false,
      error: { code: "INVALID_INPUT" },
    });
    expect(fetcher).not.toHaveBeenCalled();
  });

  it("keeps timeout active while a response body is stalled and cancels it", async () => {
    const cancel = vi.fn();
    const stalled = new ReadableStream<Uint8Array>({
      cancel,
    });
    const fetcher = vi.fn(async () =>
      responseAt(ENDPOINTS.GET_STATUS, stalled, {
        status: 200,
        headers: { "content-type": "application/json" },
      }),
    ) as unknown as typeof fetch;
    const transport = createTransport(fetcher, { timeoutMs: 5 });

    const result = await transport.execute({
      operation: "GET_STATUS",
      body: {},
      signal: activeSignal,
    });
    await Promise.resolve();

    expect(result).toMatchObject({
      ok: false,
      error: {
        code: "UNAVAILABLE",
        retryable: true,
        recovery: "RESUME",
      },
    });
    expect(cancel).toHaveBeenCalledTimes(1);
  });

  it("propagates parent abort during a stalled body and cancels promptly", async () => {
    const cancel = vi.fn();
    const stalled = new ReadableStream<Uint8Array>({
      cancel,
    });
    const fetcher = vi.fn(async () =>
      responseAt(ENDPOINTS.GET_STATUS, stalled, {
        status: 200,
        headers: { "content-type": "application/json" },
      }),
    ) as unknown as typeof fetch;
    const transport = createTransport(fetcher, { timeoutMs: 1_000 });
    const controller = new AbortController();

    const pending = transport.execute({
      operation: "GET_STATUS",
      body: {},
      signal: controller.signal,
    });
    await Promise.resolve();
    controller.abort();
    const result = await pending;
    await Promise.resolve();

    expect(result).toMatchObject({
      ok: false,
      error: { code: "ABORTED", retryable: false },
    });
    expect(cancel).toHaveBeenCalledTimes(1);
  });

  it("requires Content-Length to match the bytes actually consumed", async () => {
    const body = JSON.stringify({ state: "ok" });
    const fetcher = vi.fn(async () =>
      responseAt(ENDPOINTS.GET_STATUS, body, {
        status: 200,
        headers: {
          "content-type": "application/json",
          "content-length": String(body.length + 1),
        },
      }),
    ) as unknown as typeof fetch;

    const result = await createTransport(fetcher).execute({
      operation: "GET_STATUS",
      body: {},
      signal: activeSignal,
    });

    expect(result).toMatchObject({
      ok: false,
      error: {
        code: "INTEGRITY_FAILED",
        recovery: "RECONCILE",
      },
    });
  });

  it("bounds Retry-After and cancels an unconsumed error body", async () => {
    const cancel = vi.fn();
    const body = new ReadableStream<Uint8Array>({
      start(controller) {
        controller.enqueue(new Uint8Array([1]));
      },
      cancel,
    });
    const fetcher = vi.fn(async () =>
      responseAt(ENDPOINTS.GET_STATUS, body, {
        status: 429,
        headers: { "retry-after": "2" },
      }),
    ) as unknown as typeof fetch;

    const result = await createTransport(fetcher).execute({
      operation: "GET_STATUS",
      body: {},
      signal: activeSignal,
    });
    await Promise.resolve();

    expect(result).toMatchObject({
      ok: false,
      error: {
        code: "UNAVAILABLE",
        retryable: true,
        retryAfterMs: 2_000,
      },
    });
    expect(cancel).toHaveBeenCalledTimes(1);
  });

  it.each([500, 599])(
    "classifies HTTP %s as retryable; runtime attempt limits remain authoritative",
    async (status) => {
      const fetcher = vi.fn(async () =>
        jsonResponseAt(
          ENDPOINTS.GET_STATUS,
          { error: "closed" },
          status,
        ),
      ) as unknown as typeof fetch;

      const result = await createTransport(fetcher).execute({
        operation: "GET_STATUS",
        body: {},
        signal: activeSignal,
      });

      expect(result).toMatchObject({
        ok: false,
        error: {
          code: "UNAVAILABLE",
          retryable: true,
          recovery: "RESUME",
        },
      });
    },
  );

  it("cancels bodies rejected by URL and content-type policy", async () => {
    const urlCancel = vi.fn();
    const typeCancel = vi.fn();
    const responses = [
      responseAt(
        `${API_ORIGIN}/unexpected`,
        new ReadableStream<Uint8Array>({ cancel: urlCancel }),
        {
          status: 200,
          headers: { "content-type": "application/json" },
        },
      ),
      responseAt(
        ENDPOINTS.GET_STATUS,
        new ReadableStream<Uint8Array>({ cancel: typeCancel }),
        {
          status: 200,
          headers: { "content-type": "text/html" },
        },
      ),
    ];
    const fetcher = vi.fn(async () => responses.shift()!) as unknown as typeof fetch;
    const transport = createTransport(fetcher);

    const wrongUrl = await transport.execute({
      operation: "GET_STATUS",
      body: {},
      signal: activeSignal,
    });
    const wrongType = await transport.execute({
      operation: "GET_STATUS",
      body: {},
      signal: activeSignal,
    });
    await Promise.resolve();

    expect(wrongUrl).toMatchObject({
      ok: false,
      error: { code: "POLICY_REJECTED" },
    });
    expect(wrongType).toMatchObject({
      ok: false,
      error: { code: "CORRUPT_DATA" },
    });
    expect(urlCancel).toHaveBeenCalledTimes(1);
    expect(typeCancel).toHaveBeenCalledTimes(1);
  });

  it.each([
    "authorization",
    "proxy-authorization",
    "set-cookie",
    "connection",
    "transfer-encoding",
  ])("rejects composition headers that can alter authority: %s", (name) => {
    const fetcher = vi.fn() as unknown as typeof fetch;
    expect(() =>
      createTransport(fetcher, {
        requestHeaders: [{ name, value: "forbidden" }],
      }),
    ).toThrow(TypeError);
  });

  it("rejects unknown expected-success operation keys", () => {
    const fetcher = vi.fn() as unknown as typeof fetch;
    expect(() =>
      createTransport(fetcher, {
        expectedSuccessStatuses: {
          DELETE_EVERYTHING: 204,
        } as unknown as Partial<
          Readonly<Record<ResumableUploadControlOperation, number>>
        >,
      }),
    ).toThrow(TypeError);
  });
});

describe("resumable upload Web Lock", () => {
  it("uses one exclusive opaque lock name and serializes mutations", async () => {
    const requests: Array<
      Readonly<{
        name: string;
        options: Readonly<{
          mode: "exclusive";
          signal?: AbortSignal;
        }>;
      }>
    > = [];
    let queue = Promise.resolve();
    const manager = {
      request<Value>(
        name: string,
        options: Readonly<{
          mode: "exclusive";
          signal?: AbortSignal;
        }>,
        callback: (lock: unknown) => Promise<Value>,
      ): Promise<Value> {
        requests.push({ name, options });
        const result = queue.then(async () => await callback({ name }));
        queue = result.then(
          () => undefined,
          () => undefined,
        );
        return result;
      },
    } as unknown as LockManager;
    const lock = createResumableUploadWebLock(
      manager,
      "upload-runtime-v1",
    );
    const events: string[] = [];
    let releaseFirst!: () => void;
    const gate = new Promise<void>((resolve) => {
      releaseFirst = resolve;
    });

    const first = lock.run(
      "upload_key_lock",
      activeSignal,
      async () => {
        events.push("first:start");
        await gate;
        events.push("first:end");
        return 1;
      },
    );
    const second = lock.run(
      "upload_key_lock",
      activeSignal,
      async () => {
        events.push("second:start");
        return 2;
      },
    );
    await Promise.resolve();
    await Promise.resolve();

    expect(events).toEqual(["first:start"]);
    releaseFirst();
    await expect(Promise.all([first, second])).resolves.toEqual([1, 2]);
    expect(events).toEqual([
      "first:start",
      "first:end",
      "second:start",
    ]);
    expect(requests).toEqual([
      {
        name: "upload-runtime-v1:upload_key_lock",
        options: { mode: "exclusive", signal: activeSignal },
      },
      {
        name: "upload-runtime-v1:upload_key_lock",
        options: { mode: "exclusive", signal: activeSignal },
      },
    ]);
  });

  /**
   * X-AUDIT-02. The public port promises a `UploadProviderResult`. A scheduler
   * that cannot install the attempt deadline must close the attempt inside that
   * contract instead of rejecting it, and must not leave the caller listener
   * attached to the parent signal.
   */
  describe("scheduler boundary", () => {
    const trackedSignal = () => {
      const controller = new AbortController();
      const added: string[] = [];
      const removed: string[] = [];
      const add = controller.signal.addEventListener.bind(controller.signal);
      const remove = controller.signal.removeEventListener.bind(
        controller.signal,
      );
      Object.defineProperty(controller.signal, "addEventListener", {
        configurable: true,
        value: (type: string, ...rest: readonly unknown[]) => {
          added.push(type);
          return (add as (...args: readonly unknown[]) => unknown)(
            type,
            ...rest,
          );
        },
      });
      Object.defineProperty(controller.signal, "removeEventListener", {
        configurable: true,
        value: (type: string, ...rest: readonly unknown[]) => {
          removed.push(type);
          return (remove as (...args: readonly unknown[]) => unknown)(
            type,
            ...rest,
          );
        },
      });
      return { controller, added, removed };
    };

    it("closes the attempt when the scheduler cannot install the deadline", async () => {
      const fetcher = vi.fn(async () => new Response(null, { status: 200 }));
      const { controller, added, removed } = trackedSignal();
      const transport = createTransport(fetcher as unknown as typeof fetch, {
        scheduler: {
          setTimeout: () => {
            throw new TypeError("upload scheduler install exploded");
          },
          clearTimeout: () => {},
        },
      });

      const result = await transport.execute({
        operation: "GET_STATUS",
        body: { sessionId: "session_01" },
        signal: controller.signal,
      });

      expect(result.ok).toBe(false);
      if (!result.ok) {
        expect(result.error.code).toBe("UNAVAILABLE");
        expect(result.error.retryable).toBe(true);
      }
      expect(fetcher).not.toHaveBeenCalled();
      expect(added.filter((type) => type === "abort")).toHaveLength(1);
      expect(removed.filter((type) => type === "abort")).toHaveLength(1);
    });

    it("keeps the classified outcome when clearing the deadline throws", async () => {
      const fetcher = vi.fn(
        async () =>
          jsonResponseAt(ENDPOINTS.GET_STATUS, { sessionId: "session_01" }),
      );
      const { controller, added, removed } = trackedSignal();
      const transport = createTransport(fetcher as unknown as typeof fetch, {
        scheduler: {
          setTimeout: (callback: () => void, delayMs: number) =>
            setTimeout(callback, delayMs),
          clearTimeout: () => {
            throw new TypeError("upload scheduler clear exploded");
          },
        },
      });

      const result = await transport.execute({
        operation: "GET_STATUS",
        body: { sessionId: "session_01" },
        signal: controller.signal,
      });

      expect(result.ok).toBe(true);
      expect(added.filter((type) => type === "abort")).toHaveLength(1);
      expect(removed.filter((type) => type === "abort")).toHaveLength(1);
    });

    it("uses the scheduler methods captured at construction", async () => {
      const fetcher = vi.fn(
        async () =>
          jsonResponseAt(ENDPOINTS.GET_STATUS, { sessionId: "session_01" }),
      );
      const scheduler = {
        setTimeout: (callback: () => void, delayMs: number) =>
          setTimeout(callback, delayMs),
        clearTimeout: (handle: unknown) => {
          clearTimeout(handle as ReturnType<typeof setTimeout>);
        },
      };
      const transport = createTransport(fetcher as unknown as typeof fetch, {
        scheduler,
      });
      scheduler.setTimeout = () => {
        throw new TypeError("mutated upload setTimeout");
      };
      scheduler.clearTimeout = () => {
        throw new TypeError("mutated upload clearTimeout");
      };

      await expect(
        transport.execute({
          operation: "GET_STATUS",
          body: { sessionId: "session_01" },
          signal: new AbortController().signal,
        }),
      ).resolves.toMatchObject({ ok: true });
    });

    it("starts no timer and no fetch for an already aborted caller", async () => {
      const fetcher = vi.fn(async () => new Response(null, { status: 200 }));
      const setTimeout_ = vi.fn(
        (callback: () => void, delayMs: number) =>
          setTimeout(callback, delayMs) as unknown,
      );
      const controller = new AbortController();
      controller.abort();
      const transport = createTransport(fetcher as unknown as typeof fetch, {
        scheduler: {
          setTimeout: setTimeout_,
          clearTimeout: (handle: unknown) => {
            clearTimeout(handle as ReturnType<typeof setTimeout>);
          },
        },
      });

      const result = await transport.execute({
        operation: "GET_STATUS",
        body: { sessionId: "session_01" },
        signal: controller.signal,
      });

      expect(result.ok).toBe(false);
      if (!result.ok) expect(result.error.code).toBe("ABORTED");
      expect(fetcher).not.toHaveBeenCalled();
      expect(setTimeout_).not.toHaveBeenCalled();
    });
  });
});