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

@@ -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 {}

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<ChunkedUploadSession> {
+  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/<uploadId>.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;
+}

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<number>;
+  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<string, ChunkedUploadSession>();
+  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<string, ChunkedUploadSession> {
+    return this.sessions;
+  }
+
+  async init(workspaceId: string, request: InitChunkedUploadRequest): Promise<ChunkedUploadSession> {
+    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<number>(),
+      createdAt: now,
+      updatedAt: now,
+      tempPath,
+    };
+    this.sessions.set(uploadId, session);
+    return session;
+  }
+
+  async appendChunk(uploadId: string, workspaceId: string, index: number, bytes: Buffer): Promise<ChunkReceipt> {
+    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<CompletionCheck> {
+    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<void> {
+    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<void> {
+    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<string[]> {
+    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<void> {
+    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;
+}

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);
+}

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<ChunkReceipt> {
+    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<VideoUploadResult> {
+    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<void> {
+    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<Buffer> {
+  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) };
+}

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<VideoUploadResult> {
+    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,