scaffold multipart video upload primitives

src/lib/media/video/multipart/aggregateProgress.test.ts

@@ -0,0 +1,37 @@
+import {createProgressAggregator} from './aggregateProgress'
+
+describe('createProgressAggregator', () => {
+  it('sums bytes across parts against the total', () => {
+    const progress: number[] = []
+    const report = createProgressAggregator(100, p => progress.push(p))
+
+    report(1, 50)
+    report(2, 25)
+    expect(progress).toEqual([0.5, 0.75])
+  })
+
+  it('overwrites a part running count rather than adding it', () => {
+    const progress: number[] = []
+    const report = createProgressAggregator(100, p => progress.push(p))
+
+    report(1, 20)
+    report(1, 40)
+    expect(progress).toEqual([0.2, 0.4])
+  })
+
+  it('clamps to 1', () => {
+    const progress: number[] = []
+    const report = createProgressAggregator(100, p => progress.push(p))
+
+    report(1, 150)
+    expect(progress).toEqual([1])
+  })
+
+  it('reports 0 when the total is 0', () => {
+    const progress: number[] = []
+    const report = createProgressAggregator(0, p => progress.push(p))
+
+    report(1, 10)
+    expect(progress).toEqual([0])
+  })
+})

src/lib/media/video/multipart/aggregateProgress.ts

@@ -0,0 +1,19 @@
+/**
+ * Tracks bytes sent per part and reports overall progress (0..1) to the
+ * existing single-value progress callback. Parts upload concurrently, so each
+ * part reports its own running byte count and this sums them against the total.
+ */
+export function createProgressAggregator(
+  totalBytes: number,
+  setProgress: (progress: number) => void,
+) {
+  const sentByPart = new Map<number, number>()
+  return function reportPartProgress(partNumber: number, bytesSent: number) {
+    sentByPart.set(partNumber, bytesSent)
+    let sum = 0
+    for (const value of sentByPart.values()) {
+      sum += value
+    }
+    setProgress(totalBytes > 0 ? Math.min(sum / totalBytes, 1) : 0)
+  }
+}

src/lib/media/video/multipart/constants.ts

@@ -0,0 +1,19 @@
+/*
+ * Multipart upload knobs. Tunable; part size must stay above the storage
+ * backend's minimum (R2/S3 require >= 5 MiB per part, except the last).
+ */
+
+/** Target size per part. */
+export const MULTIPART_PART_SIZE = 8 * 1024 * 1024 // 8 MiB
+
+/**
+ * Files below this skip multipart and use the single-shot POST. A file that
+ * would be a single part gains nothing from the multipart machinery.
+ */
+export const MULTIPART_MIN_FILE_SIZE = 16 * 1024 * 1024 // 16 MiB
+
+/** Max parts uploaded concurrently. */
+export const MULTIPART_CONCURRENCY = 3
+
+/** Per-part upload attempts before the part (and the upload) fails. */
+export const MULTIPART_MAX_ATTEMPTS = 3

src/lib/media/video/multipart/planParts.test.ts

@@ -0,0 +1,40 @@
+import {planParts} from './planParts'
+
+describe('planParts', () => {
+  it('splits an evenly divisible size into full parts', () => {
+    expect(planParts(20, 10)).toEqual([
+      {partNumber: 1, offset: 0, size: 10},
+      {partNumber: 2, offset: 10, size: 10},
+    ])
+  })
+
+  it('puts the remainder in the last part', () => {
+    expect(planParts(25, 10)).toEqual([
+      {partNumber: 1, offset: 0, size: 10},
+      {partNumber: 2, offset: 10, size: 10},
+      {partNumber: 3, offset: 20, size: 5},
+    ])
+  })
+
+  it('returns a single part when the file is smaller than a part', () => {
+    expect(planParts(5, 10)).toEqual([{partNumber: 1, offset: 0, size: 5}])
+  })
+
+  it('returns no parts for a non-positive size', () => {
+    expect(planParts(0, 10)).toEqual([])
+  })
+
+  it('covers the whole file with no gaps or overlaps', () => {
+    const parts = planParts(1000, 128)
+    expect(parts[0].offset).toBe(0)
+    for (let i = 1; i < parts.length; i++) {
+      expect(parts[i].offset).toBe(parts[i - 1].offset + parts[i - 1].size)
+    }
+    const last = parts[parts.length - 1]
+    expect(last.offset + last.size).toBe(1000)
+  })
+
+  it('throws for a non-positive part size', () => {
+    expect(() => planParts(100, 0)).toThrow()
+  })
+})

