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> = {}, ): 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 = {}, ): 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("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 => { 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>; signal: AbortSignal; }>, ) => ReturnType; 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({ 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({ 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({ 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({ cancel: urlCancel }), { status: 200, headers: { "content-type": "application/json" }, }, ), responseAt( ENDPOINTS.GET_STATUS, new ReadableStream({ 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> >, }), ).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( name: string, options: Readonly<{ mode: "exclusive"; signal?: AbortSignal; }>, callback: (lock: unknown) => Promise, ): Promise { 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((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 }, }, ]); }); });