tech-log-frontend/tests/unit/resumable-upload-http-control-plane.test.ts

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

import { PRESIGNED_TRANSFER_PROTOCOL } from "../../src/application/ports/browser-transfer/presigned-transfer.ts";

import type {
  PresignedUploadPartCapabilityProvider,
} from "../../src/application/ports/browser-transfer/presigned-transfer.ts";
import type {
  ResumableUploadCheckpoint,
  ResumableUploadCheckpointStore,
  UploadFileFingerprint,
  UploadPartReceipt,
  UploadSession,
} from "../../src/application/ports/browser-transfer/resumable-upload.ts";
import { RESUMABLE_UPLOAD_PROTOCOL } from "../../src/application/ports/browser-transfer/resumable-upload.ts";
import type { BrowserDataResult } from "../../src/application/ports/browser-file-storage/shared.ts";
import {
  browserDataFailure,
  browserDataSuccess,
} from "../../src/adapters/browser-file-storage/result.ts";
import {
  createPresignedCapabilityVault,
  createSingleUsePresignedReplayGuard,
} from "../../src/adapters/browser-transfer/presigned/presigned-capability-vault.ts";
import { createPresignedCapabilityHttpProvider } from "../../src/adapters/browser-transfer/presigned/presigned-capability-http-provider.ts";
import { createPresignedTransferExecutor } from "../../src/adapters/browser-transfer/presigned/presigned-transfer-executor.ts";
import { createResumableUploadFetchJsonTransport } from "../../src/adapters/browser-transfer/resumable-upload/fetch-json-transport.ts";
import {
  createResumableUploadHttpControlPlane,
  type ResumableUploadJsonTransport,
} from "../../src/adapters/browser-transfer/resumable-upload/http-control-plane-adapter.ts";
import { createPresignedUploadPartExecutor } from "../../src/adapters/browser-transfer/resumable-upload/presigned-upload-part-executor.ts";
import { createResumableUploadRuntime } from "../../src/adapters/browser-transfer/resumable-upload/resumable-upload-runtime.ts";
import type { UploadMutationLock } from "../../src/adapters/browser-transfer/resumable-upload/upload-mutation-lock.ts";

const NOW = 1_000_000;
const API_ORIGIN = "https://api.example";
const OBJECT_ORIGIN = "https://objects.example";
const CAPABILITY_ENDPOINT = `${API_ORIGIN}/capabilities`;
const CONTROL_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 CHECKSUM_HEADER = "x-checksum-sha256";
const POLICY_HEADER = "x-policy-version";
const activeSignal = new AbortController().signal;

const noContentionLock: UploadMutationLock = Object.freeze({
  async run<Value>(
    _uploadKey: string,
    _signal: AbortSignal,
    task: () => Promise<Value>,
  ): Promise<Value> {
    return await task();
  },
});

class MemoryCheckpointStore implements ResumableUploadCheckpointStore {
  readonly rows = new Map<string, ResumableUploadCheckpoint>();

  async read(
    uploadKey: string,
  ): Promise<BrowserDataResult<ResumableUploadCheckpoint | null>> {
    return browserDataSuccess(
      structuredClone(this.rows.get(uploadKey) ?? null),
    );
  }

  async compareAndSwap(
    input: Parameters<
      ResumableUploadCheckpointStore["compareAndSwap"]
    >[0],
  ): Promise<BrowserDataResult<ResumableUploadCheckpoint>> {
    const current = this.rows.get(input.checkpoint.uploadKey);
    if (
      (input.expectedRevision === null && current) ||
      (input.expectedRevision !== null &&
        current?.revision !== input.expectedRevision)
    ) {
      return browserDataFailure("CONFLICT", "UPLOAD_RECONCILE", {
        recovery: "RECONCILE",
      });
    }
    const snapshot = structuredClone(input.checkpoint);
    this.rows.set(snapshot.uploadKey, snapshot);
    return browserDataSuccess(snapshot);
  }

  async remove(
    input: Parameters<
      ResumableUploadCheckpointStore["remove"]
    >[0],
  ): Promise<BrowserDataResult<void>> {
    const current = this.rows.get(input.uploadKey);
    if (current?.revision !== input.expectedRevision) {
      return browserDataFailure("CONFLICT", "UPLOAD_RECONCILE", {
        recovery: "RECONCILE",
      });
    }
    this.rows.delete(input.uploadKey);
    return browserDataSuccess(undefined);
  }