src/lib/media/video/multipart/planParts.ts

@@ -0,0 +1,22 @@
+import {type PartPlan} from './types'
+
+/**
+ * Splits a file of `totalSize` bytes into parts of at most `partSize` bytes.
+ * The last part carries the remainder. Part numbers are 1-indexed. Returns an
+ * empty list for a non-positive size.
+ */
+export function planParts(totalSize: number, partSize: number): PartPlan[] {
+  if (partSize <= 0) {
+    throw new Error('partSize must be positive')
+  }
+  const parts: PartPlan[] = []
+  let offset = 0
+  let partNumber = 1
+  while (offset < totalSize) {
+    const size = Math.min(partSize, totalSize - offset)
+    parts.push({partNumber, offset, size})
+    offset += size
+    partNumber += 1
+  }
+  return parts
+}

src/lib/media/video/multipart/readChunk.ts

@@ -0,0 +1,21 @@
+import {File} from 'expo-file-system'
+
+import {type CompressedVideo} from '#/lib/media/video/types'
+import {type ChunkReader} from './types'
+
+/**
+ * Native chunk reader. Opens one file handle and seeks per read, so the video
+ * bytes are never all held in JS memory. Call `close` when the upload finishes.
+ */
+export function createChunkReader(video: CompressedVideo): ChunkReader {
+  const handle = new File(video.uri).open()
+  return {
+    read(offset, size) {
+      handle.offset = offset
+      return Promise.resolve(handle.readBytes(size))
+    },
+    close() {
+      handle.close()
+    },
+  }
+}

src/lib/media/video/multipart/readChunk.web.ts

@@ -0,0 +1,27 @@
+import {type CompressedVideo} from '#/lib/media/video/types'
+import {type ChunkReader} from './types'
+
+/**
+ * Web chunk reader. Web compression already produces the full buffer, so this
+ * slices it in memory. Falls back to fetching the uri once if `bytes` is
+ * missing. `close` is a no-op.
+ */
+export function createChunkReader(video: CompressedVideo): ChunkReader {
+  let bytesPromise: Promise<ArrayBuffer> | null = null
+  const getBytes = () => {
+    if (video.bytes) {
+      return Promise.resolve(video.bytes)
+    }
+    if (!bytesPromise) {
+      bytesPromise = fetch(video.uri).then(res => res.arrayBuffer())
+    }
+    return bytesPromise
+  }
+  return {
+    async read(offset, size) {
+      const buffer = await getBytes()
+      return new Uint8Array(buffer, offset, size)
+    },
+    close() {},
+  }
+}

src/lib/media/video/multipart/types.ts

@@ -0,0 +1,42 @@
+/**
+ * One part of a multipart upload. `partNumber` is 1-indexed to match the S3
+ * convention the backend uses.
+ */
+export type PartPlan = {
+  partNumber: number
+  offset: number
+  size: number
+}
+
+/**
+ * Result of uploading a single part. `etag` is the value the storage backend
+ * (R2) returns for the part; the complete request lists these back to assemble
+ * the object.
+ */
+export type PartUploadResult = {
+  partNumber: number
+  etag: string
+}
+
+/**
+ * Reads a byte range off the compressed video. Native opens a file handle and
+ * seeks; web slices an in-memory buffer. `close` releases the native handle and
+ * is a no-op on web.
+ */
+export type ChunkReader = {
+  read: (offset: number, size: number) => Promise<Uint8Array>
+  close: () => void
+}
+
+/**
+ * Uploads one part to storage and resolves with its ETag. This is the only
+ * protocol-specific seam - for the presigned R2 approach it does a direct PUT
+ * to the presigned URL and reads the ETag off the response. Injected into the
+ * orchestrator so the transport can be filled in once the backend lands.
+ */
+export type UploadPartFn = (args: {
+  part: PartPlan
+  chunk: Uint8Array
+  onProgress: (bytesSent: number) => void
+  signal: AbortSignal
+}) => Promise<PartUploadResult>

