From 93edeab5ca9d0cc2a13110f488280b277ec93d40 Mon Sep 17 00:00:00 2001 From: glass beadian Date: Thu, 15 Oct 2026 22:28:45 +0000 Subject: [PATCH] feat: chunked video ingest API (init / chunk / complete) Cloudflare's proxy plan rejects bodies above ~100 MB, so whole-body video uploads cannot exceed it no matter the origin limits. Clients now slice into 10 MB chunks: POST /video/upload/init, PUT /video/upload/:id/chunk/:index, POST /video/upload/:id/complete reassembles to disk, seals, stores to R2 and runs the normal probe + job pipeline. Max chunked upload: 100 GiB. Tests 17/17. --- apps/api/src/app.module.ts | 3 +- apps/api/src/video/chunked-upload.test.ts | 327 ++++++++++++++++++++++ apps/api/src/video/chunked-upload.ts | 299 ++++++++++++++++++++ apps/api/src/video/upload-limits.ts | 39 +++ apps/api/src/video/video.controller.ts | 138 ++++++++- apps/api/src/video/video.service.ts | 42 ++- 6 files changed, 843 insertions(+), 5 deletions(-) create mode 100644 apps/api/src/video/chunked-upload.test.ts create mode 100644 apps/api/src/video/chunked-upload.ts diff --git a/apps/api/src/app.module.ts b/apps/api/src/app.module.ts index ee1dd29..985d1ff 100644 --- a/apps/api/src/app.module.ts +++ b/apps/api/src/app.module.ts @@ -20,6 +20,7 @@ import { GroupsController } from "./groups/groups.controller"; import { GroupsService } from "./groups/groups.service"; import { VideoController } from "./video/video.controller"; import { VideoService } from "./video/video.service"; +import { ChunkedUploadService } from "./video/chunked-upload.js"; import { LiveMonitorModule } from "./live-monitor/live-monitor.module"; import { MediaControlModule } from "./media-control/media-control.module"; import { ProductionModule } from "./production/production.module"; @@ -66,6 +67,6 @@ import { MediaModule } from "./media/media.module.js"; { provide: PublishingExecutionRepository, useFactory: () => new PublishingExecutionRepository(pool) }, TemporalWorkflowStarter, { provide: PUBLISHING_DEPENDENCIES, inject: [PublishingExecutionRepository, TemporalWorkflowStarter, ApprovalGateService], useFactory: (repository: PublishingExecutionRepository, workflowStarter: TemporalWorkflowStarter, approvalGate: ApprovalGateService) => ({ repository, workflowStarter, approvalGate, tenantTransaction: withTenant, assertReady: () => assertPublishingSchema(pool) }) }, - PublishingService, ApprovalGateService, AuthService, ConnectionsService, TeamService, InviteLinkService, AuditQueryService, VideoService, GroupsService, SnapshotService, SettingsService, BrandingService], + PublishingService, ApprovalGateService, AuthService, ConnectionsService, TeamService, InviteLinkService, AuditQueryService, VideoService, ChunkedUploadService, GroupsService, SnapshotService, SettingsService, BrandingService], }) export class AppModule {} diff --git a/apps/api/src/video/chunked-upload.test.ts b/apps/api/src/video/chunked-upload.test.ts new file mode 100644 index 0000000..f91a560 --- /dev/null +++ b/apps/api/src/video/chunked-upload.test.ts @@ -0,0 +1,327 @@ +import { mkdtemp, readFile, rm, stat, writeFile } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { + ChunkedUploadService, + chunkOffset, + type ChunkedUploadSession, +} from "./chunked-upload.js"; +import { + CHUNK_SIZE_BYTES, + MAX_CHUNKED_UPLOAD_BYTES, + chunkBytesFor, + totalChunksFor, +} from "./upload-limits.js"; +import { VideoService } from "./video.service.js"; +import { assetId, serviceWithDatabase, workspaceId, type FakeVideoDatabase } from "./test-fixtures.js"; + +const otherWorkspace = "90000000-0000-4000-8000-00000000000f"; +let directory: string; +let service: ChunkedUploadService; +let counter = 0; + +beforeEach(async () => { + directory = await mkdtemp(join(tmpdir(), "chunked-test-")); + counter += 1; + service = new ChunkedUploadService( + directory, + Number.MAX_SAFE_INTEGER, + () => `00000000-0000-4000-8000-${String(counter).padStart(12, "0")}`, + ); +}); + +afterEach(async () => { + await service.onModuleDestroy(); + await rm(directory, { recursive: true, force: true }); +}); + +function init(totalBytes: number, overrides: Partial<{ filename: string; mimeType: string }> = {}): Promise { + return service.init(workspaceId, { + filename: overrides.filename ?? "raw-capture.mp4", + mimeType: overrides.mimeType ?? "video/mp4", + totalBytes, + }); +} + +/** Deterministic pseudo-random chunk bytes; offset-tagged so mis-writes are visible. */ +function chunkBytes(totalBytes: number, index: number): Buffer { + return Buffer.alloc(chunkBytesFor(totalBytes, index), (index % 251) + 1); +} + +describe("chunk plan math (pure)", () => { + it("splits files into 10 MiB chunks with a short final chunk", () => { + expect(CHUNK_SIZE_BYTES).toBe(10 * 1024 * 1024); + expect(totalChunksFor(CHUNK_SIZE_BYTES)).toBe(1); + expect(totalChunksFor(CHUNK_SIZE_BYTES + 1)).toBe(2); + expect(totalChunksFor(3 * CHUNK_SIZE_BYTES)).toBe(3); + expect(totalChunksFor(3 * CHUNK_SIZE_BYTES + 123)).toBe(4); + // 20 GB target size stays within the 100 GiB ceiling. + expect(totalChunksFor(20 * 1024 ** 3)).toBe(2048); + expect(MAX_CHUNKED_UPLOAD_BYTES).toBe(100 * 1024 ** 3); + }); + + it("computes chunk sizes and byte offsets", () => { + const total = 2 * CHUNK_SIZE_BYTES + 5; + expect(chunkBytesFor(total, 0)).toBe(CHUNK_SIZE_BYTES); + expect(chunkBytesFor(total, 1)).toBe(CHUNK_SIZE_BYTES); + expect(chunkBytesFor(total, 2)).toBe(5); + expect(chunkOffset(0)).toBe(0); + expect(chunkOffset(1)).toBe(CHUNK_SIZE_BYTES); + expect(chunkOffset(2047)).toBe(2047 * CHUNK_SIZE_BYTES); + expect(() => chunkBytesFor(total, 3)).toThrow("outside the 3-chunk upload"); + expect(() => chunkOffset(-1)).toThrow("non-negative"); + }); +}); + +describe("init", () => { + it("creates a 0600 temp file under /tmp/uploads/.tmp and tracks the session", async () => { + const session = await init(2 * CHUNK_SIZE_BYTES); + expect(session.uploadId).toMatch(/^[0-9a-f-]{36}$/); + expect(session.totalChunks).toBe(2); + expect(session.tempPath).toBe(join(directory, `${session.uploadId}.tmp`)); + expect((await stat(session.tempPath)).size).toBe(0); + expect((await stat(session.tempPath)).mode & 0o777).toBe(0o600); + expect(service.activeSessions.get(session.uploadId)).toBe(session); + expect(session.workspaceId).toBe(workspaceId); + }); + + it("rejects non-video MIME types, bad filenames, and bad byte counts", async () => { + await expect(init(1, { mimeType: "image/png" })).rejects.toMatchObject({ status: 400 }); + await expect(init(1, { filename: " " })).rejects.toMatchObject({ status: 400 }); + await expect(init(0)).rejects.toMatchObject({ status: 400, message: expect.stringContaining("positive safe integer") }); + await expect(init(1.5)).rejects.toMatchObject({ status: 400 }); + await expect(init(Number.MAX_SAFE_INTEGER + 1)).rejects.toMatchObject({ status: 400 }); + // Spec ceiling: totalBytes ≤ 100 GiB. + await expect(init(MAX_CHUNKED_UPLOAD_BYTES + 1)).rejects.toMatchObject({ + status: 400, + message: expect.stringContaining("100 GiB"), + }); + }); +}); + +describe("appendChunk — correct offset writes", () => { + it("writes chunks at index × 10 MiB regardless of arrival order (sparse middle-out)", async () => { + const total = 3 * CHUNK_SIZE_BYTES; + await init(total); + const uploadId = firstUploadId(); + // Deliberately out of order: last, first, middle. + const third = chunkBytes(total, 2); + await service.appendChunk(uploadId, workspaceId, 2, third); + await service.appendChunk(uploadId, workspaceId, 0, chunkBytes(total, 0)); + const receipt = await service.appendChunk(uploadId, workspaceId, 1, chunkBytes(total, 1)); + expect(receipt).toEqual({ received: true, bytesSoFar: total }); + const assembled = await readFile(join(directory, `${uploadId}.tmp`)); + expect(assembled.length).toBe(total); + expect(assembled.subarray(0, CHUNK_SIZE_BYTES)).toEqual(chunkBytes(total, 0)); + expect(assembled.subarray(CHUNK_SIZE_BYTES, 2 * CHUNK_SIZE_BYTES)).toEqual(chunkBytes(total, 1)); + expect(assembled.subarray(2 * CHUNK_SIZE_BYTES)).toEqual(third); + }); + + it("fills holes with zero bytes before their chunk arrives, and reports progressive bytesSoFar", async () => { + const total = 3 * CHUNK_SIZE_BYTES; + await init(total); + const uploadId = firstUploadId(); + const afterFirst = await service.appendChunk(uploadId, workspaceId, 0, chunkBytes(total, 0)); + expect(afterFirst.bytesSoFar).toBe(CHUNK_SIZE_BYTES); + const afterLast = await service.appendChunk(uploadId, workspaceId, 2, chunkBytes(total, 2)); + expect(afterLast.bytesSoFar).toBe(2 * CHUNK_SIZE_BYTES); + // Chunk 1 has not arrived: its range must read as zeros, not garbage. + const partial = await readFile(join(directory, `${uploadId}.tmp`)); + expect(partial.subarray(CHUNK_SIZE_BYTES, 2 * CHUNK_SIZE_BYTES)).toEqual(Buffer.alloc(CHUNK_SIZE_BYTES, 0)); + // The file is already full length (sparse). + expect(partial.length).toBe(total); + }); + + it("is idempotent for a retransmitted chunk and rejects wrong-sized bodies", async () => { + const total = 2 * CHUNK_SIZE_BYTES; + await init(total); + const uploadId = firstUploadId(); + await service.appendChunk(uploadId, workspaceId, 0, chunkBytes(total, 0)); + await service.appendChunk(uploadId, workspaceId, 0, chunkBytes(total, 0)); + expect((await service.verifyComplete(uploadId, workspaceId)).bytesSoFar).toBe(CHUNK_SIZE_BYTES); + // Too short. + await expect(service.appendChunk(uploadId, workspaceId, 1, Buffer.alloc(CHUNK_SIZE_BYTES - 1, 7))) + .rejects.toMatchObject({ status: 400, message: expect.stringContaining("exactly") }); + // Too long. + await expect(service.appendChunk(uploadId, workspaceId, 1, Buffer.alloc(CHUNK_SIZE_BYTES + 1, 7))) + .rejects.toMatchObject({ status: 400 }); + // Out of range index. + await expect(service.appendChunk(uploadId, workspaceId, 2, Buffer.alloc(1))) + .rejects.toMatchObject({ status: 400, message: expect.stringContaining("between 0 and 1") }); + await expect(service.appendChunk(uploadId, workspaceId, -1, Buffer.alloc(1))).rejects.toMatchObject({ status: 400 }); + }); + + it("enforces tenant isolation and rejects unknown/finished uploads", async () => { + const session = await init(CHUNK_SIZE_BYTES); + await expect(service.appendChunk(session.uploadId, otherWorkspace, 0, Buffer.alloc(CHUNK_SIZE_BYTES))) + .rejects.toMatchObject({ status: 404 }); + await expect(service.appendChunk("not-a-uuid", workspaceId, 0, Buffer.alloc(1))).rejects.toMatchObject({ status: 400 }); + await expect(service.appendChunk("11111111-1111-4111-8111-111111111111", workspaceId, 0, Buffer.alloc(1))) + .rejects.toMatchObject({ status: 404 }); + await service.appendChunk(session.uploadId, workspaceId, 0, Buffer.alloc(CHUNK_SIZE_BYTES)); + service.complete(session.uploadId); + await expect(service.appendChunk(session.uploadId, workspaceId, 0, Buffer.alloc(CHUNK_SIZE_BYTES))) + .rejects.toMatchObject({ status: 409 }); + expect(service.activeSessions.has(session.uploadId)).toBe(true); + }); +}); + +describe("verifyComplete / assertAssembled — missing chunk + size verification", () => { + it("reports missing chunks and under-length files until every chunk lands", async () => { + const total = 3 * CHUNK_SIZE_BYTES; + await init(total); + const uploadId = firstUploadId(); + const first = await service.verifyComplete(uploadId, workspaceId); + expect(first.missing).toEqual([0, 1, 2]); + expect(first.bytesSoFar).toBe(0); + // Sparse file is already totalBytes long on disk, so size alone would lie — + // the chunk-count check is what catches the gap. + expect(first.fileBytes).toBe(0); // nothing written yet + expect(() => service.assertAssembled(first)).toThrow("missing 3 chunk(s)"); + + await service.appendChunk(uploadId, workspaceId, 1, chunkBytes(total, 1)); + const middle = await service.verifyComplete(uploadId, workspaceId); + expect(middle.missing).toEqual([0, 2]); + expect(() => service.assertAssembled(middle)).toThrow("missing 2 chunk(s): 0, 2"); + + await service.appendChunk(uploadId, workspaceId, 0, chunkBytes(total, 0)); + await service.appendChunk(uploadId, workspaceId, 2, chunkBytes(total, 2)); + const done = await service.verifyComplete(uploadId, workspaceId); + expect(done.missing).toEqual([]); + expect(done.bytesSoFar).toBe(total); + expect(done.fileBytes).toBe(total); + expect(() => service.assertAssembled(done)).not.toThrow(); + }); + + it("detects a declared size that disagrees with the assembled file", async () => { + const total = 2 * CHUNK_SIZE_BYTES; + const session = await init(total); + await service.appendChunk(session.uploadId, workspaceId, 0, chunkBytes(total, 0)); + await service.appendChunk(session.uploadId, workspaceId, 1, chunkBytes(total, 1)); + // Simulate a truncated/oversized file behind the assembler's back. + await writeFile(session.tempPath, Buffer.alloc(total - 1, 9)); + const check = await service.verifyComplete(session.uploadId, workspaceId); + expect(check.missing).toEqual([]); + expect(check.fileBytes).toBe(total - 1); + expect(() => service.assertAssembled(check)).toThrow("bytes on disk"); + expect(() => service.assertAssembled(check)).toThrow(`expected ${total}`); + }); + + it("detects byte-accounting drift when a received set over/under-counts totalBytes", async () => { + const total = 2 * CHUNK_SIZE_BYTES; + const session = await init(total); + session.received.add(0); + session.received.add(1); + session.received.add(1); // Set: no duplicate effect, but simulate accounting drift below. + const check = await service.verifyComplete(session.uploadId, workspaceId); + expect(check.bytesSoFar).toBe(2 * CHUNK_SIZE_BYTES); + // Forcing a mismatched declaration is caught by the service-level check. + const drifted: typeof check = { ...check, session: { ...session, totalBytes: total + 5 } }; + expect(() => service.assertAssembled(drifted)).toThrow(/received \d+ bytes but declared/); + }); +}); + +describe("abort, cleanup, and expiry sweeper", () => { + it("abort removes the temp file and session, and is idempotent per-uploadId", async () => { + const session = await init(CHUNK_SIZE_BYTES); + await service.appendChunk(session.uploadId, workspaceId, 0, Buffer.alloc(CHUNK_SIZE_BYTES)); + await service.abort(session.uploadId, workspaceId); + await expect(stat(session.tempPath)).rejects.toMatchObject({ code: "ENOENT" }); + expect(service.activeSessions.has(session.uploadId)).toBe(false); + await expect(service.abort(session.uploadId, workspaceId)).rejects.toMatchObject({ status: 404 }); + await expect(service.verifyComplete(session.uploadId, workspaceId)).rejects.toMatchObject({ status: 404 }); + }); + + it("cleanup drops the entry after complete; failure paths can still see the file", async () => { + const session = await init(CHUNK_SIZE_BYTES); + await service.appendChunk(session.uploadId, workspaceId, 0, Buffer.alloc(CHUNK_SIZE_BYTES)); + service.complete(session.uploadId); + await service.cleanup(session.uploadId); + await expect(stat(session.tempPath)).rejects.toMatchObject({ code: "ENOENT" }); + expect(service.activeSessions.has(session.uploadId)).toBe(false); + await expect(service.cleanup(session.uploadId)).resolves.toBeUndefined(); + }); + + it("sweeps sessions idle past the 24h TTL along with their temp files", async () => { + const session = await init(CHUNK_SIZE_BYTES); + await service.appendChunk(session.uploadId, workspaceId, 0, Buffer.alloc(CHUNK_SIZE_BYTES)); + const now = Date.now(); + // Fresh session is retained. + expect(await service.sweepExpired(now)).toEqual([]); + // 24h minus a second: still retained. + expect(await service.sweepExpired(now + 24 * 60 * 60 * 1000 - 1000)).toEqual([]); + expect(service.activeSessions.has(session.uploadId)).toBe(true); + // TTL reached: swept. + const swept = await service.sweepExpired(now + 24 * 60 * 60 * 1000 + 1); + expect(swept).toEqual([session.uploadId]); + await expect(stat(session.tempPath)).rejects.toMatchObject({ code: "ENOENT" }); + expect(service.activeSessions.has(session.uploadId)).toBe(false); + }); +}); + +describe("end-to-end assemble → seal → register → probe (VideoService tail end)", () => { + it("uploadVideoFromPath stores, probes, and returns the same VideoUploadResult shape as POST /video/upload", async () => { + const database: FakeVideoDatabase = {}; + const bytes = Buffer.concat([Buffer.from("MP4DATA-"), Buffer.alloc(512, 3)]); + const probe = vi.fn(async (path: string) => { + expect(await readFile(path)).toEqual(bytes); + expect(database.job?.state).toBe("probing"); + return { width: 1920, height: 1080, durationMs: 90_000, frameRate: 24, codec: "hevc" }; + }); + const { service: video, storage, appendAudit } = serviceWithDatabase(database, probe); + + // Assemble through the real chunk pipeline (small single-chunk file). + const session = await service.init(workspaceId, { + filename: "capture.mp4", + mimeType: "video/mp4", + totalBytes: bytes.length, + }); + await service.appendChunk(session.uploadId, workspaceId, 0, bytes); + const check = await service.verifyComplete(session.uploadId, workspaceId); + expect(() => service.assertAssembled(check)).not.toThrow(); + + const uploaded = await video.uploadVideoFromPath(workspaceId, session.tempPath, "video/mp4", bytes.length); + expect(storage.putContent).toHaveBeenCalledWith(bytes, "video/mp4"); + expect(uploaded.videoAsset).toMatchObject({ + storageKey: expect.stringMatching(/^sha256\/[a-f0-9]{64}\.mp4$/), + sizeBytes: bytes.length, + width: 1920, + height: 1080, + durationMs: 90_000, + codec: "hevc", + }); + // Identical shape/outcome to the single-shot path: a completed probe job. + expect(uploaded.processingJob).toMatchObject({ state: "complete", progressPercent: 100, outputKey: uploaded.videoAsset.storageKey }); + expect(database.transitions).toEqual(["queued", "probing", "transcoding", "complete"]); + expect(appendAudit).toHaveBeenCalledWith(workspaceId, expect.objectContaining({ action: "video.uploaded", targetId: assetId })); + // Cleanup is the caller's (controller's) job, verified separately. + expect(probe).toHaveBeenCalledTimes(1); + }); + + it("rejects a declared size that disagrees with the file on disk", async () => { + const { service: video } = serviceWithDatabase({}, async () => ({})); + const path = join(directory, "mismatch.tmp"); + await writeFile(path, Buffer.alloc(16, 1)); + await expect(video.uploadVideoFromPath(workspaceId, path, "video/mp4", 15)) + .rejects.toMatchObject({ status: 400, message: expect.stringContaining("16 bytes on disk but declared 15") }); + await expect(video.uploadVideoFromPath(workspaceId, join(directory, "missing.tmp"), "video/mp4", 16)) + .rejects.toMatchObject({ status: 400, message: expect.stringContaining("missing on disk") }); + }); + + it("refuses non-video MIME types and empty files", async () => { + const { service: video } = serviceWithDatabase({}, async () => ({})); + const path = join(directory, "notvideo.tmp"); + await writeFile(path, Buffer.alloc(4, 1)); + await expect(video.uploadVideoFromPath(workspaceId, path, "image/png", 4)) + .rejects.toMatchObject({ status: 400, message: expect.stringContaining("video media type") }); + await expect(video.uploadVideoFromPath(workspaceId, path, "video/mp4", 0)) + .rejects.toMatchObject({ status: 400, message: expect.stringContaining("positive safe integer") }); + }); +}); + +function firstUploadId(): string { + const ids = [...service.activeSessions.keys()]; + if (ids.length !== 1) throw new Error(`expected exactly one session, found ${ids.length}`); + return ids[0] as string; +} diff --git a/apps/api/src/video/chunked-upload.ts b/apps/api/src/video/chunked-upload.ts new file mode 100644 index 0000000..00bbadc --- /dev/null +++ b/apps/api/src/video/chunked-upload.ts @@ -0,0 +1,299 @@ +import { randomUUID } from "node:crypto"; +import { mkdir, open, rm, stat } from "node:fs/promises"; +import { join } from "node:path"; +import { tmpdir } from "node:os"; +import { BadRequestException, ConflictException, Injectable, NotFoundException, OnModuleDestroy } from "@nestjs/common"; +import { + CHUNK_SIZE_BYTES, + CHUNKED_UPLOAD_TTL_MS, + MAX_CHUNKED_UPLOAD_BYTES, + chunkBytesFor, + totalChunksFor, +} from "./upload-limits.js"; + +/** + * Chunked (resumable-style) large-file ingest. + * + * Cloudflare's free plan rejects proxied request bodies above 100 MB, so a 20 GB + * camera original can never arrive in one POST /video/upload. The client instead + * slices the file into CHUNK_SIZE_BYTES (10 MiB) chunks — each far under the + * limit — and PUTs them one at a time. Each chunk is written at its fixed offset + * (`index × 10 MiB`) into a sparse temp file, so chunks may arrive in ANY order + * and may be retransmitted after a network failure without corrupting the file. + * + * On complete, the assembled file goes through the SAME SPENC1 seal → S3 + * putContent path as every other upload. + * + * ── MEMORY REQUIREMENT (read this) ──────────────────────────────────────────── + * The seal function (apps/api/src/storage/storage.ts) takes a whole Buffer, and + * S3 PutObject is likewise handed one body. Assembling therefore peaks at + * roughly 2× file size of RAM: one Buffer of plaintext for the seal plus the + * sealed copy the S3 client transmits (plus ~1× transiently for + * Buffer.concat in seal). There is no way to bound that without streaming + * encryption + multipart S3, which would change the storage contract for every + * other caller — out of scope here. + * + * Practical sizing: a 20 GB upload needs a host with ≥ 48 GB RAM + * (MAX_OLD_SPACE_SIZE raised, or --max_old-space-size=32768+, plus headroom for + * the sealed copy). Up to ~2 GiB the ordinary multipart path is fine and cheaper. + * Disk also matters: /tmp must hold the assembled file (see SP_CHUNK_UPLOAD_DIR, + * default /tmp/uploads) on top of whatever S3 buffers. + * ───────────────────────────────────────────────────────────────────────────── + */ + +export interface ChunkedUploadSession { + uploadId: string; + /** Owning workspace; chunks and complete are rejected for other tenants. */ + workspaceId: string; + filename: string; + mimeType: string; + totalBytes: number; + totalChunks: number; + /** Indices of chunks whose bytes are durably on disk. */ + received: Set; + createdAt: number; + updatedAt: number; + /** Set once /complete has been accepted; the entry is dropped afterwards. */ + completed?: boolean; + tempPath: string; +} + +export interface InitChunkedUploadRequest { + filename: string; + mimeType: string; + totalBytes: number; +} + +export interface ChunkReceipt { + received: true; + bytesSoFar: number; +} + +export interface CompletionCheck { + session: ChunkedUploadSession; + /** Missing chunk indices, ascending — empty when the upload is complete. */ + missing: number[]; + /** Bytes durably written so far (sum of received chunk sizes). */ + bytesSoFar: number; + /** Size of the assembled file on disk (sparse holes read as zero bytes). */ + fileBytes: number; +} + +/** A pure function that makes the on-disk layout testable without touching a disk. */ +export function chunkOffset(index: number, chunkSize: number = CHUNK_SIZE_BYTES): number { + if (!Number.isSafeInteger(index) || index < 0) throw new Error("index must be a non-negative integer"); + return index * chunkSize; +} + +/** Max chunk byte length a client may PUT for the given session. */ +export function maxChunkBytesFor(totalBytes: number, index: number, chunkSize: number = CHUNK_SIZE_BYTES): number { + return chunkBytesFor(totalBytes, index, chunkSize); +} + +const UUID_PATTERN = /^[0-9a-f]{8}-[0-9a-f]{4}-[1-8][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/i; + +@Injectable() +export class ChunkedUploadService implements OnModuleDestroy { + /** uploadId → session. Module-level by design: one assembler per API process. */ + private readonly sessions = new Map(); + private sweeper: NodeJS.Timeout | undefined; + + constructor( + private readonly directory: string = process.env.SP_CHUNK_UPLOAD_DIR ?? join(tmpdir(), "uploads"), + private readonly sweeperIntervalMs: number = 30 * 60 * 1000, + private readonly generateId: () => string = randomUUID, + ) { + // Only arm the sweeper when running under a live event loop (not unit tests + // that construct the service directly and would hold the process open). + if (process.env.VITEST === undefined) { + this.sweeper = setInterval(() => void this.sweepExpired(), this.sweeperIntervalMs); + this.sweeper.unref?.(); + } + } + + /** Test seam: the live sessions map. */ + get activeSessions(): Map { + return this.sessions; + } + + async init(workspaceId: string, request: InitChunkedUploadRequest): Promise { + const filename = request.filename.trim(); + if (filename === "" || filename.length > 512) { + throw new BadRequestException("filename must be a non-empty string of at most 512 characters"); + } + const mimeType = request.mimeType.trim(); + if (!mimeType.toLowerCase().startsWith("video/")) { + throw new BadRequestException("mimeType must be a video media type"); + } + if (!Number.isSafeInteger(request.totalBytes) || request.totalBytes < 1) { + throw new BadRequestException("totalBytes must be a positive safe integer"); + } + if (request.totalBytes > MAX_CHUNKED_UPLOAD_BYTES) { + throw new BadRequestException("totalBytes exceeds the 100 GiB chunked upload limit"); + } + const totalChunks = totalChunksFor(request.totalBytes); + const uploadId = this.generateId(); + await mkdir(this.directory, { recursive: true }); + const tempPath = join(this.directory, `${uploadId}.tmp`); + // Create the file eagerly (and exclusively per uploadId) so a later chunk + // never races a first-time create, and so the path is known to be writable. + const handle = await open(tempPath, "wx", 0o600).catch((error: NodeJS.ErrnoException) => { + if (error.code === "EEXIST") throw new ConflictException("temp upload path already exists; retry init"); + throw error; + }); + await handle.close(); + const now = Date.now(); + const session: ChunkedUploadSession = { + uploadId, + workspaceId, + filename, + mimeType, + totalBytes: request.totalBytes, + totalChunks, + received: new Set(), + createdAt: now, + updatedAt: now, + tempPath, + }; + this.sessions.set(uploadId, session); + return session; + } + + async appendChunk(uploadId: string, workspaceId: string, index: number, bytes: Buffer): Promise { + const session = this.requireSession(uploadId, workspaceId); + if (!Number.isSafeInteger(index) || index < 0 || index >= session.totalChunks) { + throw new BadRequestException(`index must be an integer between 0 and ${session.totalChunks - 1}`); + } + const expected = chunkBytesFor(session.totalBytes, index); + if (bytes.length !== expected) { + throw new BadRequestException( + `chunk ${index} must be exactly ${expected} bytes (got ${bytes.length}); the final chunk carries the remainder`, + ); + } + // Positional write at index × 10 MiB: arrival order does not matter and a + // retransmitted chunk is idempotent (same bytes, same offset). + const handle = await open(session.tempPath, "r+").catch((error: NodeJS.ErrnoException) => { + if (error.code === "ENOENT") { + throw new NotFoundException(`upload ${uploadId} temp file is gone; start a new upload`); + } + throw error; + }); + try { + const { bytesWritten } = await handle.write(bytes, 0, bytes.length, chunkOffset(index)); + if (bytesWritten !== expected) { + throw new Error(`short write for chunk ${index}: ${bytesWritten} of ${expected} bytes`); + } + // Durability before acknowledgement: a chunk reported received must + // actually be on disk if the API restarts mid-upload. + await handle.datasync(); + } finally { + await handle.close(); + } + session.received.add(index); + session.updatedAt = Date.now(); + return { received: true, bytesSoFar: this.bytesSoFar(session) }; + } + + /** + * Verifies every chunk landed before ingest: exact chunk-count check against + * the session, byte accounting against the assembled file on disk, and — the + * authoritative check — that the file size equals the declared totalBytes. + */ + async verifyComplete(uploadId: string, workspaceId: string): Promise { + const session = this.requireSession(uploadId, workspaceId); + const missing: number[] = []; + for (let index = 0; index < session.totalChunks; index += 1) { + if (!session.received.has(index)) missing.push(index); + } + const fileBytes = await stat(session.tempPath).then((info) => info.size).catch(() => -1); + return { session, missing, bytesSoFar: this.bytesSoFar(session), fileBytes }; + } + + /** Size verification for tests / callers: the file must be exactly totalBytes. */ + assertAssembled(check: CompletionCheck): void { + if (check.missing.length > 0) { + throw new BadRequestException( + `upload ${check.session.uploadId} is missing ${check.missing.length} chunk(s): ${describeIndices(check.missing)}`, + ); + } + if (check.bytesSoFar !== check.session.totalBytes) { + throw new BadRequestException( + `upload ${check.session.uploadId} received ${check.bytesSoFar} bytes but declared ${check.session.totalBytes}`, + ); + } + if (check.fileBytes !== check.session.totalBytes) { + throw new BadRequestException( + `assembled file for ${check.session.uploadId} is ${check.fileBytes} bytes on disk; expected ${check.session.totalBytes}`, + ); + } + } + + /** Marks the session consumed so no further chunk can be appended. */ + complete(uploadId: string): void { + const session = this.sessions.get(uploadId); + if (session === undefined) throw new NotFoundException(`upload ${uploadId} not found`); + session.completed = true; + } + + /** Removes the temp file and the session entry. Idempotent (abort + sweep). */ + async abort(uploadId: string, workspaceId: string): Promise { + const session = this.requireSession(uploadId, workspaceId); + this.sessions.delete(uploadId); + await rm(session.tempPath, { force: true }); + } + + /** Drops the temp file and entry after complete; keeps failures visible for retry. */ + async cleanup(uploadId: string): Promise { + const session = this.sessions.get(uploadId); + if (session === undefined) return; + this.sessions.delete(uploadId); + await rm(session.tempPath, { force: true }); + } + + /** Removes sessions idle past the TTL along with their temp files. */ + async sweepExpired(now: number = Date.now()): Promise { + const expired: string[] = []; + for (const [uploadId, session] of this.sessions) { + if (now - session.updatedAt <= CHUNKED_UPLOAD_TTL_MS) continue; + expired.push(uploadId); + this.sessions.delete(uploadId); + await rm(session.tempPath, { force: true }); + } + return expired; + } + + async onModuleDestroy(): Promise { + if (this.sweeper !== undefined) clearInterval(this.sweeper); + // In-flight sessions are resumable in principle, but the process is going + // away: drop the entries so nothing can half-append later. + for (const [uploadId, session] of this.sessions) { + this.sessions.delete(uploadId); + await rm(session.tempPath, { force: true }); + } + } + + private requireSession(uploadId: string, workspaceId: string): ChunkedUploadSession { + if (!UUID_PATTERN.test(uploadId)) throw new BadRequestException("uploadId must be a UUID"); + const session = this.sessions.get(uploadId); + if (session === undefined) throw new NotFoundException(`upload ${uploadId} not found or expired`); + if (session.workspaceId !== workspaceId) { + // Do not leak existence across tenants. + throw new NotFoundException(`upload ${uploadId} not found or expired`); + } + if (session.completed === true) { + throw new ConflictException(`upload ${uploadId} has already been completed`); + } + return session; + } + + private bytesSoFar(session: ChunkedUploadSession): number { + let total = 0; + for (const index of session.received) total += chunkBytesFor(session.totalBytes, index); + return total; + } +} + +function describeIndices(indices: number[]): string { + const shown = indices.slice(0, 8).map(String).join(", "); + return indices.length > 8 ? `${shown}, … (+${indices.length - 8} more)` : shown; +} diff --git a/apps/api/src/video/upload-limits.ts b/apps/api/src/video/upload-limits.ts index 0cfa028..c0207a6 100644 --- a/apps/api/src/video/upload-limits.ts +++ b/apps/api/src/video/upload-limits.ts @@ -1,2 +1,41 @@ // StorageService seals whole Buffers; bound multipart memory use per request. export const MAX_VIDEO_UPLOAD_BYTES = 2 * 1024 * 1024 * 1024; + +/** + * Chunked ingest: POST /video/upload/init → PUT /video/upload/:uploadId/chunk/:index + * → POST /video/upload/:uploadId/complete (see chunked-upload.ts). + * + * Cloudflare's free plan rejects proxied request bodies above 100 MB, so the + * client slices any file into CHUNK_SIZE_BYTES chunks — each comfortably under + * the limit — and the server reassembles them on disk before the normal + * SPENC1 seal → S3 putContent path. This removes the practical size limit: + * only MAX_CHUNKED_UPLOAD_BYTES (and server RAM, documented in chunked-upload.ts) + * bounds an upload. + */ +export const CHUNK_SIZE_BYTES = 10 * 1024 * 1024; + +/** Hard ceiling for a single chunked upload (spec: validate totalBytes ≤ 100 GiB). */ +export const MAX_CHUNKED_UPLOAD_BYTES = 100 * 1024 * 1024 * 1024; + +/** An upload session that receives nothing for this long is swept (file + entry). */ +export const CHUNKED_UPLOAD_TTL_MS = 24 * 60 * 60 * 1000; + +/** Number of 10 MiB chunks a client must send for a file of totalBytes. */ +export function totalChunksFor(totalBytes: number, chunkSize: number = CHUNK_SIZE_BYTES): number { + if (!Number.isSafeInteger(totalBytes) || totalBytes < 1) { + throw new Error("totalBytes must be a positive safe integer"); + } + return Math.ceil(totalBytes / chunkSize); +} + +/** + * Exact byte length of chunk `index`: every chunk is a full `chunkSize` except + * the last, which carries the remainder. Enforced on the server so the + * assembled file can never be short or over-long. + */ +export function chunkBytesFor(totalBytes: number, index: number, chunkSize: number = CHUNK_SIZE_BYTES): number { + if (!Number.isSafeInteger(index) || index < 0) throw new Error("index must be a non-negative integer"); + const totalChunks = totalChunksFor(totalBytes, chunkSize); + if (index >= totalChunks) throw new Error(`index ${index} is outside the ${totalChunks}-chunk upload`); + return Math.min(chunkSize, totalBytes - index * chunkSize); +} diff --git a/apps/api/src/video/video.controller.ts b/apps/api/src/video/video.controller.ts index 5fd8233..01468d2 100644 --- a/apps/api/src/video/video.controller.ts +++ b/apps/api/src/video/video.controller.ts @@ -6,7 +6,9 @@ import { Get, Inject, Param, + ParseIntPipe, Post, + Put, Query, Req, UseGuards, @@ -25,7 +27,17 @@ import { type VideoUploadResult, } from "./video.service.js"; import { parseProduceRequest } from "./produce-request.js"; -import { MAX_VIDEO_UPLOAD_BYTES } from "./upload-limits.js"; +import { + CHUNK_SIZE_BYTES, + MAX_CHUNKED_UPLOAD_BYTES, + MAX_VIDEO_UPLOAD_BYTES, + totalChunksFor, +} from "./upload-limits.js"; +import { + ChunkedUploadService, + type ChunkReceipt, + type InitChunkedUploadRequest, +} from "./chunked-upload.js"; import { loadDriveBytes, loadMatrixBytes, parseMxcUri } from "./import-sources.js"; interface VideoFile { @@ -42,7 +54,80 @@ interface UploadVideoDto { @Controller("video") @UseGuards(RbacGuard) export class VideoController { - constructor(@Inject(VideoService) private readonly video: VideoService) {} + constructor( + @Inject(VideoService) private readonly video: VideoService, + private readonly chunked: ChunkedUploadService, + ) {} + + /** + * Chunked ingest for files that cannot clear Cloudflare's 100 MB proxied-body + * cap (or the 2 GiB whole-body limit). The client slices the file into 10 MiB + * chunks and PUTs them sequentially; the server reassembles on disk, then runs + * the SAME SPENC1 seal → S3 putContent path as POST /video/upload. + * + * MEMORY: reading the assembled file for the seal peaks around 2× file size of + * RAM (see chunked-upload.ts) — a 20 GB upload needs a ≥ 48 GB host. + */ + @Post("upload/init") + @RequirePermission("media.manage") + async initChunkedUpload( + @Req() request: Request & AuthenticatedRequest, + @Body() body: unknown, + ): Promise<{ uploadId: string; chunkSizeBytes: number; totalChunks: number }> { + const session = requireSession(request); + const dto = parseInitUpload(body); + const created = await this.chunked.init(session.workspaceId, dto); + return { + uploadId: created.uploadId, + chunkSizeBytes: CHUNK_SIZE_BYTES, + totalChunks: created.totalChunks, + }; + } + + @Put("upload/:uploadId/chunk/:index") + @RequirePermission("media.manage") + async uploadChunk( + @Req() request: Request & AuthenticatedRequest, + @Param("uploadId") uploadId: string, + @Param("index", ParseIntPipe) index: number, + ): Promise { + const session = requireSession(request); + const bytes = await readRawBody(request); + return this.chunked.appendChunk(uploadId, session.workspaceId, index, bytes); + } + + @Post("upload/:uploadId/complete") + @RequirePermission("media.manage") + async completeChunkedUpload( + @Req() request: Request & AuthenticatedRequest, + @Param("uploadId") uploadId: string, + ): Promise { + const session = requireSession(request); + const check = await this.chunked.verifyComplete(uploadId, session.workspaceId); + this.chunked.assertAssembled(check); + this.chunked.complete(uploadId); + try { + return await this.video.uploadVideoFromPath( + session.workspaceId, + check.session.tempPath, + check.session.mimeType, + check.session.totalBytes, + ); + } finally { + // Never leave an assembled multi-GB temp file behind — success or failure. + await this.chunked.cleanup(uploadId); + } + } + + @Delete("upload/:uploadId") + @RequirePermission("media.manage") + async abortChunkedUpload( + @Req() request: Request & AuthenticatedRequest, + @Param("uploadId") uploadId: string, + ): Promise { + const session = requireSession(request); + await this.chunked.abort(uploadId, session.workspaceId); + } @Post("upload") @RequirePermission("media.manage") @@ -138,3 +223,52 @@ function parseUploadVideo(value: unknown): UploadVideoDto { sizeBytes, }; } + +/** + * Raw octet-stream body reader for chunk PUTs. Express leaves a non-multipart + * body unread until something consumes it; read exactly Content-Length bytes so + * a truncated client upload is rejected instead of silently padding the file. + */ +async function readRawBody(request: Request): Promise { + const declared = Number(request.headers["content-length"] ?? NaN); + const declaredBytes = Number.isSafeInteger(declared) ? declared : undefined; + const chunks: Buffer[] = []; + let received = 0; + for await (const part of request) { + const chunk = Buffer.isBuffer(part) ? part : Buffer.from(String(part)); + received += chunk.length; + if (declaredBytes !== undefined && received > declaredBytes) { + throw new BadRequestException("request body exceeded its Content-Length"); + } + if (received > CHUNK_SIZE_BYTES) { + throw new BadRequestException(`chunk exceeds the ${CHUNK_SIZE_BYTES} byte chunk size`); + } + chunks.push(chunk); + } + const body = Buffer.concat(chunks); + if (declaredBytes !== undefined && body.length !== declaredBytes) { + throw new BadRequestException("request body was shorter than its Content-Length"); + } + return body; +} + +function parseInitUpload(value: unknown): InitChunkedUploadRequest { + const body = requireObject(value, "body"); + const totalBytes = body.totalBytes; + if (typeof totalBytes !== "number" || !Number.isSafeInteger(totalBytes) || totalBytes < 1) { + throw new BadRequestException("totalBytes must be a positive safe integer"); + } + if (totalBytes > MAX_CHUNKED_UPLOAD_BYTES) { + throw new BadRequestException("totalBytes exceeds the 100 GiB chunked upload limit"); + } + return { + filename: requireString(body.filename, "filename"), + mimeType: requireString(body.mimeType, "mimeType"), + totalBytes, + }; +} + +/** Client-facing chunk math (kept server-side so a stale client cannot lie). */ +export function chunkPlan(totalBytes: number): { chunkSizeBytes: number; totalChunks: number } { + return { chunkSizeBytes: CHUNK_SIZE_BYTES, totalChunks: totalChunksFor(totalBytes) }; +} diff --git a/apps/api/src/video/video.service.ts b/apps/api/src/video/video.service.ts index 5e56a97..251c09b 100644 --- a/apps/api/src/video/video.service.ts +++ b/apps/api/src/video/video.service.ts @@ -1,4 +1,4 @@ -import { mkdtemp, rm, writeFile } from "node:fs/promises"; +import { mkdtemp, readFile, rm, stat, writeFile } from "node:fs/promises"; import { tmpdir } from "node:os"; import { join } from "node:path"; import { BadRequestException, Inject, Injectable, NotFoundException, Optional, ServiceUnavailableException } from "@nestjs/common"; @@ -8,7 +8,7 @@ import { withTenant } from "../db/client.js"; import { probeVideoFile, type VideoProbeResult } from "./ffprobe.js"; import { S3Storage, s3ConfigFromEnv, type StorageService } from "../storage/storage.js"; import { type ProduceRequest } from "./produce-request.js"; -import { MAX_VIDEO_UPLOAD_BYTES } from "./upload-limits.js"; +import { MAX_CHUNKED_UPLOAD_BYTES, MAX_VIDEO_UPLOAD_BYTES } from "./upload-limits.js"; const NO_OP_TRANSCODE_NOTE = "Transcode skipped: original stored object retained; rendering belongs to the media-worker compositor."; @@ -139,6 +139,44 @@ export class VideoService { return result; } + /** + * Chunked-ingest tail end: seal + store + probe an already-assembled file on + * local disk (see chunked-upload.ts). The bytes never sat in a request body, + * so the 2 GiB whole-body cap does not apply — MAX_CHUNKED_UPLOAD_BYTES does. + * + * MEMORY: reading the assembled file into one Buffer peaks around 2× file + * size of RAM (seal takes a Buffer and S3 takes the sealed copy). A 20 GB + * upload therefore needs a host with ≥ 48 GB RAM; /tmp must also hold the + * file itself. + */ + async uploadVideoFromPath( + workspaceId: string, + inputPath: string, + mimeType: string, + sizeBytes: number, + ): Promise { + validateUpload("pending", mimeType, sizeBytes); + if (sizeBytes > MAX_CHUNKED_UPLOAD_BYTES) { + throw new BadRequestException("video exceeds the 100 GiB chunked upload limit"); + } + const onDisk = await stat(inputPath).catch(() => { + throw new BadRequestException("assembled upload file is missing on disk"); + }); + if (onDisk.size !== sizeBytes) { + throw new BadRequestException( + `assembled upload is ${onDisk.size} bytes on disk but declared ${sizeBytes}`, + ); + } + const storage = this.requireStorage(); + const bytes = await readFile(inputPath).catch(() => { + throw new ServiceUnavailableException("could not read the assembled upload from disk"); + }); + const stored = await storage.putContent(bytes, mimeType).catch(() => { + throw new ServiceUnavailableException("Video storage unavailable: could not store encrypted upload in S3/MinIO."); + }); + return this.registerAndProbe(workspaceId, stored.key, stored.contentType, stored.sizeBytes); + } + async uploadVideo( workspaceId: string, storageKey: string,