clean-architecture-frontend-template/tests/unit/resumable-upload-runtime-fixture.ts

import { vi } from "vitest";

import type {
  ResumableUploadCheckpoint,
  ResumableUploadCheckpointStore,
  ResumableUploadControlPlane,
  ResumableUploadSource,
  UploadPartExecutor,
  UploadPartReceipt,
  UploadProviderResult,
  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 { resolveResumableUploadRuntimePolicy } from "../../src/adapters/browser-transfer/resumable-upload/runtime-policy.ts";
import type {
  UploadCancellationChannel,
  UploadCancellationListener,
} from "../../src/adapters/browser-transfer/resumable-upload/upload-cancellation-channel.ts";
import type { UploadMutationLock } from "../../src/adapters/browser-transfer/resumable-upload/upload-mutation-lock.ts";

export type TestCapability = Readonly<{ id: string }>;

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

export function createSerialMutationLock(): UploadMutationLock {
  let tail = Promise.resolve();
  return Object.freeze({
    run<Value>(
      _uploadKey: string,
      signal: AbortSignal,
      task: () => Promise<Value>,
    ): Promise<Value> {
      const result = tail.then(async () => {
        if (signal.aborted) {
          throw new DOMException(
            "The operation was aborted.",
            "AbortError",
          );
        }
        return await task();
      });
      tail = result.then(
        () => undefined,
        () => undefined,
      );
      return result;
    },
  });
}

export function createMemoryCancellationPair(): readonly [
  UploadCancellationChannel,
  UploadCancellationChannel,
] {
  const listeners = [
    new Set<UploadCancellationListener>(),
    new Set<UploadCancellationListener>(),
  ] as const;
  const channels = listeners.map((ownListeners, ownIndex) => {
    let closed = false;
    return Object.freeze({
      publish(uploadKey: string) {
        if (closed) return false;
        for (const [index, peerListeners] of listeners.entries()) {
          if (index === ownIndex) continue;
          for (const listener of [...peerListeners]) {
            listener(uploadKey);
          }
        }
        return true;
      },
      subscribe(listener: UploadCancellationListener) {
        if (closed) throw new TypeError("closed");
        ownListeners.add(listener);
        return () => ownListeners.delete(listener);
      },
      close() {
        closed = true;
        ownListeners.clear();
      },
    });
  });
  return channels as unknown as readonly [
    UploadCancellationChannel,
    UploadCancellationChannel,
  ];
}

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

  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 {
    this.closed = true;
  }
}

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

export function byteStreamSource(bytes: Uint8Array): ResumableUploadSource {
  return Object.freeze({
    kind: "FILE_BYTE_SOURCE" as const,
    bytes: Object.freeze({
      byteLength: bytes.byteLength,
      async *stream(signal: AbortSignal) {
        if (signal.aborted) {
          yield browserDataFailure("ABORTED", "FILE_READ");
          return;
        }
        yield browserDataSuccess(bytes.slice(0, 3));
        yield browserDataSuccess(bytes.slice(3));
      },
    }),
  });
}

export function runtimePolicy(
  overrides: Partial<
    Parameters<typeof resolveResumableUploadRuntimePolicy>[0]
  > = {},
) {
  return {
    partSizeBytes: 4,
    maxFileBytes: 100,
    maxPartCount: 25,
    maxConcurrency: 3,
    maxInFlightBytes: 48,
    partBufferCopyFactor: 4,
    maxSourceChunkBytes: 8,
    maxRetries: 2,
    retryBaseDelayMs: 1,
    retryMaxDelayMs: 10,
    maxRetryAfterMs: 100,
    capabilityRefreshSkewMs: 5,
    maxSessionLifetimeMs: 10_000,
    providerAttemptTimeoutMs: 100,
    ...overrides,
  };
}

export type ControlHarness = Readonly<{
  control: ResumableUploadControlPlane<TestCapability>;
  accepted: Map<number, UploadPartReceipt>;
  issued: ReturnType<typeof vi.fn>;
  completedParts: UploadPartReceipt[][];
  getSession(): UploadSession | null;
}>;

export function createControlHarness(options: Readonly<{
  now?: number;
  serverMaxConcurrency?: number;
  sessionId?: (createIndex: number) => string;
  statusParts?: (
    session: UploadSession,
    accepted: Map<number, UploadPartReceipt>,
  ) => readonly UploadPartReceipt[];
  issueCapability?: (
    input: Parameters<
      ResumableUploadControlPlane<TestCapability>["issuePartCapability"]
    >[0],
    callIndex: number,
  ) => UploadProviderResult<Readonly<{
    capability: TestCapability;
    uploadBindingSha256: string;
    expiresAtEpochMs: number;
  }>>;
}> = {}): ControlHarness {
  const now = options.now ?? 1_000;
  const accepted = new Map<number, UploadPartReceipt>();
  const completedParts: UploadPartReceipt[][] = [];
  let session: UploadSession | null = null;
  let createCount = 0;
  let issueCount = 0;
  const issued = vi.fn();
  const control: ResumableUploadControlPlane<TestCapability> = {
    async createSession(input) {
      createCount += 1;
      session = Object.freeze({
        protocol: RESUMABLE_UPLOAD_PROTOCOL,
        sessionId:
          options.sessionId?.(createCount) ?? "session_01",
        requestBindingSha256: input.requestBindingSha256,
        fingerprint: input.fingerprint,
        partSizeBytes: input.requestedPartSizeBytes,
        partCount: input.fingerprint.partCount,
        maxConcurrency: options.serverMaxConcurrency ?? 2,
        expiresAtEpochMs: now + 5_000,
      });
      return browserDataSuccess(session);
    },
    async getStatus() {
      if (!session) {
        return browserDataFailure("NOT_FOUND", "UPLOAD_RECONCILE");
      }
      const parts =
        options.statusParts?.(session, accepted) ??
        [...accepted.values()].sort(
          (left, right) => left.partNumber - right.partNumber,
        );
      return browserDataSuccess({
        state: "ACTIVE",
        session,
        acceptedParts: parts,
      });
    },
    async issuePartCapability(input) {
      issueCount += 1;
      issued(input);
      return (
        options.issueCapability?.(input, issueCount) ??
        browserDataSuccess({
          capability: Object.freeze({ id: `cap-${issueCount}` }),
          uploadBindingSha256: input.uploadBindingSha256,
          expiresAtEpochMs: now + 4_000,
        })
      );
    },
    async complete(input) {
      completedParts.push([...input.orderedParts]);
      return browserDataSuccess({
        state: "QUARANTINED",
        protocol: RESUMABLE_UPLOAD_PROTOCOL,
        sessionId: input.sessionId,
        requestBindingSha256: input.requestBindingSha256,
        fingerprint: input.fingerprint,
        resourceId: "resource_01",
      });
    },
    async abort() {
      return browserDataSuccess({ state: "ABORTED" });
    },
  };
  return {
    control,
    accepted,
    issued,
    completedParts,
    getSession: () => session,
  };
}

export function executorFor(
  harness: ControlHarness,
  options: Readonly<{
    delay?: () => Promise<void>;
    onActive?: (active: number) => void;
  }> = {},
): UploadPartExecutor<TestCapability> {
  let active = 0;
  return {
    async uploadPart(input) {
      active += 1;
      options.onActive?.(active);
      await options.delay?.();
      active -= 1;
      const receipt = Object.freeze({
        ...input.part,
        receiptToken: `etag-part-${input.part.partNumber}`,
      });
      harness.accepted.set(input.part.partNumber, receipt);
      return browserDataSuccess(receipt);
    },
  };
}