src/lib/media/video/multipart/uploadParts.test.ts

@@ -0,0 +1,128 @@
+import {type ChunkReader, type UploadPartFn} from './types'
+import {uploadParts} from './uploadParts'
+
+function fakeReader(): ChunkReader {
+  return {
+    read: (_offset, size) => Promise.resolve(new Uint8Array(size)),
+    close: () => {},
+  }
+}
+
+const parts = [
+  {partNumber: 1, offset: 0, size: 10},
+  {partNumber: 2, offset: 10, size: 10},
+  {partNumber: 3, offset: 20, size: 5},
+]
+
+describe('uploadParts', () => {
+  it('uploads every part and returns results ordered by part number', async () => {
+    const uploadPart: UploadPartFn = ({part}) =>
+      Promise.resolve({
+        partNumber: part.partNumber,
+        etag: `etag-${part.partNumber}`,
+      })
+
+    const results = await uploadParts({
+      parts,
+      reader: fakeReader(),
+      uploadPart,
+      totalBytes: 25,
+      setProgress: () => {},
+      signal: new AbortController().signal,
+    })
+
+    expect(results.map(r => r.partNumber)).toEqual([1, 2, 3])
+    expect(results.map(r => r.etag)).toEqual(['etag-1', 'etag-2', 'etag-3'])
+  })
+
+  it('respects the concurrency cap', async () => {
+    let active = 0
+    let maxActive = 0
+    const uploadPart: UploadPartFn = async ({part}) => {
+      active++
+      maxActive = Math.max(maxActive, active)
+      await new Promise(r => setTimeout(r, 5))
+      active--
+      return {partNumber: part.partNumber, etag: `e${part.partNumber}`}
+    }
+
+    await uploadParts({
+      parts,
+      reader: fakeReader(),
+      uploadPart,
+      totalBytes: 25,
+      setProgress: () => {},
+      signal: new AbortController().signal,
+      concurrency: 2,
+    })
+
+    expect(maxActive).toBeLessThanOrEqual(2)
+  })
+
+  it('retries a failing part and succeeds', async () => {
+    const attemptsByPart = new Map<number, number>()
+    const uploadPart: UploadPartFn = ({part}) => {
+      const n = (attemptsByPart.get(part.partNumber) ?? 0) + 1
+      attemptsByPart.set(part.partNumber, n)
+      if (part.partNumber === 2 && n === 1) {
+        return Promise.reject(new Error('transient'))
+      }
+      return Promise.resolve({
+        partNumber: part.partNumber,
+        etag: `e${part.partNumber}`,
+      })
+    }
+
+    const results = await uploadParts({
+      parts,
+      reader: fakeReader(),
+      uploadPart,
+      totalBytes: 25,
+      setProgress: () => {},
+      signal: new AbortController().signal,
+    })
+
+    expect(attemptsByPart.get(2)).toBe(2)
+    expect(results).toHaveLength(3)
+  })
+
+  it('throws after exhausting attempts', async () => {
+    const uploadPart: UploadPartFn = () =>
+      Promise.reject(new Error('always fails'))
+
+    await expect(
+      uploadParts({
+        parts,
+        reader: fakeReader(),
+        uploadPart,
+        totalBytes: 25,
+        setProgress: () => {},
+        signal: new AbortController().signal,
+        maxAttempts: 2,
+      }),
+    ).rejects.toThrow('always fails')
+  })
+
+  it('reports progress that reaches 1 when all parts complete', async () => {
+    const progress: number[] = []
+    const uploadPart: UploadPartFn = ({part, chunk, onProgress}) => {
+      onProgress(chunk.byteLength)
+      return Promise.resolve({
+        partNumber: part.partNumber,
+        etag: `e${part.partNumber}`,
+      })
+    }
+
+    await uploadParts({
+      parts,
+      reader: fakeReader(),
+      uploadPart,
+      totalBytes: 25,
+      setProgress: p => progress.push(p),
+      signal: new AbortController().signal,
+      concurrency: 1,
+    })
+
+    expect(progress[progress.length - 1]).toBe(1)
+  })
+})

