import { createHash, type Hash } from "node:crypto"; import { sniffMime, SNIFF_BYTES, type SniffedMime } from "./mime.js"; export const MAX_ATTACHMENT_BYTES = 20 * 1024 * 1024; /** Non-retryable payload problems: the outbox row is marked failed immediately. */ export class PermanentArchiveError extends Error {} export interface InspectedPayload { sha256: string; byteSize: number; sniffed: SniffedMime; } /** * Incremental size/hash/magic-byte inspection. Bytes pass straight through to * the upload; nothing is buffered beyond the first SNIFF_BYTES. */ export class PayloadInspector { private readonly hash: Hash = createHash("sha256"); private readonly head = new Uint8Array(SNIFF_BYTES); private headLength = 0; private total = 0; constructor( private readonly declaredLength: number | null, private readonly maxBytes = MAX_ATTACHMENT_BYTES, ) { if (declaredLength !== null && declaredLength > maxBytes) { throw new PermanentArchiveError(`declared size ${declaredLength} exceeds ${maxBytes} bytes`); } } update(chunk: Uint8Array): void { this.total += chunk.byteLength; if (this.total > this.maxBytes) throw new PermanentArchiveError(`payload exceeds ${this.maxBytes} bytes`); if (this.declaredLength !== null && this.total > this.declaredLength) { throw new PermanentArchiveError("payload longer than Content-Length"); } if (this.headLength < SNIFF_BYTES) { const take = chunk.subarray(0, SNIFF_BYTES - this.headLength); this.head.set(take, this.headLength); this.headLength += take.byteLength; } this.hash.update(chunk); } finish(): InspectedPayload { if (this.declaredLength !== null && this.total !== this.declaredLength) { throw new PermanentArchiveError("payload shorter than Content-Length"); } if (this.total === 0) throw new PermanentArchiveError("empty payload"); return { sha256: this.hash.digest("hex"), byteSize: this.total, sniffed: sniffMime(this.head.subarray(0, this.headLength)), }; } } /** * Wrap a byte stream so it is inspected in flight. Any violation errors the * stream, which aborts the consuming upload before it can complete. */ export function inspectStream( source: ReadableStream, declaredLength: number | null, maxBytes = MAX_ATTACHMENT_BYTES, ): { stream: ReadableStream; result: Promise } { const inspector = new PayloadInspector(declaredLength, maxBytes); let resolve!: (value: InspectedPayload) => void; let reject!: (reason: unknown) => void; const result = new Promise((res, rej) => { resolve = res; reject = rej; }); // The caller awaits the upload first; avoid an unhandled rejection in the meantime. result.catch(() => undefined); const transform = new TransformStream({ transform(chunk, controller) { try { inspector.update(chunk); } catch (err) { reject(err); throw err; } controller.enqueue(chunk); }, flush() { try { resolve(inspector.finish()); } catch (err) { reject(err); throw err; } }, }); return { stream: source.pipeThrough(transform), result }; } export function parseContentLength(header: string | null): number | null { if (header === null) return null; if (!/^\d+$/.test(header.trim())) throw new PermanentArchiveError("invalid Content-Length"); return Number(header.trim()); }