import type { MediaUpload } from "@rakazo/contracts"; import { Sha256 } from "./media-sha256"; /* * Resumable chunked upload of one device original (spec 2.7). The server owns * the protocol: `uploads.begin` opens or resumes an upload for the same asset * and content, chunks go to `PUT /api/media/uploads/:id/chunks/:index` with * their exact length, `uploads.status.receivedChunks` says what already * arrived, and an upload idle for too long answers 409 "Upload expired" and * must begin again. This module is the driver; the native glue injects I/O. */ /** A failed byte route or RPC with the HTTP status the server answered. */ export class MediaUploadHttpError extends Error { constructor( readonly status: number, message: string, ) { super(message); this.name = "MediaUploadHttpError"; } } /** The local original no longer matches what the upload was opened for. */ export class MediaFileChangedError extends Error { constructor() { super("The file changed while reading it"); this.name = "MediaFileChangedError"; } } export function chunkCount(sizeBytes: number, chunkBytes: number): number { if (!Number.isSafeInteger(sizeBytes) || sizeBytes <= 0) throw new Error("Invalid upload size"); if (!Number.isSafeInteger(chunkBytes) || chunkBytes <= 0) throw new Error("Invalid chunk size"); return Math.ceil(sizeBytes / chunkBytes); } /** Byte range of chunk `index`: every chunk is full except the last. */ export function chunkRange( index: number, sizeBytes: number, chunkBytes: number, ): { start: number; length: number } { const count = chunkCount(sizeBytes, chunkBytes); if (!Number.isInteger(index) || index < 0 || index >= count) { throw new Error("Chunk index out of range"); } const start = index * chunkBytes; return { start, length: Math.min(chunkBytes, sizeBytes - start) }; } /** Chunk indexes the server has not confirmed yet, in upload order. */ export function missingChunks( upload: Pick, ) { const received = new Set(upload.receivedChunks); const missing: number[] = []; for (let index = 0; index < chunkCount(upload.sizeBytes, upload.chunkBytes); index++) { if (!received.has(index)) missing.push(index); } return missing; } export function isUploadExpired(error: unknown): boolean { return ( error instanceof MediaUploadHttpError && error.status === 409 && /expired/i.test(error.message) ); } /** Network failures and server hiccups are worth another try; client errors are not. */ export function isRetryableUploadError(error: unknown): boolean { if (error instanceof MediaUploadHttpError) { return error.status === 408 || error.status === 429 || error.status >= 500; } if (error instanceof MediaFileChangedError) return false; return !(error instanceof Error && error.name === "AbortError"); } /** Bytes of one chunk, backed by a plain ArrayBuffer so fetch can send them. */ export type ChunkBytes = Uint8Array; /** Reads `length` bytes at `start` from the original. */ export type ReadChunk = (start: number, length: number) => Promise | ChunkBytes; export async function hashOriginal( sizeBytes: number, read: ReadChunk, options: { readBytes?: number; signal?: AbortSignal; onProgress?: (fraction: number) => void; } = {}, ): Promise { const step = options.readBytes ?? 4 * 1024 * 1024; const hash = new Sha256(); for (let start = 0; start < sizeBytes; start += step) { options.signal?.throwIfAborted(); const length = Math.min(step, sizeBytes - start); const bytes = await read(start, length); if (bytes.byteLength !== length) throw new MediaFileChangedError(); hash.update(bytes); options.onProgress?.(Math.min(1, (start + length) / sizeBytes)); } return hash.hex(); } export type UploadTransport = { begin(input: { assetId: string; sizeBytes: number; sha256: string; fileName: string; }): Promise; status(uploadId: string): Promise; putChunk( uploadId: string, index: number, bytes: ChunkBytes, signal?: AbortSignal, ): Promise; complete(uploadId: string): Promise; }; export type UploadOriginalInput = { assetId: string; fileName: string; sizeBytes: number; sha256: string; read: ReadChunk; transport: UploadTransport; signal?: AbortSignal; /** Fraction of bytes the server confirmed, 0..1. */ onProgress?: (fraction: number) => void; /** Consecutive failed attempts before giving up. */ maxAttempts?: number; /** How often an expired upload may start over. */ maxRestarts?: number; wait?: (ms: number, signal?: AbortSignal) => Promise; }; const defaultWait = (ms: number, signal?: AbortSignal) => new Promise((resolve, reject) => { const timer = setTimeout(resolve, ms); signal?.addEventListener( "abort", () => { clearTimeout(timer); reject(signal.reason); }, { once: true }, ); }); export function retryDelayMs(attempt: number): number { return Math.min(30_000, 1_000 * 2 ** Math.max(0, attempt - 1)); } function confirmedFraction(upload: MediaUpload): number { const count = chunkCount(upload.sizeBytes, upload.chunkBytes); let bytes = 0; for (const index of new Set(upload.receivedChunks)) { if (index >= 0 && index < count) { bytes += chunkRange(index, upload.sizeBytes, upload.chunkBytes).length; } } return bytes / upload.sizeBytes; } /** * Upload one original and complete it. Resumes from the server's * `receivedChunks` after any failure, begins again when the upload expired, * and backs off between retries. Returns the completed upload. */ export async function uploadOriginal(input: UploadOriginalInput): Promise { const { transport, signal } = input; const maxAttempts = input.maxAttempts ?? 5; const maxRestarts = input.maxRestarts ?? 2; const wait = input.wait ?? defaultWait; const begin = () => transport.begin({ assetId: input.assetId, sizeBytes: input.sizeBytes, sha256: input.sha256, fileName: input.fileName, }); let restarts = 0; let failures = 0; let upload = await begin(); for (;;) { signal?.throwIfAborted(); try { if (upload.status === "complete") { input.onProgress?.(1); return upload; } if (upload.status === "failed") { throw new MediaUploadHttpError(409, "Upload expired"); } if (upload.sizeBytes !== input.sizeBytes) throw new MediaFileChangedError(); input.onProgress?.(confirmedFraction(upload)); for (const index of missingChunks(upload)) { signal?.throwIfAborted(); const range = chunkRange(index, upload.sizeBytes, upload.chunkBytes); const bytes = await input.read(range.start, range.length); if (bytes.byteLength !== range.length) throw new MediaFileChangedError(); upload = await transport.putChunk(upload.id, index, bytes, signal); failures = 0; input.onProgress?.(confirmedFraction(upload)); } upload = await transport.complete(upload.id); if (upload.status !== "complete") throw new Error("Upload did not complete"); input.onProgress?.(1); return upload; } catch (error) { if (signal?.aborted) throw signal.reason ?? error; if (isUploadExpired(error)) { if (restarts >= maxRestarts) throw error; restarts += 1; failures = 0; upload = await begin(); continue; } if (!isRetryableUploadError(error)) throw error; failures += 1; if (failures >= maxAttempts) throw error; await wait(retryDelayMs(failures), signal); // Ask the server what arrived; a chunk may have landed before the error. try { upload = await transport.status(upload.id); } catch (statusError) { if (isUploadExpired(statusError)) { if (restarts >= maxRestarts) throw statusError; restarts += 1; upload = await begin(); } else if (!isRetryableUploadError(statusError)) { throw statusError; } } } } } /** Overall progress of several uploads, weighted by size. */ export function combinedProgress(items: Array<{ sizeBytes: number; fraction: number }>): number { const total = items.reduce((sum, item) => sum + item.sizeBytes, 0); if (total === 0) return 0; return items.reduce((sum, item) => sum + item.sizeBytes * item.fraction, 0) / total; }