src/lib/media/video/multipart/uploadParts.ts

@@ -0,0 +1,120 @@
+import {AbortError} from '#/lib/async/cancelable'
+import {createProgressAggregator} from './aggregateProgress'
+import {MULTIPART_CONCURRENCY, MULTIPART_MAX_ATTEMPTS} from './constants'
+import {
+  type ChunkReader,
+  type PartPlan,
+  type PartUploadResult,
+  type UploadPartFn,
+} from './types'
+
+/**
+ * Uploads every part with a concurrency cap and per-part retry, aggregating
+ * byte progress into `setProgress`. Reads each chunk lazily just before its
+ * upload so only `concurrency` chunks are in memory at once. Resolves with the
+ * part results ordered by part number, ready for the complete request.
+ *
+ * `uploadPart` is the transport seam - it stays stubbed until the presigned R2
+ * PUT is wired up.
+ */
+export async function uploadParts({
+  parts,
+  reader,
+  uploadPart,
+  totalBytes,
+  setProgress,
+  signal,
+  concurrency = MULTIPART_CONCURRENCY,
+  maxAttempts = MULTIPART_MAX_ATTEMPTS,
+}: {
+  parts: PartPlan[]
+  reader: ChunkReader
+  uploadPart: UploadPartFn
+  totalBytes: number
+  setProgress: (progress: number) => void
+  signal: AbortSignal
+  concurrency?: number
+  maxAttempts?: number
+}): Promise<PartUploadResult[]> {
+  const reportPartProgress = createProgressAggregator(totalBytes, setProgress)
+  const results: PartUploadResult[] = new Array(parts.length)
+
+  let nextIndex = 0
+  async function worker() {
+    while (true) {
+      if (signal.aborted) {
+        throw new AbortError()
+      }
+      const index = nextIndex++
+      if (index >= parts.length) {
+        return
+      }
+      const part = parts[index]
+      const chunk = await reader.read(part.offset, part.size)
+      results[index] = await uploadPartWithRetry({
+        part,
+        chunk,
+        uploadPart,
+        maxAttempts,
+        signal,
+        onProgress: bytesSent => reportPartProgress(part.partNumber, bytesSent),
+      })
+    }
+  }
+
+  const workers = Array.from(
+    {length: Math.min(concurrency, parts.length)},
+    () => worker(),
+  )
+  await Promise.all(workers)
+  return results
+}
+
+async function uploadPartWithRetry({
+  part,
+  chunk,
+  uploadPart,
+  maxAttempts,
+  signal,
+  onProgress,
+}: {
+  part: PartPlan
+  chunk: Uint8Array
+  uploadPart: UploadPartFn
+  maxAttempts: number
+  signal: AbortSignal
+  onProgress: (bytesSent: number) => void
+}): Promise<PartUploadResult> {
+  let lastError: unknown
+  for (let attempt = 1; attempt <= maxAttempts; attempt++) {
+    if (signal.aborted) {
+      throw new AbortError()
+    }
+    try {
+      return await uploadPart({part, chunk, onProgress, signal})
+    } catch (err) {
+      if (signal.aborted) {
+        throw new AbortError()
+      }
+      lastError = err
+      if (attempt < maxAttempts) {
+        await delay(500 * 2 ** (attempt - 1), signal)
+      }
+    }
+  }
+  throw lastError
+}
+
+function delay(ms: number, signal: AbortSignal) {
+  return new Promise<void>((resolve, reject) => {
+    const timer = setTimeout(() => {
+      signal.removeEventListener('abort', onAbort)
+      resolve()
+    }, ms)
+    function onAbort() {
+      clearTimeout(timer)
+      reject(new AbortError())
+    }
+    signal.addEventListener('abort', onAbort, {once: true})
+  })
+}