  close(): void {}
}

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,
): 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),
    },
  });
}

function rangeSource(bytes: Uint8Array) {
  return Object.freeze({
    kind: "RANGE_READER" as const,
    reader: Object.freeze({
      byteLength: bytes.byteLength,
      async readRange(input: Readonly<{
        offset: number;
        length: number;
        signal: AbortSignal;
      }>) {
        return input.signal.aborted
          ? browserDataFailure("ABORTED", "FILE_READ")
          : browserDataSuccess(
              bytes.slice(
                input.offset,
                input.offset + input.length,
              ),
            );
      },
    }),
  });
}

describe("resumable upload HTTP control plane", () => {
  /**
   * TR-RR-08. `Object.keys` sees only enumerable own string keys, so a symbol
   * or non-enumerable extra passed unseen and a later property read invoked
   * whatever accessor the sender installed — escaping the Result contract as a
   * rejection of a public method.
   */
  it("closes every hostile control-plane object as typed CORRUPT_DATA", async () => {
    const fingerprint: UploadFileFingerprint = Object.freeze({
      algorithm: "SHA-256-PARTS-V1",
      digestHex: "a".repeat(64),
      byteLength: 4,
      partSizeBytes: 4,
      partCount: 1,
    });
    const validSession = () => ({
      protocol: RESUMABLE_UPLOAD_PROTOCOL,
      sessionId: "session_01",
      requestBindingSha256: "b".repeat(64),
      fingerprint,
      partSizeBytes: 4,
      partCount: 1,
      maxConcurrency: 1,
      expiresAtEpochMs: NOW + 10_000,
    });

    const hostile: readonly (readonly [string, () => unknown])[] = [
      [
        "throwing getter",
        () => {
          const value = validSession() as Record<string, unknown>;
          Object.defineProperty(value, "sessionId", {
            configurable: true,
            enumerable: true,
            get: () => {
              throw new TypeError("hostile getter");
            },
          });
          return value;
        },
      ],
      [
        "symbol key",
        () => ({ ...validSession(), [Symbol("injected")]: "leak" }),
      ],
      [
        "non-enumerable extra",
        () => {
          const value = validSession() as Record<string, unknown>;
          Object.defineProperty(value, "signedUrl", {
            configurable: true,
            enumerable: false,
            value: "https://objects.example/secret?signature=leak",
          });
          return value;
        },
      ],
      [
        "ownKeys trap",
        () =>
          new Proxy(validSession() as Record<string, unknown>, {
            ownKeys() {
              throw new TypeError("hostile ownKeys");
            },
          }),
      ],
      [
        "getOwnPropertyDescriptor trap",
        () =>
          new Proxy(validSession() as Record<string, unknown>, {
            getOwnPropertyDescriptor() {
              throw new TypeError("hostile descriptor");
            },
          }),
      ],
      [
        "a nested fingerprint with an extra field",
        () => ({
          ...validSession(),
          fingerprint: { ...fingerprint, injected: true },
        }),
      ],
      [
        "a nested fingerprint behind an accessor",
        () => {
          const value = validSession() as Record<string, unknown>;
          Object.defineProperty(value, "fingerprint", {
            configurable: true,
            enumerable: true,
            get: () => fingerprint,
          });
          return value;
        },
      ],
      [
        "a custom prototype",
        () =>
          Object.assign(Object.create({ injected: true }), validSession()),
      ],
    ];

    for (const [label, build] of hostile) {
      const control = createResumableUploadHttpControlPlane({
        transport: {
          async execute() {
            return browserDataSuccess(build());
          },
        },
        partCapabilities: { issueUploadPart: vi.fn() },
      });
      const result = await control.createSession({
        protocol: RESUMABLE_UPLOAD_PROTOCOL,
        uploadKey: "upload_key_strict",
        purpose: "attachment",
        mediaType: "application/octet-stream",
        requestBindingSha256: "b".repeat(64),
        fingerprint,
        requestedPartSizeBytes: 4,
        requestedMaxConcurrency: 1,
        idempotencyKey: "upload-create-idempotency-01",
        signal: activeSignal,
      });
      expect(result, label).toMatchObject({
        ok: false,
        error: { code: "CORRUPT_DATA", operation: "UPLOAD_SESSION" },
      });
      expect(JSON.stringify(result)).not.toContain("signature=leak");
    }

    /**
     * TR-05. The decoder checked the sender's object and then read it again to
     * build the result, so a stateful answer could show a safe `sessionId` to
     * the regex and hand an unvalidated one to the receipt. Reading once means
     * the value that was validated is the value that is returned.
     */
    let sessionIdReads = 0;
    const statefulControl = createResumableUploadHttpControlPlane({
      transport: {
        async execute() {
          return browserDataSuccess(
            new Proxy(validSession() as Record<string, unknown>, {
              getOwnPropertyDescriptor(target, key) {
                if (key === "sessionId") {
                  sessionIdReads += 1;
                  return {
                    configurable: true,
                    enumerable: true,
                    value: sessionIdReads > 1 ? "../../unsafe" : "session_01",
                  };
                }
                return Reflect.getOwnPropertyDescriptor(target, key);
              },
            }),
          );
        },
      },
      partCapabilities: { issueUploadPart: vi.fn() },
    });
    const stateful = await statefulControl.createSession({
      protocol: RESUMABLE_UPLOAD_PROTOCOL,
      uploadKey: "upload_key_strict",
      purpose: "attachment",
      mediaType: "application/octet-stream",
      requestBindingSha256: "b".repeat(64),
      fingerprint,
      requestedPartSizeBytes: 4,
      requestedMaxConcurrency: 1,
      idempotencyKey: "upload-create-idempotency-02",
      signal: activeSignal,
    });
    expect(sessionIdReads).toBe(1);
    expect(JSON.stringify(stateful)).not.toContain("../../unsafe");
    if (stateful.ok) {
      expect(stateful.value.sessionId).toBe("session_01");
    }
  });

  it("rejects unknown response fields so URLs cannot cross the DTO boundary", async () => {
    const fingerprint: UploadFileFingerprint = Object.freeze({
      algorithm: "SHA-256-PARTS-V1",
      digestHex: "a".repeat(64),
      byteLength: 4,
      partSizeBytes: 4,
      partCount: 1,
    });
    const transport: ResumableUploadJsonTransport = {
      async execute() {
        return browserDataSuccess({
          protocol: RESUMABLE_UPLOAD_PROTOCOL,
          sessionId: "session_01",
          requestBindingSha256: "b".repeat(64),
          fingerprint,
          partSizeBytes: 4,
          partCount: 1,
          maxConcurrency: 1,
          expiresAtEpochMs: NOW + 10_000,
          signedUrl: "https://objects.example/secret?signature=leak",
        });
      },
    };
    const partCapabilities: PresignedUploadPartCapabilityProvider = {
      issueUploadPart: vi.fn(),
    };
    const control = createResumableUploadHttpControlPlane({
      transport,
      partCapabilities,
    });

    const result = await control.createSession({
      protocol: RESUMABLE_UPLOAD_PROTOCOL,
      uploadKey: "upload_key_strict",
      purpose: "attachment",
      mediaType: "application/octet-stream",
      requestBindingSha256: "b".repeat(64),
      fingerprint,
      requestedPartSizeBytes: 4,
      requestedMaxConcurrency: 1,
      idempotencyKey: "upload-create-idempotency-01",
      signal: activeSignal,
    });

    expect(result).toMatchObject({
      ok: false,
      error: {
        code: "CORRUPT_DATA",
        operation: "UPLOAD_SESSION",
        recovery: "RECONCILE",
      },
    });
    expect(JSON.stringify(result)).not.toContain("signature=leak");
  });

  it("runs create, list-parts, session-bound capability PUT and ordered complete through actual fetch adapters", async () => {
    const checkpoints = new MemoryCheckpointStore();
    const accepted = new Map<number, UploadPartReceipt>();
    const issuedBindings: Array<Readonly<Record<string, unknown>>> = [];
    const objectPutOptions: RequestInit[] = [];
    const completedParts: UploadPartReceipt[][] = [];
    const partByHref = new Map<
      string,
      Readonly<{
        partNumber: number;
        offset: number;
        byteLength: number;
        checksumSha256: string;
      }>
    >();
    let session: UploadSession | null = null;
    let completed = false;
    let capabilitySequence = 0;

    const fetcher = vi.fn(
      async (
        input: RequestInfo | URL,
        init?: RequestInit,
      ): Promise<Response> => {
        const url = String(input);
        if (url === CAPABILITY_ENDPOINT) {
          const request = JSON.parse(String(init?.body)) as Readonly<{
            method: string;
            binding: Readonly<{
              kind: string;
              sessionId: string;
              requestBindingSha256: string;
              uploadBindingSha256: string;
              partNumber: number;
              offset: number;
              idempotencyKey: string;
            }>;
            mediaType: string;
            byteLength: number;
            expectedSha256: string;
          }>;
          issuedBindings.push(
            Object.freeze({ ...request.binding }),
          );
          capabilitySequence += 1;
          const path =
            `/multipart/${request.binding.sessionId}` +
            `/part-${request.binding.partNumber}`;
          const href =
            `${OBJECT_ORIGIN}${path}` +
            `?signature=opaque-${capabilitySequence}`;
          partByHref.set(
            href,
            Object.freeze({
              partNumber: request.binding.partNumber,
              offset: request.binding.offset,
              byteLength: request.byteLength,
              checksumSha256: request.expectedSha256,
            }),
          );
          return jsonResponseAt(CAPABILITY_ENDPOINT, {
            // BT-PRE-02. The capability envelope declares its wire protocol.
            protocol: PRESIGNED_TRANSFER_PROTOCOL,
            capabilityReceipt: `capability-upload-${capabilitySequence}`,
            method: "PUT",
            binding: request.binding,
            href,
            origin: OBJECT_ORIGIN,
            path,
            allowedQueryParameters: ["signature"],
            requestHeaders: [
              {
                name: "content-type",
                value: request.mediaType,
              },
              {
                name: CHECKSUM_HEADER,
                value: request.expectedSha256,
              },
            ],
            requiredResponseHeaders: [
              { name: POLICY_HEADER, value: "v1" },
            ],
            digestRequestHeader: CHECKSUM_HEADER,
            digestResponseHeader: null,
            receiptResponseHeader: "etag",
            expectedStatus: 200,
            expectedResponseByteLength: 0,
            mediaType: request.mediaType,
            byteLength: request.byteLength,
            maxBytes: request.byteLength,
            expectedSha256: request.expectedSha256,
            expiresAtEpochMs: NOW + 30_000,
            singleUse: true,
          });
        }
        const part = partByHref.get(url);
        if (part) {
          objectPutOptions.push(init ?? {});
          const receiptToken = `etag-part-${part.partNumber}`;
          accepted.set(
            part.partNumber,
            Object.freeze({
              ...part,
              receiptToken,
            }),
          );
          return responseAt(url, null, {
            status: 200,
            headers: {
              "content-length": "0",
              [POLICY_HEADER]: "v1",
              etag: receiptToken,
            },
          });
        }

        const request = JSON.parse(String(init?.body)) as Record<
          string,
          unknown
        >;
        if (url === CONTROL_ENDPOINTS.CREATE_SESSION) {
          const fingerprint =
            request.fingerprint as UploadFileFingerprint;
          session = Object.freeze({
            protocol: RESUMABLE_UPLOAD_PROTOCOL,
            sessionId: "session_integration_01",
            requestBindingSha256: String(
              request.requestBindingSha256,
            ),
            fingerprint,
            partSizeBytes: Number(
              request.requestedPartSizeBytes,
            ),
            partCount: fingerprint.partCount,
            maxConcurrency: 2,
            expiresAtEpochMs: NOW + 60_000,
          });
          return jsonResponseAt(url, session, 201);
        }
        if (url === CONTROL_ENDPOINTS.GET_STATUS && session) {
          return completed
            ? jsonResponseAt(url, {
                state: "QUARANTINED",
                session,
                resourceId: "resource_integration_01",
              })
            : jsonResponseAt(url, {
                state: "ACTIVE",
                session,
                acceptedParts: [...accepted.values()].sort(
                  (left, right) =>
                    left.partNumber - right.partNumber,
                ),
              });
        }
        if (url === CONTROL_ENDPOINTS.COMPLETE && session) {
          completedParts.push(
            request.orderedParts as UploadPartReceipt[],
          );
          completed = true;
          return jsonResponseAt(url, {
            state: "QUARANTINED",
            protocol: RESUMABLE_UPLOAD_PROTOCOL,
            sessionId: session.sessionId,
            requestBindingSha256:
              session.requestBindingSha256,
            fingerprint: session.fingerprint,
            resourceId: "resource_integration_01",
          });
        }
        if (url === CONTROL_ENDPOINTS.ABORT) {
          return jsonResponseAt(url, { state: "ABORTED" });
        }
        throw new TypeError("Unexpected test endpoint.");
      },
    ) as unknown as typeof fetch;

    const vault = createPresignedCapabilityVault({
      maxActiveCapabilities: 16,
      now: () => NOW,
    });
    const capabilityProvider =
      createPresignedCapabilityHttpProvider({
        endpoint: CAPABILITY_ENDPOINT,
        vault,
        allowedDataOrigins: [OBJECT_ORIGIN],
        allowedDataPathPrefixes: ["/multipart/"],
        allowedQueryParameters: ["signature"],
        allowedRequestHeaders: [
          "content-type",
          CHECKSUM_HEADER,
        ],
        allowedResponseHeaders: [POLICY_HEADER, "etag"],
        hardMaxTransferBytes: 64,
        hardMaxUploadResponseBytes: 16,
        maxCapabilityTtlMs: 60_000,
        minimumRemainingLifetimeMs: 100,
        timeoutMs: 5_000,
        fetcher,
        now: () => NOW,
      });
    const presignedExecutor = createPresignedTransferExecutor({
      vault,
      replayGuard: createSingleUsePresignedReplayGuard(),
      hardMaxTransferBytes: 64,
      hardMaxChunkBytes: 4,
      hardMaxUploadResponseBytes: 16,
      minimumRemainingLifetimeMs: 100,
      timeoutMs: 5_000,
      fetcher,
      now: () => NOW,
    });
    const transport = createResumableUploadFetchJsonTransport({
      endpoints: CONTROL_ENDPOINTS,
      allowedOrigins: [API_ORIGIN],
      credentials: "same-origin",
      fetcher,
      timeoutMs: 5_000,
      maxRequestBytes: 64 * 1024,
      maxResponseBytes: 64 * 1024,
    });
    const controlPlane = createResumableUploadHttpControlPlane({
      transport,
      partCapabilities: capabilityProvider,
    });
    const runtime = createResumableUploadRuntime({
      controlPlane,
      partExecutor: createPresignedUploadPartExecutor(
        presignedExecutor.uploadParts,
        () => NOW,
      ),
      checkpoints,
      mutationLock: noContentionLock,
      crypto,
      policy: {
        partSizeBytes: 4,
        maxFileBytes: 64,
        maxPartCount: 16,
        maxConcurrency: 2,
        maxInFlightBytes: 32,
        partBufferCopyFactor: 4,
        maxSourceChunkBytes: 4,
        maxRetries: 1,
        retryBaseDelayMs: 1,
        retryMaxDelayMs: 10,
        maxRetryAfterMs: 100,
        capabilityRefreshSkewMs: 100,
        maxSessionLifetimeMs: 120_000,
        providerAttemptTimeoutMs: 5_000,
      },
      now: () => NOW,
      random: () => 0,
      sleep: async () => {},
    });

    const result = await runtime.upload({
      uploadKey: "upload_key_integration",
      purpose: "attachment",
      mediaType: "application/octet-stream",
      source: rangeSource(
        new Uint8Array([1, 2, 3, 4, 5, 6]),
      ),
      signal: activeSignal,
    });

    expect(result).toMatchObject({
      ok: true,
      value: {
        state: "QUARANTINED",
        resourceId: "resource_integration_01",
        byteLength: 6,
        replayed: false,
      },
    });
    expect(issuedBindings).toHaveLength(2);
    expect(
      issuedBindings.every(
        (binding) =>
          binding.protocol === RESUMABLE_UPLOAD_PROTOCOL &&
          binding.sessionId === "session_integration_01" &&
          binding.requestBindingSha256 ===
            session?.requestBindingSha256,
      ),
    ).toBe(true);
    expect(objectPutOptions).toHaveLength(2);
    expect(
      objectPutOptions.every(
        (options) =>
          options.method === "PUT" &&
          options.credentials === "omit" &&
          options.redirect === "error",
      ),
    ).toBe(true);
    expect(
      completedParts[0]?.map((part) => part.partNumber),
    ).toEqual([1, 2]);
    expect(checkpoints.rows.size).toBe(0);
  });
});