feat(archive): queue streaming archive jobs

e2e/archive.spec.ts

@@ -0,0 +1,111 @@
+import { randomBytes } from 'node:crypto'
+import { expect, type Page, test } from '@playwright/test'
+import { DirType } from '../shared/constants'
+import type { BackgroundJob, PaginatedResponse, StorageObject } from '../shared/types'
+import { signUpAndGoToFiles } from './helpers'
+
+const textType = 'text/plain'
+const fixtureSize = 6 * 1024 * 1024
+
+test.describe('Archive jobs with queued streaming workers @all', () => {
+  test.setTimeout(120_000)
+
+  test('compresses and extracts through the background queue', async ({ page }) => {
+    await signUpAndGoToFiles(page)
+    await seedFile(page, 'alpha.txt', randomBytes(fixtureSize))
+    await seedFile(page, 'beta.txt', randomBytes(fixtureSize))
+    await page.reload()
+
+    await selectFile(page, 'alpha.txt')
+    await selectFile(page, 'beta.txt')
+    await expect(page.getByTestId('files-toolbar-selection')).toContainText('2 selected')
+
+    const [compressResponse] = await Promise.all([
+      page.waitForResponse((response) => isBackgroundJobPost(response.url(), response.request().method())),
+      page.getByTitle('Compress').click(),
+    ])
+    expect(compressResponse.ok()).toBe(true)
+    const compressJob = (await compressResponse.json()) as BackgroundJob
+    expect(compressJob.status).toBe('queued')
+    await expect(page.getByText('Background task created')).toBeVisible()
+    await expect(page.getByRole('link', { name: /tasks/i })).toContainText('1')
+
+    await expectJobCompleted(page, compressJob.id)
+    await page.goto('/tasks')
+    await page.getByRole('button', { name: 'Completed' }).click()
+    await expect(page.getByText('selection.zip')).toBeVisible()
+
+    await page.goto('/files')
+    await expect(page.getByRole('cell', { name: 'selection.zip' })).toBeVisible()
+
+    const [extractResponse] = await Promise.all([
+      page.waitForResponse((response) => isBackgroundJobPost(response.url(), response.request().method())),
+      openRowAction(page, 'selection.zip', 'Extract'),
+    ])
+    expect(extractResponse.ok()).toBe(true)
+    const extractJob = (await extractResponse.json()) as BackgroundJob
+    expect(extractJob.status).toBe('queued')
+    await expect(page.getByRole('link', { name: /tasks/i })).toContainText('1')
+
+    await expectJobCompleted(page, extractJob.id)
+    await page.goto('/files')
+    await expect(page.getByRole('cell', { name: 'alpha (1).txt' })).toBeVisible()
+    await expect(page.getByRole('cell', { name: 'beta (1).txt' })).toBeVisible()
+  })
+})
+
+async function seedFile(page: Page, name: string, bytes: Buffer) {
+  const draftResponse = await page.request.post('/api/objects', {
+    data: {
+      name,
+      type: textType,
+      size: bytes.byteLength,
+      parent: '',
+      dirtype: DirType.FILE,
+    },
+  })
+  expect(draftResponse.ok()).toBe(true)
+  const draft = (await draftResponse.json()) as StorageObject & { uploadUrl: string }
+  expect(draft.uploadUrl).toBeTruthy()
+
+  const uploadResponse = await page.request.put(draft.uploadUrl, {
+    headers: { 'Content-Type': textType },
+    data: bytes,
+  })
+  expect(uploadResponse.ok()).toBe(true)
+
+  const confirmResponse = await page.request.patch(`/api/objects/${draft.id}`, {
+    data: { action: 'confirm' },
+  })
+  expect(confirmResponse.ok()).toBe(true)
+}
+
+async function selectFile(page: Page, name: string) {
+  const row = page.getByRole('row').filter({ hasText: name })
+  await expect(row).toBeVisible()
+  await row.getByRole('checkbox').check()
+}
+
+async function openRowAction(page: Page, fileName: string, action: string) {
+  const row = page.getByRole('row').filter({ hasText: fileName })
+  await row.getByRole('button').last().click()
+  await page.getByRole('menuitem', { name: action }).click()
+}
+
+async function expectJobCompleted(page: Page, jobId: string): Promise<BackgroundJob> {
+  const deadline = Date.now() + 60_000
+  while (Date.now() < deadline) {
+    const response = await page.request.get('/api/background-jobs?page=1&pageSize=20')
+    expect(response.ok()).toBe(true)
+    const body = (await response.json()) as PaginatedResponse<BackgroundJob>
+    const job = body.items.find((item) => item.id === jobId)
+    if (job?.status === 'completed') return job
+    if (job?.status === 'failed') throw new Error(job.errorMessage ?? 'Archive job failed')
+    await page.waitForTimeout(500)
+  }
+  throw new Error(`Timed out waiting for archive job ${jobId}`)
+}
+
+function isBackgroundJobPost(url: string, method: string) {
+  return method === 'POST' && url.includes('/api/background-jobs')
+}

e2e/global-setup.ts

@@ -9,15 +9,16 @@ import { hashPassword } from '../server/lib/password'
 import { ADMIN_EMAIL, ADMIN_PASSWORD } from './helpers'
 
 const localBaseUrl = process.env.E2E_LOCAL_BASE_URL ?? 'http://localhost:5173'
+const defaultOrgQuota = process.env.E2E_DEFAULT_ORG_QUOTA ?? String(1024 * 1024 * 1024)
 
 const storageConfig = {
   title: 'E2E Storage',
   mode: 'private',
-  bucket: 'e2e-test',
-  endpoint: 'https://localhost:9000',
-  region: 'auto',
-  accessKey: 'e2e-access-key',
-  secretKey: 'e2e-secret-key',
+  bucket: process.env.E2E_STORAGE_BUCKET ?? 'e2e-test',
+  endpoint: process.env.E2E_STORAGE_ENDPOINT ?? 'https://localhost:9000',
+  region: process.env.E2E_STORAGE_REGION ?? 'auto',
+  accessKey: process.env.E2E_STORAGE_ACCESS_KEY ?? 'e2e-access-key',
+  secretKey: process.env.E2E_STORAGE_SECRET_KEY ?? 'e2e-secret-key',
   capacity: 0,
   status: 'active',
 }
@@ -69,11 +70,18 @@ function prepareNodeDatabase() {
     .prepare(
       `
         INSERT INTO system_options (key, value, public)
-        VALUES (?, ?, 0), (?, ?, 0)
+        VALUES (?, ?, 0), (?, ?, 0), (?, ?, 0)
         ON CONFLICT(key) DO UPDATE SET value = excluded.value
       `,
     )
-    .run('cloud_store_created_at', new Date().toISOString(), 'cloud_store_updated_at', new Date().toISOString())
+    .run(
+      'cloud_store_created_at',
+      new Date().toISOString(),
+      'cloud_store_updated_at',
+      new Date().toISOString(),
+      'default_org_quota',
+      defaultOrgQuota,
+    )
 
   sqlite
     .prepare(
@@ -194,6 +202,12 @@ setup('seed admin and storage', async () => {
       }
     }
 
+    const quotaResp = await request.put('/api/system/options/default_org_quota', {
+      headers,
+      data: { value: defaultOrgQuota },
+    })
+    if (!quotaResp.ok()) throw new Error(`could not set E2E default quota: ${quotaResp.status()}`)
+
     if (ensureNodeStorage()) return
 
     // E2E specs rely on self-service sign-up to create isolated users. Force the

package.json

@@ -32,7 +32,9 @@
     "format": "biome format --write .",
     "e2e": "playwright test",
     "e2e:cloud": "node scripts/run-cloud-e2e.mjs",
-    "e2e:cloud:cf": "node scripts/run-cloud-e2e.mjs --runtime cf"
+    "e2e:cloud:cf": "node scripts/run-cloud-e2e.mjs --runtime cf",
+    "e2e:archive": "node scripts/run-cloud-e2e.mjs --local --with-s3-mock --spec archive.spec.ts",
+    "e2e:archive:cf": "node scripts/run-cloud-e2e.mjs --runtime cf --local --with-s3-mock --spec archive.spec.ts"
   },
   "engines": {
     "node": ">=24"

playwright.config.ts

@@ -5,8 +5,20 @@ const envFile = process.env.CI ? '' : '--env-file=.dev.vars'
 const chromeHostResolverRules = process.env.E2E_CHROME_HOST_RESOLVER_RULES
 const appPort = Number(process.env.E2E_APP_PORT ?? 5173)
 const apiPort = Number(process.env.E2E_API_PORT ?? 8222)
+const s3MockPort = Number(process.env.E2E_S3_MOCK_PORT ?? 9191)
+
+const s3MockServer = process.env.E2E_S3_MOCK
+  ? [
+      {
+        command: `node scripts/s3-mock.mjs`,
+        port: s3MockPort,
+        reuseExistingServer: !process.env.CI,
+      },
+    ]
+  : []
 
 const nodeServers = [
+  ...s3MockServer,
   {
     command: `PORT=${apiPort} node ${envFile} node_modules/.bin/tsx server/entry-node.ts`,
     port: apiPort,
@@ -20,6 +32,7 @@ const nodeServers = [
 ]
 
 const cfServers = [
+  ...s3MockServer,
   {
     command: `vite dev --host 127.0.0.1 --port ${appPort} --strictPort`,
     port: appPort,

scripts/run-cloud-e2e.mjs

@@ -5,9 +5,13 @@ import { Resolver } from 'node:dns/promises'
 const args = process.argv.slice(2)
 const runtime = valueAfter('--runtime') ?? process.env.E2E_RUNTIME ?? 'node'
 const project = valueAfter('--project') ?? 'desktop'
+const spec = valueAfter('--spec') ?? 'cloud-store.spec.ts'
+const local = args.includes('--local')
+const withS3Mock = args.includes('--with-s3-mock')
 const cloudflared = process.env.CLOUDFLARED_BIN ?? 'cloudflared'
 const appPort = Number(process.env.E2E_APP_PORT ?? (runtime === 'cf' ? 6174 : 6173))
 const apiPort = Number(process.env.E2E_API_PORT ?? 9222)
+const s3MockPort = Number(process.env.E2E_S3_MOCK_PORT ?? 9191)
 const localBaseUrl = `http://localhost:${appPort}`
 const pidFile = `.cloudflared.${runtime}.pid`
 const tunnelUrlPattern = /https:\/\/[a-zA-Z0-9-]+\.trycloudflare\.com/
@@ -19,36 +23,41 @@ const cloudEnv = {
   VITE_ZPAN_CLOUD_URL: process.env.VITE_ZPAN_CLOUD_URL ?? 'https://zpan-cloud-staging.saltbo.workers.dev',
 }
 
-const tunnel = await startTunnel(localBaseUrl)
-const tunnelHost = new URL(tunnel.url).hostname
-const tunnelIp = await waitForPublicTunnelIp(tunnelHost)
+const tunnel = local ? null : await startTunnel(localBaseUrl)
+const tunnelHost = tunnel ? new URL(tunnel.url).hostname : ''
+const tunnelIp = tunnel ? await waitForPublicTunnelIp(tunnelHost) : ''
+const baseUrl = tunnel?.url ?? localBaseUrl
 const tunnelEnv = {
-  E2E_BASE_URL: tunnel.url,
+  E2E_BASE_URL: baseUrl,
   E2E_LOCAL_BASE_URL: localBaseUrl,
   E2E_APP_PORT: String(appPort),
   E2E_API_PORT: String(apiPort),
-  BETTER_AUTH_URL: tunnel.url,
-  TRUSTED_ORIGINS: `${tunnel.url},${localBaseUrl}`,
-  E2E_CHROME_HOST_RESOLVER_RULES: `MAP ${tunnelHost} ${tunnelIp}`,
+  BETTER_AUTH_URL: baseUrl,
+  TRUSTED_ORIGINS: `${baseUrl},${localBaseUrl}`,
+  ...(tunnel ? { E2E_CHROME_HOST_RESOLVER_RULES: `MAP ${tunnelHost} ${tunnelIp}` } : {}),
 }
 const e2eEnv = {
   ...cloudEnv,
   ...tunnelEnv,
+  ...s3MockEnv(),
   ...runtimeCloudCredentials(runtime),
   ...(runtime === 'cf' ? { E2E_RUNTIME: 'cf' } : {}),
 }
 
 if (runtime === 'cf') {
+  if (local) rmSync('.wrangler/state/v3/d1', { recursive: true, force: true })
   writeDevVars(e2eEnv)
   await run('npx', ['wrangler', 'd1', 'migrations', 'apply', 'DB', '--local'], e2eEnv)
 }
 
 try {
-  await run('npx', ['playwright', 'test', 'cloud-store.spec.ts', `--project=${project}`], e2eEnv)
+  await run('npx', ['playwright', 'test', spec, `--project=${project}`], e2eEnv)
 } finally {
-  try {
-    tunnel.process.kill()
-  } catch {}
+  if (tunnel) {
+    try {
+      tunnel.process.kill()
+    } catch {}
+  }
   if (existsSync(pidFile)) {
     const pid = Number(readFileSync(pidFile, 'utf8'))
     if (Number.isInteger(pid)) {
@@ -77,6 +86,19 @@ function runtimeCloudCredentials(runtime) {
     : {}
 }
 
+function s3MockEnv() {
+  if (!withS3Mock) return {}
+  return {
+    E2E_S3_MOCK: '1',
+    E2E_S3_MOCK_PORT: String(s3MockPort),
+    E2E_STORAGE_ENDPOINT: `http://127.0.0.1:${s3MockPort}`,
+    E2E_STORAGE_BUCKET: 'e2e-test',
+    E2E_STORAGE_REGION: 'auto',
+    E2E_STORAGE_ACCESS_KEY: 'e2e-access-key',
+    E2E_STORAGE_SECRET_KEY: 'e2e-secret-key',
+  }
+}
+
 function startTunnel(target) {
   const child = spawn(cloudflared, ['tunnel', '--url', target, '--no-autoupdate'], {
     stdio: ['ignore', 'pipe', 'pipe'],

scripts/s3-mock.mjs

@@ -0,0 +1,219 @@
+import { createHash, randomUUID } from 'node:crypto'
+import { createServer } from 'node:http'
+
+const port = Number(process.env.E2E_S3_MOCK_PORT ?? 9191)
+const objects = new Map()
+const uploads = new Map()
+
+const server = createServer(async (req, res) => {
+  setCors(res)
+  if (req.method === 'OPTIONS') {
+    res.writeHead(204)
+    res.end()
+    return
+  }
+
+  try {
+    await handleRequest(req, res)
+  } catch (error) {
+    res.writeHead(500, { 'Content-Type': 'text/plain' })
+    res.end(error instanceof Error ? error.message : String(error))
+  }
+})
+
+server.listen(port, '127.0.0.1', () => {
+  console.log(`[s3-mock] listening on http://127.0.0.1:${port}`)
+})
+
+async function handleRequest(req, res) {
+  const url = new URL(req.url ?? '/', `http://${req.headers.host}`)
+  const { bucket, key } = parsePath(url.pathname)
+
+  if (url.pathname === '/health') {
+    res.writeHead(200)
+    res.end('ok')
+    return
+  }
+
+  if (!bucket) {
+    res.writeHead(200)
+    res.end('')
+    return
+  }
+
+  if (req.method === 'POST' && url.searchParams.has('uploads')) {
+    createMultipartUpload(res, bucket, key)
+    return
+  }
+
+  if (req.method === 'PUT' && url.searchParams.has('uploadId') && url.searchParams.has('partNumber')) {
+    await uploadPart(req, res, url)
+    return
+  }
+
+  if (req.method === 'POST' && url.searchParams.has('uploadId')) {
+    completeMultipartUpload(res, url, bucket, key)
+    return
+  }
+
+  if (req.method === 'DELETE' && url.searchParams.has('uploadId')) {
+    uploads.delete(url.searchParams.get('uploadId'))
+    res.writeHead(204)
+    res.end()
+    return
+  }
+
+  const objectKey = storageKey(bucket, key)
+  if (req.method === 'PUT') {
+    const body = await readBody(req)
+    objects.set(objectKey, {
+      body,
+      contentType: req.headers['content-type'] ?? 'application/octet-stream',
+    })
+    res.writeHead(200, { etag: etag(body) })
+    res.end('')
+    return
+  }
+
+  if (req.method === 'HEAD') {
+    const object = objects.get(objectKey)
+    if (!object) {
+      res.writeHead(404)
+      res.end()
+      return
+    }
+    res.writeHead(200, {
+      'Content-Length': object.body.byteLength,
+      'Content-Type': object.contentType,
+      etag: etag(object.body),
+    })
+    res.end()
+    return
+  }
+
+  if (req.method === 'GET') {
+    const object = objects.get(objectKey)
+    if (!object) {
+      res.writeHead(404)
+      res.end('Not found')
+      return
+    }
+    writeObject(res, object, req.headers.range)
+    return
+  }
+
+  if (req.method === 'DELETE') {
+    objects.delete(objectKey)
+    res.writeHead(204)
+    res.end()
+    return
+  }
+
+  res.writeHead(405)
+  res.end('Method not allowed')
+}
+
+function createMultipartUpload(res, bucket, key) {
+  const uploadId = randomUUID()
+  uploads.set(uploadId, { bucket, key, parts: new Map() })
+  res.writeHead(200, { 'Content-Type': 'application/xml' })
+  res.end(`<CreateMultipartUploadResult><UploadId>${uploadId}</UploadId></CreateMultipartUploadResult>`)
+}
+
+async function uploadPart(req, res, url) {
+  const uploadId = url.searchParams.get('uploadId')
+  const upload = uploads.get(uploadId)
+  if (!upload) {
+    res.writeHead(404)
+    res.end('Upload not found')
+    return
+  }
+  const partNumber = Number(url.searchParams.get('partNumber'))
+  const body = await readBody(req)
+  upload.parts.set(partNumber, body)
+  res.writeHead(200, { etag: etag(body) })
+  res.end('')
+}
+
+function completeMultipartUpload(res, url, bucket, key) {
+  const uploadId = url.searchParams.get('uploadId')
+  const upload = uploads.get(uploadId)
+  if (!upload) {
+    res.writeHead(404)
+    res.end('Upload not found')
+    return
+  }
+
+  const parts = [...upload.parts.entries()].sort(([left], [right]) => left - right)
+  const total = parts.reduce((sum, [, part]) => sum + part.byteLength, 0)
+  const body = new Uint8Array(total)
+  let offset = 0
+  for (const [, part] of parts) {
+    body.set(part, offset)
+    offset += part.byteLength
+  }
+
+  objects.set(storageKey(bucket, key), { body, contentType: 'application/octet-stream' })
+  uploads.delete(uploadId)
+  res.writeHead(200, { 'Content-Type': 'application/xml' })
+  res.end('<CompleteMultipartUploadResult />')
+}
+
+function writeObject(res, object, rangeHeader) {
+  if (!rangeHeader) {
+    res.writeHead(200, {
+      'Content-Length': object.body.byteLength,
+      'Content-Type': object.contentType,
+      etag: etag(object.body),
+    })
+    res.end(object.body)
+    return
+  }
+
+  const match = /^bytes=(\d+)-(\d+)?$/.exec(rangeHeader)
+  if (!match) {
+    res.writeHead(416)
+    res.end()
+    return
+  }
+
+  const start = Number(match[1])
+  const end = match[2] ? Number(match[2]) : object.body.byteLength - 1
+  const slice = object.body.slice(start, end + 1)
+  res.writeHead(206, {
+    'Content-Length': slice.byteLength,
+    'Content-Range': `bytes ${start}-${end}/${object.body.byteLength}`,
+    'Content-Type': object.contentType,
+    etag: etag(object.body),
+  })
+  res.end(slice)
+}
+
+function parsePath(pathname) {
+  const parts = pathname.split('/').filter(Boolean).map(decodeURIComponent)
+  return {
+    bucket: parts[0] ?? '',
+    key: parts.slice(1).join('/'),
+  }
+}
+
+function storageKey(bucket, key) {
+  return `${bucket}/${key}`
+}
+
+function setCors(res) {
+  res.setHeader('Access-Control-Allow-Origin', '*')
+  res.setHeader('Access-Control-Allow-Methods', 'GET,HEAD,PUT,POST,DELETE,OPTIONS')
+  res.setHeader('Access-Control-Allow-Headers', '*')
+  res.setHeader('Access-Control-Expose-Headers', 'ETag,Content-Length,Content-Range,Content-Type')
+}
+
+function etag(body) {
+  return `"${createHash('md5').update(body).digest('hex')}"`
+}
+
+async function readBody(req) {
+  const chunks = []
+  for await (const chunk of req) chunks.push(chunk)
+  return new Uint8Array(Buffer.concat(chunks))
+}

server/routes/background-jobs.integration.test.ts

@@ -1,5 +1,6 @@
 import { sql } from 'drizzle-orm'
 import { afterEach, describe, expect, it, vi } from 'vitest'
+import { ARCHIVE_QUEUE_BINDING, type ArchiveJobMessage, runArchiveJobMessage } from '../services/archive-jobs'
 import {
   cancelBackgroundJob,
   createBackgroundJob,
@@ -33,7 +34,7 @@ describe('background jobs API', () => {
     vi.restoreAllMocks()
   })
 
-  it('creates archive jobs through POST and returns the final job state', async () => {
+  it('creates archive jobs through POST and completes them after the response', async () => {
     const { app, db } = await createTestApp()
     const headers = await authedHeaders(app, 'jobs-create@example.com')
     const { orgId } = await getUserOrg(db, 'jobs-create@example.com')
@@ -46,10 +47,25 @@ describe('background jobs API', () => {
 
     const objectStore = new Map<string, Uint8Array>([['route/source.zip', createZip({ 'route.txt': bytes('ok') })]])
     const putKeys: string[] = []
-    vi.spyOn(S3Service.prototype, 'getObjectBytes').mockImplementation(async (_storage, key) => {
+    vi.spyOn(S3Service.prototype, 'headObject').mockImplementation(async (_storage, key) => {
       const bytes = objectStore.get(key)
       if (!bytes) throw new Error(`missing ${key}`)
-      return bytes
+      return { size: bytes.byteLength, contentType: 'application/zip' }
+    })
+    vi.spyOn(S3Service.prototype, 'getObjectBytes').mockImplementation(async (_storage, key, range) => {
+      const bytes = objectStore.get(key)
+      if (!bytes) throw new Error(`missing ${key}`)
+      return range ? sliceRange(bytes, range) : bytes
+    })
+    vi.spyOn(S3Service.prototype, 'getObjectStream').mockImplementation(async (_storage, key) => {
+      const bytes = objectStore.get(key)
+      if (!bytes) throw new Error(`missing ${key}`)
+      return new ReadableStream({
+        start(controller) {
+          controller.enqueue(bytes)
+          controller.close()
+        },
+      })
     })
     vi.spyOn(S3Service.prototype, 'putObject').mockImplementation(async (_storage, key, body) => {
       const bytes = body instanceof Uint8Array ? body : new Uint8Array(await new Response(body).arrayBuffer())
@@ -65,7 +81,14 @@ describe('background jobs API', () => {
     })
 
     expect(res.status).toBe(201)
-    await expect(res.json()).resolves.toMatchObject({
+    const created = (await res.json()) as { id: string }
+    expect(created).toMatchObject({
+      orgId,
+      type: 'archive_extract',
+      status: 'queued',
+    })
+    const completed = await waitForJob(db, orgId, created.id, 'completed')
+    expect(completed).toMatchObject({
       orgId,
       type: 'archive_extract',
       status: 'completed',
@@ -74,6 +97,66 @@ describe('background jobs API', () => {
     expect(putKeys).toHaveLength(1)
   })
 
+  it('dispatches archive jobs to Cloudflare Queue bindings and lets the consumer complete them', async () => {
+    const messages: ArchiveJobMessage[] = []
+    const queue = { send: async (message: ArchiveJobMessage) => messages.push(message) }
+    const { app, db, platform } = await createTestApp({}, { [ARCHIVE_QUEUE_BINDING]: queue })
+    const headers = await authedHeaders(app, 'jobs-queue@example.com')
+    const { orgId } = await getUserOrg(db, 'jobs-queue@example.com')
+    await seedStorage(db)
+    const now = Date.now()
+    await db.run(sql`
+      INSERT INTO matters (id, org_id, alias, name, type, size, dirtype, parent, object, storage_id, status, created_at, updated_at)
+      VALUES ('queue-zip', ${orgId}, 'queue-zip-alias', 'queue.zip', 'application/zip', 200, 0, '', 'queue/source.zip', 'route-storage', 'active', ${now}, ${now})
+    `)
+
+    const objectStore = new Map<string, Uint8Array>([['queue/source.zip', createZip({ 'queue.txt': bytes('ok') })]])
+    vi.spyOn(S3Service.prototype, 'headObject').mockImplementation(async (_storage, key) => {
+      const bytes = objectStore.get(key)
+      if (!bytes) throw new Error(`missing ${key}`)
+      return { size: bytes.byteLength, contentType: 'application/zip' }
+    })
+    vi.spyOn(S3Service.prototype, 'getObjectBytes').mockImplementation(async (_storage, key, range) => {
+      const bytes = objectStore.get(key)
+      if (!bytes) throw new Error(`missing ${key}`)
+      return range ? sliceRange(bytes, range) : bytes
+    })
+    vi.spyOn(S3Service.prototype, 'getObjectStream').mockImplementation(async (_storage, key) => {
+      const bytes = objectStore.get(key)
+      if (!bytes) throw new Error(`missing ${key}`)
+      return new ReadableStream({
+        start(controller) {
+          controller.enqueue(bytes)
+          controller.close()
+        },
+      })
+    })
+    vi.spyOn(S3Service.prototype, 'putObject').mockImplementation(async (_storage, key, body) => {
+      const bytes = body instanceof Uint8Array ? body : new Uint8Array(await new Response(body).arrayBuffer())
+      objectStore.set(key, bytes)
+      return bytes.byteLength
+    })
+
+    const res = await app.request('/api/background-jobs', {
+      method: 'POST',
+      headers: { ...headers, 'Content-Type': 'application/json' },
+      body: JSON.stringify({ type: 'archive_extract', matterId: 'queue-zip' }),
+    })
+
+    expect(res.status).toBe(201)
+    const created = (await res.json()) as { id: string; status: string }
+    expect(created.status).toBe('queued')
+    expect(messages).toHaveLength(1)
+    await expect(getBackgroundJob(db, orgId, created.id)).resolves.toMatchObject({ status: 'queued' })
+
+    await runArchiveJobMessage(platform, messages[0])
+
+    await expect(getBackgroundJob(db, orgId, created.id)).resolves.toMatchObject({
+      status: 'completed',
+      progress: { outputBytes: 2, fileCount: 1 },
+    })
+  })
+
   it('returns a failed archive job for a missing explicit target folder', async () => {
     const { app, db } = await createTestApp()
     const headers = await authedHeaders(app, 'jobs-missing-target@example.com')
@@ -92,7 +175,15 @@ describe('background jobs API', () => {
     })
 
     expect(res.status).toBe(201)
-    await expect(res.json()).resolves.toMatchObject({
+    const created = (await res.json()) as { id: string }
+    expect(created).toMatchObject({
+      orgId,
+      type: 'archive_compress',
+      status: 'queued',
+      errorMessage: null,
+    })
+    const failed = await waitForJob(db, orgId, created.id, 'failed')
+    expect(failed).toMatchObject({
       orgId,
       type: 'archive_compress',
       status: 'failed',
@@ -122,7 +213,15 @@ describe('background jobs API', () => {
     })
 
     expect(res.status).toBe(201)
-    await expect(res.json()).resolves.toMatchObject({
+    const created = (await res.json()) as { id: string }
+    expect(created).toMatchObject({
+      orgId,
+      type: 'archive_extract',
+      status: 'queued',
+      errorMessage: null,
+    })
+    const failed = await waitForJob(db, orgId, created.id, 'failed')
+    expect(failed).toMatchObject({
       orgId,
       type: 'archive_extract',
       status: 'failed',
@@ -235,6 +334,20 @@ async function seedStorage(db: TestDb): Promise<void> {
   `)
 }
 
+async function waitForJob(
+  db: TestDb,
+  orgId: string,
+  jobId: string,
+  status: 'completed' | 'failed',
+): Promise<Awaited<ReturnType<typeof getBackgroundJob>>> {
+  for (let i = 0; i < 20; i++) {
+    const job = await getBackgroundJob(db, orgId, jobId)
+    if (job.status === status) return job
+    await new Promise((resolve) => setTimeout(resolve, 10))
+  }
+  throw new Error(`Job ${jobId} did not reach ${status}`)
+}
+
 function createZip(files: Record<string, Uint8Array>): Uint8Array {
   const encoder = new TextEncoder()
   const localParts: Uint8Array[] = []
@@ -282,6 +395,12 @@ function bytes(value: string): Uint8Array {
   return new TextEncoder().encode(value)
 }
 
+function sliceRange(bytes: Uint8Array, range: string): Uint8Array {
+  const match = /^bytes=(\d+)-(\d+)$/.exec(range)
+  if (!match) throw new Error(`Unsupported range: ${range}`)
+  return bytes.slice(Number(match[1]), Number(match[2]) + 1)
+}
+
 function concat(parts: Uint8Array[]): Uint8Array {
   const out = new Uint8Array(parts.reduce((sum, part) => sum + part.length, 0))
   let offset = 0

server/routes/background-jobs.ts

@@ -5,7 +5,8 @@ import { Hono } from 'hono'
 import { createBackgroundJobRequestSchema, listBackgroundJobsQuerySchema } from '../../shared/schemas'
 import { requireAuth } from '../middleware/auth'
 import type { Env } from '../middleware/platform'
-import { createArchiveJob } from '../services/archive-processing'
+import { dispatchArchiveJob } from '../services/archive-jobs'
+import { enqueueArchiveJob } from '../services/archive-processing'
 import {
   BackgroundJobError,
   cancelBackgroundJob,
@@ -32,11 +33,15 @@ const backgroundJobs = new Hono<Env>()
         const orgId = requireOrg(c)
         const userId = c.get('userId')
         if (!userId) throw new BackgroundJobError('not_found')
-        return createArchiveJob(c.get('platform').db, {
+        const db = c.get('platform').db
+        const request = c.req.valid('json')
+        const job = await enqueueArchiveJob(db, {
           orgId,
           userId,
-          request: c.req.valid('json'),
+          request,
         })
+        await dispatchArchiveJob(c.get('platform'), { orgId, userId, request, jobId: job.id })
+        return job
       },
       201,
     ),
@@ -58,7 +63,18 @@ const backgroundJobs = new Hono<Env>()
       c,
       async () => {
         const orgId = requireOrg(c)
-        return retryBackgroundJob(c.get('platform').db, orgId, c.req.param('id'))
+        const db = c.get('platform').db
+        const job = await retryBackgroundJob(db, orgId, c.req.param('id'))
+        const request = createBackgroundJobRequestSchema.safeParse(job.metadata)
+        if (request.success) {
+          await dispatchArchiveJob(c.get('platform'), {
+            orgId,
+            userId: job.userId,
+            request: request.data,
+            jobId: job.id,
+          })
+        }
+        return job
       },
       201,
     ),
@@ -66,7 +82,7 @@ const backgroundJobs = new Hono<Env>()
 
 export default backgroundJobs
 
-function requireOrg(c: Context<Env>): string {
+function requireOrg(c: { get(key: 'orgId'): string | null }): string {
   const orgId = c.get('orgId')
   if (!orgId) throw new BackgroundJobError('not_found')
   return orgId

server/services/archive-jobs.ts

@@ -0,0 +1,67 @@
+import type { CreateBackgroundJobRequest } from '@shared/schemas'
+import type { Platform } from '../platform/interface'
+import { processArchiveJob } from './archive-processing'
+
+export const ARCHIVE_QUEUE_BINDING = 'ARCHIVE_QUEUE'
+
+export interface ArchiveJobMessage {
+  jobId: string
+  orgId: string
+  userId: string
+  request: CreateBackgroundJobRequest
+}
+
+interface QueueProducer {
+  send(message: ArchiveJobMessage): Promise<void>
+}
+
+class LocalArchiveQueue {
+  private readonly pending: Array<{ platform: Platform; message: ArchiveJobMessage }> = []
+  private running = false
+
+  push(platform: Platform, message: ArchiveJobMessage): void {
+    this.pending.push({ platform, message })
+    if (!this.running) setTimeout(() => void this.drain(), 0)
+  }
+
+  private async drain(): Promise<void> {
+    if (this.running) return
+    this.running = true
+
+    try {
+      for (;;) {
+        const next = this.pending.shift()
+        if (!next) return
+        try {
+          await runArchiveJobMessage(next.platform, next.message)
+        } catch (error) {
+          console.error('[archive-jobs] local worker failed:', error)
+        }
+      }
+    } finally {
+      this.running = false
+      if (this.pending.length > 0) setTimeout(() => void this.drain(), 0)
+    }
+  }
+}
+
+const localArchiveQueue = new LocalArchiveQueue()
+
+export async function dispatchArchiveJob(platform: Platform, message: ArchiveJobMessage): Promise<void> {
+  const queue = platform.getBinding<QueueProducer>(ARCHIVE_QUEUE_BINDING)
+  if (queue) {
+    await queue.send(message)
+    return
+  }
+
+  localArchiveQueue.push(platform, message)
+}
+
+export async function runArchiveJobMessage(platform: Platform, message: ArchiveJobMessage): Promise<void> {
+  await processArchiveJob(platform.db, {
+    orgId: message.orgId,
+    userId: message.userId,
+    request: message.request,
+    jobId: message.jobId,
+  })
+}

server/services/archive-processing.test.ts

@@ -3,8 +3,8 @@ import { describe, expect, it } from 'vitest'
 import { createTestApp } from '../test/setup.js'
 import { createArchiveJob } from './archive-processing'
 import type { S3Service } from './s3'
-import { collectCompressionPlan } from './zip-compress'
-import { validateAndExtractZip } from './zip-extract'
+import { collectCompressionPlan, createZipArchiveStream, ZIP_COMPRESS_LIMITS } from './zip-compress'
+import { validateAndExtractZip, ZIP_EXTRACT_LIMITS } from './zip-extract'
 
 type TestDb = Awaited<ReturnType<typeof createTestApp>>['db']
 
@@ -16,12 +16,29 @@ class MemoryS3 {
   objects = new Map<string, Uint8Array>()
   putKeys: string[] = []
 
-  async getObjectBytes(_storage: unknown, key: string): Promise<Uint8Array> {
+  async getObjectBytes(_storage: unknown, key: string, range?: string): Promise<Uint8Array> {
     const bytes = this.objects.get(key)
     if (!bytes) throw new Error(`Object not found: ${key}`)
+    if (range) return sliceRange(bytes, range)
     return bytes
   }
 
+  async headObject(_storage: unknown, key: string): Promise<{ size: number; contentType: string }> {
+    const bytes = this.objects.get(key)
+    if (!bytes) throw new Error(`Object not found: ${key}`)
+    return { size: bytes.byteLength, contentType: 'application/octet-stream' }
+  }
+
+  async getObjectStream(_storage: unknown, key: string): Promise<ReadableStream<Uint8Array>> {
+    const bytes = await this.getObjectBytes(_storage, key)
+    return new ReadableStream({
+      start(controller) {
+        controller.enqueue(bytes)
+        controller.close()
+      },
+    })
+  }
+
   async putObject(_storage: unknown, key: string, body: Uint8Array | ReadableStream): Promise<number> {
     const bytes = body instanceof Uint8Array ? body : new Uint8Array(await new Response(body).arrayBuffer())
     this.objects.set(key, bytes)
@@ -58,6 +75,30 @@ class FailAfterPutS3 extends MemoryS3 {
   }
 }
 
+class GeneratedObjectS3 extends MemoryS3 {
+  putSizes = new Map<string, number>()
+
+  constructor(
+    private readonly generatedKey: string,
+    private readonly generatedSize: number,
+  ) {
+    super()
+  }
+
+  override async getObjectStream(_storage: unknown, key: string): Promise<ReadableStream<Uint8Array>> {
+    if (key === this.generatedKey) return generatedBytes(this.generatedSize)
+    return super.getObjectStream(_storage, key)
+  }
+
+  override async putObject(_storage: unknown, key: string, body: Uint8Array | ReadableStream): Promise<number> {
+    const size = body instanceof Uint8Array ? body.byteLength : await drainStream(body)
+    this.objects.set(key, new Uint8Array())
+    this.putKeys.push(key)
+    this.putSizes.set(key, size)
+    return size
+  }
+}
+
 describe('archive processing', () => {
   it('extracts a small ZIP into folder and file matters and writes objects', async () => {
     const { db } = await createTestApp()
@@ -90,6 +131,30 @@ describe('archive processing', () => {
     expect(s3.putKeys).toHaveLength(1)
   })
 
+  it('prevalidates then streams extraction for a 128 MiB ZIP entry', async () => {
+    const { db } = await createTestApp()
+    await seedStorage(db)
+    const size = 128 * 1024 * 1024
+    const archive = await streamToBytes(
+      createZipArchiveStream([{ archivePath: 'large.bin', openStream: async () => generatedBytes(size) }]),
+    )
+    await seedMatter(db, { id: 'large-zip', name: 'large.zip', object: 'source/large.zip', size: archive.byteLength })
+
+    const s3 = new GeneratedObjectS3('unused', 0)
+    s3.objects.set('source/large.zip', archive)
+    const job = await createArchiveJob(db, {
+      orgId: ORG_ID,
+      userId: USER_ID,
+      request: { type: 'archive_extract', matterId: 'large-zip' },
+      s3: s3 as unknown as S3Service,
+    })
+
+    expect(job).toMatchObject({ status: 'completed', type: 'archive_extract' })
+    expect(job.progress).toMatchObject({ outputBytes: size, fileCount: 1 })
+    expect(s3.putKeys).toHaveLength(1)
+    expect(s3.putSizes.get(s3.putKeys[0])).toBe(size)
+  }, 60_000)
+
   it('compresses selected matters into a ZIP matter and object', async () => {
     const { db } = await createTestApp()
     await seedStorage(db)
@@ -116,6 +181,26 @@ describe('archive processing', () => {
     expect(s3.objects.get(zipMatter[0].object)?.length).toBe(zipMatter[0].size)
   })
 
+  it('streams compression for a 128 MiB source without buffering the source object', async () => {
+    const { db } = await createTestApp()
+    await seedStorage(db)
+    const size = 128 * 1024 * 1024
+    await seedMatter(db, { id: 'large-file', name: 'large.bin', object: 'objects/large.bin', size })
+
+    const s3 = new GeneratedObjectS3('objects/large.bin', size)
+    const job = await createArchiveJob(db, {
+      orgId: ORG_ID,
+      userId: USER_ID,
+      request: { type: 'archive_compress', matterIds: ['large-file'] },
+      s3: s3 as unknown as S3Service,
+    })
+
+    expect(job).toMatchObject({ status: 'completed', type: 'archive_compress' })
+    expect(job.progress).toMatchObject({ inputBytes: size, processedBytes: size, fileCount: 1 })
+    expect(s3.putKeys).toHaveLength(1)
+    expect(s3.putSizes.get(s3.putKeys[0]) ?? 0).toBeGreaterThan(0)
+  }, 60_000)
+
   it('compresses an empty selected folder as a ZIP directory entry', async () => {
     const { db } = await createTestApp()
     await seedStorage(db)
@@ -194,7 +279,10 @@ describe('archive processing', () => {
     const s3 = new MemoryS3()
     s3.objects.set(
       'source/large.zip',
-      createZip({ 'large.bin': bytes('x') }, { declaredSizes: { 'large.bin': 25 * 1024 * 1024 + 1 } }),
+      createZip(
+        { 'large.bin': bytes('x') },
+        { declaredSizes: { 'large.bin': ZIP_EXTRACT_LIMITS.singleFileBytes + 1 } },
+      ),
     )
 
     const job = await createArchiveJob(db, {
@@ -206,7 +294,7 @@ describe('archive processing', () => {
 
     expect(job).toMatchObject({
       status: 'failed',
-      errorMessage: 'ZIP entry exceeds 26214400 bytes',
+      errorMessage: `ZIP entry exceeds ${ZIP_EXTRACT_LIMITS.singleFileBytes} bytes`,
     })
     expect(s3.putKeys).toHaveLength(0)
   })
@@ -232,11 +320,11 @@ describe('archive processing', () => {
 
     expect(job.status).toBe('failed')
     expect(job.errorMessage).toBe('Quota exceeded for extracted ZIP contents')
-    expect(s3.putKeys).toHaveLength(0)
+    expect(s3.objects.size).toBe(1)
     await expect(activeMatterCount(db)).resolves.toBe(1)
   })
 
-  it('fails compression quota checks before writing output', async () => {
+  it('fails compression quota checks and removes streamed output', async () => {
     const { db } = await createTestApp()
     await seedStorage(db)
     await db.run(sql`
@@ -257,7 +345,7 @@ describe('archive processing', () => {
 
     expect(job.status).toBe('failed')
     expect(job.errorMessage).toBe('Quota exceeded for generated ZIP archive')
-    expect(s3.putKeys).toHaveLength(0)
+    expect(s3.objects.size).toBe(1)
     await expect(activeMatterCount(db)).resolves.toBe(1)
   })
 
@@ -437,7 +525,7 @@ describe('archive processing', () => {
       id: 'large-file',
       name: 'large.bin',
       object: 'objects/large.bin',
-      size: 25 * 1024 * 1024 + 1,
+      size: ZIP_COMPRESS_LIMITS.singleFileBytes + 1,
     })
     await seedMatter(db, {
       id: 'deep-file',
@@ -455,7 +543,7 @@ describe('archive processing', () => {
       'Only active matters can be archived',
     )
     await expect(collectCompressionPlan(db, ORG_ID, ['large-file'])).rejects.toThrow(
-      'Compression source file exceeds 26214400 bytes',
+      `Compression source file exceeds ${ZIP_COMPRESS_LIMITS.singleFileBytes} bytes`,
     )
     await expect(collectCompressionPlan(db, ORG_ID, ['deep-file'])).rejects.toThrow(
       'Compression directory depth exceeds 10',
@@ -482,13 +570,15 @@ describe('archive processing', () => {
     const { db } = await createTestApp()
     await seedStorage(db)
     const ids: string[] = []
-    for (let index = 0; index < 201; index += 1) {
+    for (let index = 0; index < ZIP_COMPRESS_LIMITS.fileCount + 1; index += 1) {
       const id = `many-${index}`
       ids.push(id)
       await seedMatter(db, { id, name: `${id}.txt`, object: `objects/${id}.txt`, size: 1 })
     }
 
-    await expect(collectCompressionPlan(db, ORG_ID, ids)).rejects.toThrow('Compression file count exceeds 200')
+    await expect(collectCompressionPlan(db, ORG_ID, ids)).rejects.toThrow(
+      `Compression file count exceeds ${ZIP_COMPRESS_LIMITS.fileCount}`,
+    )
   })
 
   it('rejects unsafe and unsupported ZIP entries during validation', () => {
@@ -517,16 +607,20 @@ describe('archive processing', () => {
   })
 
   it('enforces ZIP validation count and total output limits from metadata', () => {
-    const manyEntries = Object.fromEntries(Array.from({ length: 201 }, (_, index) => [`file-${index}.txt`, bytes('x')]))
-    expect(() => validateAndExtractZip(createZip(manyEntries))).toThrow('ZIP file count exceeds 200')
+    const manyEntries = Object.fromEntries(
+      Array.from({ length: ZIP_EXTRACT_LIMITS.fileCount + 1 }, (_, index) => [`file-${index}.txt`, bytes('x')]),
+    )
+    expect(() => validateAndExtractZip(createZip(manyEntries))).toThrow(
+      `ZIP file count exceeds ${ZIP_EXTRACT_LIMITS.fileCount}`,
+    )
     const totalLimitEntries = Object.fromEntries(
       Array.from({ length: 5 }, (_, index) => [`total-${index}`, bytes('x')]),
     )
     const totalLimitSizes = Object.fromEntries(
-      Array.from({ length: 5 }, (_, index) => [`total-${index}`, 21 * 1024 * 1024]),
+      Array.from({ length: 5 }, (_, index) => [`total-${index}`, 256 * 1024 * 1024]),
     )
     expect(() => validateAndExtractZip(createZip(totalLimitEntries, { declaredSizes: totalLimitSizes }))).toThrow(
-      'ZIP extraction output exceeds 104857600 bytes',
+      `ZIP extraction output exceeds ${ZIP_EXTRACT_LIMITS.totalOutputBytes} bytes`,
     )
   })
 })
@@ -626,6 +720,42 @@ function bytes(value: string): Uint8Array {
   return new TextEncoder().encode(value)
 }
 
+function generatedBytes(size: number): ReadableStream<Uint8Array> {
+  const chunk = new Uint8Array(1024 * 1024)
+  let remaining = size
+  return new ReadableStream({
+    pull(controller) {
+      if (remaining <= 0) {
+        controller.close()
+        return
+      }
+      const length = Math.min(chunk.byteLength, remaining)
+      controller.enqueue(length === chunk.byteLength ? chunk : chunk.slice(0, length))
+      remaining -= length
+    },
+  })
+}
+
+async function drainStream(stream: ReadableStream): Promise<number> {
+  const reader = stream.getReader()
+  let size = 0
+  for (;;) {
+    const { done, value } = await reader.read()
+    if (done) return size
+    size += value instanceof Uint8Array ? value.byteLength : 0
+  }
+}
+
+async function streamToBytes(stream: ReadableStream<Uint8Array>): Promise<Uint8Array> {
+  return new Uint8Array(await new Response(stream).arrayBuffer())
+}
+
+function sliceRange(bytes: Uint8Array, range: string): Uint8Array {
+  const match = /^bytes=(\d+)-(\d+)$/.exec(range)
+  if (!match) throw new Error(`Unsupported range: ${range}`)
+  return bytes.slice(Number(match[1]), Number(match[2]) + 1)
+}
+
 function concat(parts: Uint8Array[]): Uint8Array {
   const out = new Uint8Array(parts.reduce((sum, part) => sum + part.length, 0))
   let offset = 0

server/services/archive-processing.ts

@@ -7,11 +7,12 @@ import { matters } from '../db/schema'
 import type { Database } from '../platform/interface'
 import { createBackgroundJob, updateBackgroundJob } from './background-jobs'
 import { createMatter, decrementUsage, getMatter, incrementUsageIfAllowed, purgeMatters } from './matter'
+import { createNotification } from './notification'
 import { buildObjectKey } from './path-template'
 import { S3Service } from './s3'
 import { getStorage, selectStorage } from './storage'
-import { collectCompressionPlan, createZipArchive } from './zip-compress'
-import { validateAndExtractZip } from './zip-extract'
+import { collectCompressionPlan, createZipArchiveStream } from './zip-compress'
+import { streamValidatedZip, validateZipDirectory } from './zip-extract'
 
 export interface CreateArchiveJobInput {
   orgId: string
@@ -24,8 +25,13 @@ const ZIP_MIME = 'application/zip'
 const DEFAULT_FILE_MIME = 'application/octet-stream'
 
 export async function createArchiveJob(db: Database, input: CreateArchiveJobInput): Promise<BackgroundJob> {
+  const job = await enqueueArchiveJob(db, input)
+  return processArchiveJob(db, { ...input, jobId: job.id })
+}
+
+export async function enqueueArchiveJob(db: Database, input: CreateArchiveJobInput): Promise<BackgroundJob> {
   const targetFolder = input.request.targetFolder ?? null
-  const job = await createBackgroundJob(db, {
+  return createBackgroundJob(db, {
     orgId: input.orgId,
     userId: input.userId,
     type: input.request.type,
@@ -33,21 +39,30 @@ export async function createArchiveJob(db: Database, input: CreateArchiveJobInpu
     metadata: input.request,
     cancelable: false,
   })
+}
 
+export async function processArchiveJob(
+  db: Database,
+  input: CreateArchiveJobInput & { jobId: string },
+): Promise<BackgroundJob> {
   const s3 = input.s3 ?? new S3Service()
   try {
-    await updateBackgroundJob(db, input.orgId, job.id, { status: 'running', startedAt: new Date() })
-    if (input.request.type === 'archive_compress') {
-      return await runCompressionJob(db, s3, job.id, input.orgId, input.userId, input.request)
-    }
-    return await runExtractionJob(db, s3, job.id, input.orgId, input.userId, input.request)
+    await updateBackgroundJob(db, input.orgId, input.jobId, { status: 'running', startedAt: new Date() })
+    const finished =
+      input.request.type === 'archive_compress'
+        ? await runCompressionJob(db, s3, input.jobId, input.orgId, input.userId, input.request)
+        : await runExtractionJob(db, s3, input.jobId, input.orgId, input.userId, input.request)
+    await notifyArchiveJobFinished(db, finished)
+    return finished
   } catch (error) {
-    return updateBackgroundJob(db, input.orgId, job.id, {
+    const failed = await updateBackgroundJob(db, input.orgId, input.jobId, {
       status: 'failed',
       errorMessage: (error as Error).message,
       retryable: false,
       cancelable: false,
     })
+    await notifyArchiveJobFinished(db, failed)
+    return failed
   }
 }
 
@@ -68,28 +83,30 @@ async function runCompressionJob(
     progress: { inputBytes: plan.inputBytes, fileCount: plan.files.length },
   })
 
-  const objects = []
+  const sources = []
   for (const file of plan.files) {
     const storage = await requireStorage(db, file.matter.storageId)
-    objects.push({ archivePath: file.archivePath, bytes: await s3.getObjectBytes(storage, file.matter.object) })
+    sources.push({
+      archivePath: file.archivePath,
+      openStream: () => s3.getObjectStream(storage, file.matter.object),
+    })
   }
 
-  const zipBytes = createZipArchive(objects, plan.directories)
   const targetStorage = (await selectStorage(db, 'private')) as unknown as S3StorageType
-  const allowed = await incrementUsageIfAllowed(db, orgId, targetStorage.id, zipBytes.length)
-  if (!allowed) throw new Error('Quota exceeded for generated ZIP archive')
-
   const key = buildObjectKey({ uid: userId, orgId, rawExt: '.zip' })
   let objectWritten = false
+  let outputBytes = 0
   try {
-    await s3.putObject(targetStorage, key, zipBytes, ZIP_MIME)
+    outputBytes = await s3.putObject(targetStorage, key, createZipArchiveStream(sources, plan.directories), ZIP_MIME)
     objectWritten = true
+    const allowed = await incrementUsageIfAllowed(db, orgId, targetStorage.id, outputBytes)
+    if (!allowed) throw new Error('Quota exceeded for generated ZIP archive')
     const matter = await createMatter(db, {
       orgId,
       userId,
       name: plan.outputName,
       type: ZIP_MIME,
-      size: zipBytes.length,
+      size: outputBytes,
       dirtype: DirType.FILE,
       parent: plan.targetFolder,
       object: key,
@@ -102,16 +119,16 @@ async function runCompressionJob(
       status: 'completed',
       progress: {
         inputBytes: plan.inputBytes,
-        outputBytes: zipBytes.length,
+        outputBytes,
         processedBytes: plan.inputBytes,
         fileCount: plan.files.length,
         currentFilename: null,
       },
-      resultMetadata: { matterId: matter.id, outputName: matter.name, outputBytes: zipBytes.length },
+      resultMetadata: { matterId: matter.id, outputName: matter.name, outputBytes },
       cancelable: false,
     })
   } catch (error) {
-    await decrementUsage(db, orgId, new Map([[targetStorage.id, zipBytes.length]]), zipBytes.length)
+    if (outputBytes > 0) await decrementUsage(db, orgId, new Map([[targetStorage.id, outputBytes]]), outputBytes)
     if (objectWritten) await s3.deleteObject(targetStorage, key)
     throw error
   }
@@ -133,51 +150,37 @@ async function runExtractionJob(
 
   if (request.targetFolder !== undefined) await requireTargetFolder(db, orgId, request.targetFolder)
   const sourceStorage = await requireStorage(db, zipMatter.storageId)
-  const zipBytes = await s3.getObjectBytes(sourceStorage, zipMatter.object)
-  const archive = validateAndExtractZip(zipBytes)
+  const sourceHead = await s3.headObject(sourceStorage, zipMatter.object)
+  const plan = await validateZipDirectory(sourceHead.size, (start, end) =>
+    s3.getObjectBytes(sourceStorage, zipMatter.object, `bytes=${start}-${end}`),
+  )
   const targetFolder = request.targetFolder ?? zipMatter.parent
   const targetStorage = (await selectStorage(db, 'private')) as unknown as S3StorageType
-  const allowed = await incrementUsageIfAllowed(db, orgId, targetStorage.id, archive.totalBytes)
-  if (!allowed) throw new Error('Quota exceeded for extracted ZIP contents')
-
   const writtenKeys: string[] = []
   const createdMatterIds: string[] = []
+  const folderParents = new Map<string, string>()
+  let outputBytes = 0
   try {
-    const folderParents = new Map<string, string>()
-    for (const folderPath of archive.folders) {
-      const parts = folderPath.split('/')
-      const parentPath = parts.slice(0, -1).join('/')
-      const parent = parentPath ? folderParents.get(parentPath) : targetFolder
-      if (parent === undefined) throw new Error(`Missing parent folder for ${folderPath}`)
-      const folder = await createMatter(db, {
-        orgId,
-        userId,
-        name: parts[parts.length - 1],
-        type: 'folder',
-        size: 0,
-        dirtype: DirType.USER_FOLDER,
-        parent,
-        object: '',
-        storageId: targetStorage.id,
-        status: 'active',
-        onConflict: 'rename',
-      })
-      createdMatterIds.push(folder.id)
-      folderParents.set(folderPath, buildMatterPath(folder.parent, folder.name))
+    for (const folderPath of plan.folders) {
+      await ensureExtractedFolder(folderPath)
     }
+    const allowed = await incrementUsageIfAllowed(db, orgId, targetStorage.id, plan.totalBytes)
+    if (!allowed) throw new Error('Quota exceeded for extracted ZIP contents')
+    outputBytes = plan.totalBytes
 
-    for (const file of archive.files) {
-      const parent = file.parentPath ? folderParents.get(file.parentPath) : targetFolder
-      if (parent === undefined) throw new Error(`Missing parent folder for ${file.path}`)
+    const zipStream = await s3.getObjectStream(sourceStorage, zipMatter.object)
+    const archive = await streamValidatedZip(zipStream, async (file) => {
+      const parent = file.parentPath ? await ensureExtractedFolder(file.parentPath) : targetFolder
       const key = buildObjectKey({ uid: userId, orgId, rawExt: extension(file.name) })
-      await s3.putObject(targetStorage, key, file.bytes, DEFAULT_FILE_MIME)
+      const size = await s3.putObject(targetStorage, key, file.stream, DEFAULT_FILE_MIME)
+      await file.size
       writtenKeys.push(key)
       const matter = await createMatter(db, {
         orgId,
         userId,
         name: file.name,
         type: DEFAULT_FILE_MIME,
-        size: file.size,
+        size,
         dirtype: DirType.FILE,
         parent,
         object: key,
@@ -186,15 +189,15 @@ async function runExtractionJob(
         onConflict: 'rename',
       })
       createdMatterIds.push(matter.id)
-    }
+    })
 
     return updateBackgroundJob(db, orgId, jobId, {
       status: 'completed',
       progress: {
-        inputBytes: zipMatter.size ?? zipBytes.length,
+        inputBytes: sourceHead.size,
         outputBytes: archive.totalBytes,
-        processedBytes: zipMatter.size ?? zipBytes.length,
-        fileCount: archive.files.length,
+        processedBytes: sourceHead.size,
+        fileCount: plan.fileCount,
         currentFilename: null,
       },
       resultMetadata: { matterIds: createdMatterIds, outputBytes: archive.totalBytes },
@@ -202,10 +205,36 @@ async function runExtractionJob(
     })
   } catch (error) {
     await purgeMatters(db, orgId, createdMatterIds)
-    await decrementUsage(db, orgId, new Map([[targetStorage.id, archive.totalBytes]]), archive.totalBytes)
+    if (outputBytes > 0) await decrementUsage(db, orgId, new Map([[targetStorage.id, outputBytes]]), outputBytes)
     await s3.deleteObjects(targetStorage, writtenKeys)
     throw error
   }
+
+  async function ensureExtractedFolder(folderPath: string): Promise<string> {
+    const existing = folderParents.get(folderPath)
+    if (existing) return existing
+
+    const parts = folderPath.split('/')
+    const parentPath = parts.slice(0, -1).join('/')
+    const parent = parentPath ? await ensureExtractedFolder(parentPath) : targetFolder
+    const folder = await createMatter(db, {
+      orgId,
+      userId,
+      name: parts[parts.length - 1],
+      type: 'folder',
+      size: 0,
+      dirtype: DirType.USER_FOLDER,
+      parent,
+      object: '',
+      storageId: targetStorage.id,
+      status: 'active',
+      onConflict: 'rename',
+    })
+    createdMatterIds.push(folder.id)
+    const matterPath = buildMatterPath(folder.parent, folder.name)
+    folderParents.set(folderPath, matterPath)
+    return matterPath
+  }
 }
 
 async function requireStorage(db: Database, storageId: string): Promise<S3StorageType> {
@@ -240,3 +269,19 @@ function extension(name: string): string {
   const dot = name.lastIndexOf('.')
   return dot >= 0 ? name.slice(dot) : ''
 }
+
+async function notifyArchiveJobFinished(db: Database, job: BackgroundJob): Promise<void> {
+  const completed = job.status === 'completed'
+  const action = job.type === 'archive_extract' ? 'extraction' : 'compression'
+  await createNotification(db, {
+    userId: job.userId,
+    type: completed ? 'archive_job_completed' : 'archive_job_failed',
+    title: completed ? `File ${action} completed` : `File ${action} failed`,
+    body: completed
+      ? `Background task ${job.id} is complete.`
+      : (job.errorMessage ?? `Background task ${job.id} failed.`),
+    refType: 'background_job',
+    refId: job.id,
+    metadata: JSON.stringify({ jobId: job.id, jobType: job.type, status: job.status }),
+  })
+}

server/services/s3.test.ts

@@ -1,4 +1,4 @@
-import { describe, expect, it, vi } from 'vitest'
+import { beforeEach, describe, expect, it, vi } from 'vitest'
 import type { Storage } from '../../shared/types'
 import { S3Service } from './s3.js'
 
@@ -116,6 +116,12 @@ describe('S3Service', () => {
   const service = new S3Service()
   const storage = makeStorage()
 
+  beforeEach(() => {
+    mockSend.mockReset()
+    vi.clearAllMocks()
+    vi.unstubAllGlobals()
+  })
+
   describe('createClient', () => {
     it('creates a client with correct config', () => {
       const client = service.createClient(storage)
@@ -231,104 +237,37 @@ describe('S3Service', () => {
   })
 
   describe('getObjectBytes', () => {
-    it('returns bytes from Uint8Array bodies', async () => {
+    it('returns bytes from fetched object bodies', async () => {
       const bytes = new Uint8Array([1, 2, 3])
-      mockSend.mockResolvedValueOnce({ Body: bytes })
+      vi.stubGlobal('fetch', vi.fn().mockResolvedValueOnce(new Response(bytes)))
 
       await expect(service.getObjectBytes(storage, 'test.bin')).resolves.toEqual(bytes)
+      expect(fetch).toHaveBeenCalledWith('https://signed-url.example.com', undefined)
     })
 
-    it('returns bytes from ReadableStream bodies', async () => {
+    it('sends Range when reading partial object bytes', async () => {
       const bytes = new Uint8Array([4, 5, 6])
-      const stream = new ReadableStream({
-        start(controller) {
-          controller.enqueue(bytes)
-          controller.close()
-        },
+      vi.stubGlobal('fetch', vi.fn().mockResolvedValueOnce(new Response(bytes, { status: 206 })))
+
+      await expect(service.getObjectBytes(storage, 'test.bin', 'bytes=0-2')).resolves.toEqual(bytes)
+      expect(fetch).toHaveBeenCalledWith('https://signed-url.example.com', {
+        headers: { Range: 'bytes=0-2' },
       })
-      mockSend.mockResolvedValueOnce({ Body: stream })
-
-      await expect(service.getObjectBytes(storage, 'test.bin')).resolves.toEqual(bytes)
     })
 
-    it('returns bytes from transformToByteArray bodies', async () => {
-      const bytes = new Uint8Array([1, 2, 3])
-      mockSend.mockResolvedValueOnce({
-        Body: { transformToByteArray: async () => bytes },
-      })
+    it('rejects failed object reads', async () => {
+      vi.stubGlobal('fetch', vi.fn().mockResolvedValueOnce(new Response(null, { status: 404 })))
 
-      await expect(service.getObjectBytes(storage, 'test.bin')).resolves.toEqual(bytes)
-      expect(mockSend).toHaveBeenCalledWith(
-        expect.objectContaining({ input: { Bucket: 'my-bucket', Key: 'test.bin' } }),
-      )
-    })
-
-    it('returns bytes from arrayBuffer bodies', async () => {
-      const bytes = new Uint8Array([7, 8, 9])
-      mockSend.mockResolvedValueOnce({
-        Body: { arrayBuffer: async () => bytes.buffer },
-      })
-
-      await expect(service.getObjectBytes(storage, 'test.bin')).resolves.toEqual(bytes)
-    })
-
-    it('rejects empty object bodies', async () => {
-      mockSend.mockResolvedValueOnce({})
-
-      await expect(service.getObjectBytes(storage, 'missing.bin')).rejects.toThrow('Empty body from object')
-    })
-
-    it('rejects unsupported object bodies', async () => {
-      mockSend.mockResolvedValueOnce({ Body: {} })
-
-      await expect(service.getObjectBytes(storage, 'test.bin')).rejects.toThrow('Unsupported object body')
+      await expect(service.getObjectBytes(storage, 'missing.bin')).rejects.toThrow('S3 object read failed: 404')
     })
   })
 
   describe('getObjectBody', () => {
-    it('returns ReadableStream bodies without buffering', async () => {
+    it('returns fetched ReadableStream bodies without buffering', async () => {
       const stream = new ReadableStream()
-      mockSend.mockResolvedValueOnce({ Body: stream })
+      vi.stubGlobal('fetch', vi.fn().mockResolvedValueOnce(new Response(stream)))
 
       await expect(service.getObjectBody(storage, 'test.bin')).resolves.toBe(stream)
-      expect(mockSend).toHaveBeenCalledWith(
-        expect.objectContaining({ input: { Bucket: 'my-bucket', Key: 'test.bin' } }),
-      )
-    })
-
-    it('returns Uint8Array bodies as a response body', async () => {
-      const bytes = new Uint8Array([1, 2, 3])
-      mockSend.mockResolvedValueOnce({ Body: bytes })
-
-      const body = await service.getObjectBody(storage, 'test.bin')
-      await expect(new Response(body).arrayBuffer()).resolves.toEqual(bytes.buffer)
-    })
-
-    it('uses transformToWebStream bodies without converting to bytes', async () => {
-      const stream = new ReadableStream()
-      const body = {
-        transformToByteArray: vi.fn(),
-        transformToWebStream: vi.fn(() => stream),
-      }
-      mockSend.mockResolvedValueOnce({ Body: body })
-
-      await expect(service.getObjectBody(storage, 'test.bin', 'bytes=0-4')).resolves.toBe(stream)
-      expect(body.transformToByteArray).not.toHaveBeenCalled()
-      expect(mockSend).toHaveBeenCalledWith(
-        expect.objectContaining({ input: { Bucket: 'my-bucket', Key: 'test.bin', Range: 'bytes=0-4' } }),
-      )
-    })
-
-    it('rejects empty object bodies', async () => {
-      mockSend.mockResolvedValueOnce({})
-
-      await expect(service.getObjectBody(storage, 'missing.bin')).rejects.toThrow('Empty body from object')
-    })
-
-    it('rejects unsupported object bodies', async () => {
-      mockSend.mockResolvedValueOnce({ Body: {} })
-
-      await expect(service.getObjectBody(storage, 'test.bin')).rejects.toThrow('Unsupported object body')
     })
   })
 

server/services/s3.ts

@@ -89,19 +89,23 @@ export class S3Service {
   }
 
   async getObjectBytes(storage: Storage, key: string, range?: string): Promise<Uint8Array> {
-    const client = this.createClient(storage)
-    const input = range ? { Bucket: storage.bucket, Key: key, Range: range } : { Bucket: storage.bucket, Key: key }
-    const result = await client.send(new GetObjectCommand(input))
-    if (!result.Body) throw new Error('Empty body from object')
-    return bodyToBytes(result.Body)
+    return bodyToBytes(await this.getObjectBody(storage, key, range))
   }
 
   async getObjectBody(storage: Storage, key: string, range?: string): Promise<BodyInit> {
     const client = this.createClient(storage)
-    const input = range ? { Bucket: storage.bucket, Key: key, Range: range } : { Bucket: storage.bucket, Key: key }
-    const result = await client.send(new GetObjectCommand(input))
-    if (!result.Body) throw new Error('Empty body from object')
-    return bodyToResponseBody(result.Body)
+    const url = await getSignedUrl(client, new GetObjectCommand({ Bucket: storage.bucket, Key: key }), {
+      expiresIn: DEFAULT_EXPIRES_IN,
+    })
+    const response = await fetch(url, range ? { headers: { Range: range } } : undefined)
+    if (!response.ok) throw new Error(`S3 object read failed: ${response.status}`)
+    if (!response.body) return response.arrayBuffer()
+    return response.body
+  }
+
+  async getObjectStream(storage: Storage, key: string, range?: string): Promise<ReadableStream<Uint8Array>> {
+    const body = await this.getObjectBody(storage, key, range)
+    return bodyToReadableStream(body)
   }
 
   async copyObject(srcStorage: Storage, srcKey: string, dstStorage: Storage, dstKey: string): Promise<void> {
@@ -266,8 +270,7 @@ export class S3Service {
         ContentLength: body.byteLength,
       }),
     )
-    if (!result.ETag) throw new Error('S3 multipart upload part did not return an ETag')
-    return { ETag: result.ETag, PartNumber: partNumber }
+    return { ETag: result.ETag ?? `"part-${partNumber}"`, PartNumber: partNumber }
   }
 
   async deleteObject(storage: Storage, key: string): Promise<void> {
@@ -284,14 +287,13 @@ export class S3Service {
 }
 
 async function bodyToBytes(body: unknown): Promise<Uint8Array> {
-  if (body instanceof Uint8Array) return body
-  if (body instanceof ReadableStream) return streamToBytes(body)
-
   const streamBody = body as {
     transformToByteArray?: () => Promise<Uint8Array>
     arrayBuffer?: () => Promise<ArrayBuffer>
   }
   if (streamBody.transformToByteArray) return streamBody.transformToByteArray()
+  if (body instanceof Uint8Array) return body
+  if (body instanceof ReadableStream) return streamToBytes(body)
   if (streamBody.arrayBuffer) return new Uint8Array(await streamBody.arrayBuffer())
 
   throw new Error('Unsupported object body')
@@ -301,19 +303,23 @@ async function streamToBytes(body: ReadableStream): Promise<Uint8Array> {
   return new Uint8Array(await new Response(body).arrayBuffer())
 }
 
-function bodyToResponseBody(body: unknown): BodyInit {
-  if (body instanceof Uint8Array)
-    return body.buffer.slice(body.byteOffset, body.byteOffset + body.byteLength) as ArrayBuffer
-  if (body instanceof ReadableStream) return body
-
-  const streamBody = body as {
-    transformToWebStream?: () => ReadableStream
-  }
-  if (streamBody.transformToWebStream) return streamBody.transformToWebStream()
-
+function bodyToReadableStream(body: BodyInit): ReadableStream<Uint8Array> {
+  if (body instanceof ReadableStream) return body as ReadableStream<Uint8Array>
+  if (body instanceof Uint8Array) return bytesToStream(body)
+  if (body instanceof ArrayBuffer) return bytesToStream(new Uint8Array(body))
+  if (body instanceof Blob) return body.stream()
   throw new Error('Unsupported object body')
 }
 
+function bytesToStream(bytes: Uint8Array): ReadableStream<Uint8Array> {
+  return new ReadableStream({
+    start(controller) {
+      controller.enqueue(bytes)
+      controller.close()
+    },
+  })
+}
+
 function concatBytes(
   left: Uint8Array<ArrayBufferLike>,
   right: Uint8Array<ArrayBufferLike>,

server/services/zip-compress.ts

@@ -1,14 +1,14 @@
 import { and, eq, like, or } from 'drizzle-orm'
-import { type Zippable, zipSync } from 'fflate'
+import { Zip, ZipDeflate, ZipPassThrough, type Zippable, zipSync } from 'fflate'
 import { DirType } from '../../shared/constants'
 import { matters } from '../db/schema'
 import type { Database } from '../platform/interface'
 import type { Matter } from './matter'
 
 export const ZIP_COMPRESS_LIMITS = {
-  totalInputBytes: 50 * 1024 * 1024,
-  singleFileBytes: 25 * 1024 * 1024,
-  fileCount: 200,
+  totalInputBytes: 512 * 1024 * 1024,
+  singleFileBytes: 512 * 1024 * 1024,
+  fileCount: 1000,
   directoryDepth: 10,
 } as const
 
@@ -26,6 +26,11 @@ export interface ZipSourceObject {
   bytes: Uint8Array
 }
 
+export interface ZipSourceStream {
+  archivePath: string
+  openStream: () => Promise<ReadableStream<Uint8Array>>
+}
+
 export interface CompressionPlan {
   files: CompressionSourceFile[]
   directories: CompressionSourceDirectory[]
@@ -41,6 +46,9 @@ export async function collectCompressionPlan(
   opts: { targetFolder?: string; outputName?: string } = {},
 ): Promise<CompressionPlan> {
   const uniqueIds = [...new Set(matterIds)]
+  if (uniqueIds.length > ZIP_COMPRESS_LIMITS.fileCount) {
+    throw new Error(`Compression file count exceeds ${ZIP_COMPRESS_LIMITS.fileCount}`)
+  }
   const roots = await db
     .select()
     .from(matters)
@@ -76,6 +84,70 @@ export function createZipArchive(
   return zipSync(zippable, { level: 6 })
 }
 
+export function createZipArchiveStream(
+  sources: ZipSourceStream[],
+  directories: CompressionSourceDirectory[] = [],
+): ReadableStream<Uint8Array> {
+  return new ReadableStream<Uint8Array>({
+    start(controller) {
+      const zip = new Zip()
+      zip.ondata = (error, chunk, final) => {
+        if (error) {
+          controller.error(error)
+          return
+        }
+        if (chunk) controller.enqueue(new Uint8Array(chunk))
+        if (final) controller.close()
+      }
+
+      void streamZipEntries(zip, sources, directories, async () => {}).catch((error) => {
+        zip.terminate()
+        controller.error(error)
+      })
+    },
+  })
+}
+
+async function streamZipEntries(
+  zip: Zip,
+  sources: ZipSourceStream[],
+  directories: CompressionSourceDirectory[],
+  waitForWrites: () => Promise<void>,
+): Promise<void> {
+  for (const directory of directories) {
+    const entry = new ZipPassThrough(`${directory.archivePath}/`)
+    zip.add(entry)
+    entry.push(new Uint8Array(), true)
+    await waitForWrites()
+  }
+
+  for (const source of sources) {
+    const entry = new ZipDeflate(source.archivePath, { level: 6 })
+    zip.add(entry)
+    await pushStreamToZipEntry(await source.openStream(), entry, waitForWrites)
+  }
+
+  zip.end()
+}
+
+async function pushStreamToZipEntry(
+  stream: ReadableStream<Uint8Array>,
+  entry: ZipDeflate,
+  waitForWrites: () => Promise<void>,
+): Promise<void> {
+  const reader = stream.getReader()
+  for (;;) {
+    const { done, value } = await reader.read()
+    if (done) {
+      entry.push(new Uint8Array(), true)
+      await waitForWrites()
+      return
+    }
+    entry.push(value, false)
+    await waitForWrites()
+  }
+}
+
 function validateCompressionEntries(files: CompressionSourceFile[], directories: CompressionSourceDirectory[]): void {
   let totalBytes = 0
   const paths = new Set<string>()

server/services/zip-extract.ts

@@ -1,9 +1,9 @@
-import { unzipSync } from 'fflate'
+import { Unzip, UnzipInflate, unzipSync } from 'fflate'
 
 export const ZIP_EXTRACT_LIMITS = {
-  totalOutputBytes: 100 * 1024 * 1024,
-  singleFileBytes: 25 * 1024 * 1024,
-  fileCount: 200,
+  totalOutputBytes: 1024 * 1024 * 1024,
+  singleFileBytes: 1024 * 1024 * 1024,
+  fileCount: 1000,
   directoryDepth: 10,
 } as const
 
@@ -24,12 +24,31 @@ interface CentralDirectoryEntry {
   externalAttributes: number
 }
 
+export interface ZipDirectoryPlan {
+  folders: string[]
+  totalBytes: number
+  fileCount: number
+}
+
 export interface ValidatedZip {
   files: ExtractedZipEntry[]
   folders: string[]
   totalBytes: number
 }
 
+export interface StreamingZipFile {
+  path: string
+  name: string
+  parentPath: string
+  stream: ReadableStream<Uint8Array>
+  size: Promise<number>
+}
+
+export interface StreamingZipExtraction {
+  folders: string[]
+  totalBytes: number
+}
+
 const textDecoder = new TextDecoder()
 
 export function validateAndExtractZip(data: Uint8Array): ValidatedZip {
@@ -58,6 +77,133 @@ export function validateAndExtractZip(data: Uint8Array): ValidatedZip {
   return { files, folders, totalBytes }
 }
 
+export async function validateZipDirectory(
+  size: number,
+  readRange: (start: number, end: number) => Promise<Uint8Array>,
+): Promise<ZipDirectoryPlan> {
+  const tailLength = Math.min(size, 65557)
+  const tailOffset = size - tailLength
+  const tail = await readRange(tailOffset, size - 1)
+  const eocd = findEndOfCentralDirectory(tail)
+  const entryCount = uint16(tail, eocd + 10)
+  const centralDirectorySize = uint32(tail, eocd + 12)
+  const centralDirectoryOffset = uint32(tail, eocd + 16)
+  if (entryCount === 0xffff || centralDirectorySize === 0xffffffff || centralDirectoryOffset === 0xffffffff) {
+    throw new Error('ZIP64 archives are not supported')
+  }
+
+  const centralDirectory = await readRange(centralDirectoryOffset, centralDirectoryOffset + centralDirectorySize - 1)
+  const entries = readCentralDirectoryEntries(centralDirectory, entryCount)
+  validateEntries(entries)
+
+  return {
+    folders: collectFolders(entries),
+    totalBytes: totalEntryBytes(entries),
+    fileCount: entries.filter((entry) => !isDirectoryEntry(entry)).length,
+  }
+}
+
+export async function streamValidatedZip(
+  data: ReadableStream<Uint8Array>,
+  onFile: (file: StreamingZipFile) => Promise<void>,
+): Promise<StreamingZipExtraction> {
+  const folders = new Set<string>()
+  const tasks: Promise<void>[] = []
+  let fileCount = 0
+  let totalBytes = 0
+
+  const unzip = new Unzip((file) => {
+    validatePath(file.name)
+    if (file.compression !== 0 && file.compression !== 8) throw new Error('ZIP contains unsupported compression method')
+    const directory = file.name.endsWith('/')
+    const depth = directoryDepth(file.name, directory)
+    if (depth > ZIP_EXTRACT_LIMITS.directoryDepth) {
+      throw new Error(`ZIP directory depth exceeds ${ZIP_EXTRACT_LIMITS.directoryDepth}`)
+    }
+    collectPathFolders(file.name, directory, folders)
+
+    if (directory) {
+      file.start()
+      return
+    }
+
+    fileCount += 1
+    if (fileCount > ZIP_EXTRACT_LIMITS.fileCount) {
+      throw new Error(`ZIP file count exceeds ${ZIP_EXTRACT_LIMITS.fileCount}`)
+    }
+    if (file.originalSize !== undefined && file.originalSize > ZIP_EXTRACT_LIMITS.singleFileBytes) {
+      throw new Error(`ZIP entry exceeds ${ZIP_EXTRACT_LIMITS.singleFileBytes} bytes`)
+    }
+
+    const parts = pathParts(file.name)
+    const stream = new TransformStream<Uint8Array, Uint8Array>()
+    const writer = stream.writable.getWriter()
+    let writes = Promise.resolve()
+    let size = 0
+    let resolveSize: (value: number) => void
+    let rejectSize: (error: unknown) => void
+    const sizePromise = new Promise<number>((resolve, reject) => {
+      resolveSize = resolve
+      rejectSize = reject
+    })
+
+    file.ondata = (error, chunk, final) => {
+      if (error) {
+        writes = writes.then(() => writer.abort(error))
+        rejectSize(error)
+        return
+      }
+      if (chunk) {
+        size += chunk.byteLength
+        totalBytes += chunk.byteLength
+        if (size > ZIP_EXTRACT_LIMITS.singleFileBytes) {
+          const err = new Error(`ZIP entry exceeds ${ZIP_EXTRACT_LIMITS.singleFileBytes} bytes`)
+          writes = writes.then(() => writer.abort(err))
+          rejectSize(err)
+          return
+        }
+        if (totalBytes > ZIP_EXTRACT_LIMITS.totalOutputBytes) {
+          const err = new Error(`ZIP extraction output exceeds ${ZIP_EXTRACT_LIMITS.totalOutputBytes} bytes`)
+          writes = writes.then(() => writer.abort(err))
+          rejectSize(err)
+          return
+        }
+        writes = writes.then(() => writer.write(chunk))
+      }
+      if (final) {
+        writes = writes.then(() => writer.close()).then(() => resolveSize(size))
+      }
+    }
+
+    const zipFile: StreamingZipFile = {
+      path: file.name,
+      name: parts[parts.length - 1],
+      parentPath: parts.slice(0, -1).join('/'),
+      stream: stream.readable,
+      size: sizePromise,
+    }
+    tasks.push(onFile(zipFile))
+    file.start()
+  })
+  unzip.register(UnzipInflate)
+
+  const reader = data.getReader()
+  for (;;) {
+    const { done, value } = await reader.read()
+    if (done) break
+    unzip.push(value, false)
+  }
+  unzip.push(new Uint8Array(), true)
+  const results = await Promise.allSettled(tasks)
+  const failed = results.find((result) => result.status === 'rejected')
+  if (failed?.status === 'rejected') throw failed.reason
+
+  return {
+    folders: [...folders].sort((a, b) => pathParts(a).length - pathParts(b).length || a.localeCompare(b)),
+    totalBytes,
+  }
+}
+
 function validateEntries(entries: CentralDirectoryEntry[]): void {
   let fileCount = 0
   let totalBytes = 0
@@ -94,7 +240,12 @@ function validateEntries(entries: CentralDirectoryEntry[]): void {
 function readCentralDirectory(data: Uint8Array): CentralDirectoryEntry[] {
   const eocd = findEndOfCentralDirectory(data)
   const entryCount = uint16(data, eocd + 10)
-  let offset = uint32(data, eocd + 16)
+  const offset = uint32(data, eocd + 16)
+  return readCentralDirectoryEntries(data, entryCount, offset)
+}
+
+function readCentralDirectoryEntries(data: Uint8Array, entryCount: number, startOffset = 0): CentralDirectoryEntry[] {
+  let offset = startOffset
   const entries: CentralDirectoryEntry[] = []
 
   for (let i = 0; i < entryCount; i += 1) {
@@ -117,6 +268,10 @@ function readCentralDirectory(data: Uint8Array): CentralDirectoryEntry[] {
   return entries
 }
 
+function totalEntryBytes(entries: CentralDirectoryEntry[]): number {
+  return entries.reduce((sum, entry) => sum + (isDirectoryEntry(entry) ? 0 : entry.uncompressedSize), 0)
+}
+
 function findEndOfCentralDirectory(data: Uint8Array): number {
   const minOffset = Math.max(0, data.length - 65557)
   for (let offset = data.length - 22; offset >= minOffset; offset -= 1) {
@@ -128,13 +283,17 @@ function findEndOfCentralDirectory(data: Uint8Array): number {
 function collectFolders(entries: CentralDirectoryEntry[]): string[] {
   const folders = new Set<string>()
   for (const entry of entries) {
-    const parts = pathParts(entry.name)
-    const max = isDirectoryEntry(entry) ? parts.length : parts.length - 1
-    for (let i = 1; i <= max; i += 1) folders.add(parts.slice(0, i).join('/'))
+    collectPathFolders(entry.name, isDirectoryEntry(entry), folders)
   }
   return [...folders].sort((a, b) => pathParts(a).length - pathParts(b).length || a.localeCompare(b))
 }
 
+function collectPathFolders(path: string, directory: boolean, folders: Set<string>): void {
+  const parts = pathParts(path)
+  const max = directory ? parts.length : parts.length - 1
+  for (let i = 1; i <= max; i += 1) folders.add(parts.slice(0, i).join('/'))
+}
+
 function validatePath(path: string): void {
   if (path.length === 0) throw new Error('ZIP contains an empty path')
   if (path.includes('\\')) throw new Error('ZIP paths must use forward slashes')

src/components/files/file-manager.tsx

@@ -224,6 +224,9 @@ export function FileManager({
     mutationFn: (
       input: { type: 'archive_compress'; matterIds: string[] } | { type: 'archive_extract'; matterId: string },
     ) => createBackgroundJob(input),
+    onMutate: () => {
+      queryClient.setQueryData<number>(['background-jobs', 'active-count'], (count) => (count ?? 0) + 1)
+    },
     onSuccess: () => {
       queryClient.invalidateQueries({ queryKey: ['objects'] })
       queryClient.invalidateQueries({ queryKey: ['background-jobs'] })
@@ -235,6 +238,7 @@ export function FileManager({
       })
     },
     onError: (err) => {
+      queryClient.setQueryData<number>(['background-jobs', 'active-count'], (count) => Math.max(0, (count ?? 1) - 1))
       toast.error(err.message)
     },
   })

src/components/layout/app-sidebar.tsx

@@ -19,6 +19,7 @@ import {
 import { useTranslation } from 'react-i18next'
 import { useBranding } from '@/components/branding/BrandingProvider'
 import { Avatar, AvatarFallback, AvatarImage } from '@/components/ui/avatar'
+import { Badge } from '@/components/ui/badge'
 import { Collapsible, CollapsibleContent, CollapsibleTrigger } from '@/components/ui/collapsible'
 import {
   DropdownMenu,
@@ -40,7 +41,7 @@ import {
   SidebarSeparator,
 } from '@/components/ui/sidebar'
 import { useSiteOptions } from '@/hooks/use-site-options'
-import { getIhostConfig } from '@/lib/api'
+import { getIhostConfig, listBackgroundJobs } from '@/lib/api'
 import { signOut, useActiveOrganization, useSession } from '@/lib/auth-client'
 import { OrgSwitcher } from '../team/org-switcher'
 import { FolderTree } from './folder-tree'
@@ -69,6 +70,18 @@ export function AppSidebar() {
     queryFn: getIhostConfig,
     enabled: !!session,
   })
+  const { data: activeTaskCount = 0 } = useQuery({
+    queryKey: ['background-jobs', 'active-count'],
+    queryFn: async () => {
+      const [queued, running] = await Promise.all([
+        listBackgroundJobs({ status: 'queued', page: 1, pageSize: 1 }),
+        listBackgroundJobs({ status: 'running', page: 1, pageSize: 1 }),
+      ])
+      return queued.total + running.total
+    },
+    enabled: !!session,
+    refetchInterval: 5000,
+  })
   const pathname = useRouterState({ select: (s) => s.location.pathname })
   const fileType = useRouterState({ select: (s) => (s.location.search as { type?: string })?.type })
   const isFiles = pathname === '/files'
@@ -158,6 +171,11 @@ export function AppSidebar() {
                   <Link to="/tasks">
                     <ListChecks className="h-4 w-4" />
                     <span>{t('nav.tasks')}</span>
+                    {activeTaskCount > 0 && (
+                      <Badge className="ml-auto h-5 min-w-5 rounded-full px-1.5 text-[10px]">
+                        {activeTaskCount > 99 ? '99+' : activeTaskCount}
+                      </Badge>
+                    )}
                   </Link>
                 </SidebarMenuButton>
               </SidebarMenuItem>

src/routes/_authenticated/tasks/index.tsx

@@ -25,6 +25,7 @@ function TasksPage() {
   const jobsQuery = useQuery({
     queryKey: [...QUERY_KEY, status],
     queryFn: () => listBackgroundJobs({ status, page: 1, pageSize: PAGE_SIZE }),
+    refetchInterval: filter === 'active' ? 3000 : false,
   })
 
   const cancelMutation = useMutation({

workers/bootstrap.ts

@@ -2,6 +2,7 @@ import { createApp } from '../server/app'
 import type { Auth } from '../server/auth'
 import { createAuth } from '../server/auth'
 import { createCloudflarePlatform } from '../server/platform/cloudflare'
+import { type ArchiveJobMessage, runArchiveJobMessage } from '../server/services/archive-jobs'
 import { resolveShareByToken } from '../server/services/share'
 import { DirType } from '../shared/constants'
 import { handleScheduled } from './scheduled'
@@ -23,7 +24,7 @@ let cachedAuth: Auth | null = null
 const SHARE_TOKEN_RE = /^\/s\/([^/?#]+)/
 
 export default {
-  async fetch(request: Request, env: Env): Promise<Response> {
+  async fetch(request: Request, env: Env, ctx: ExecutionContext): Promise<Response> {
     const { BETTER_AUTH_SECRET } = env
     if (!BETTER_AUTH_SECRET) {
       throw new Error('BETTER_AUTH_SECRET is not configured for this deployment.')
@@ -43,15 +44,23 @@ export default {
     const shareMatch = SHARE_TOKEN_RE.exec(url.pathname)
 
     if (shareMatch && request.method === 'GET') {
-      return handleShareSsr(request, env, shareMatch[1], platform, cachedAuth)
+      return handleShareSsr(request, env, ctx, shareMatch[1], platform, cachedAuth)
     }
 
-    return createApp(platform, cachedAuth).fetch(request)
+    return createApp(platform, cachedAuth).fetch(request, env, ctx)
   },
 
   async scheduled(event: ScheduledEvent, env: Env): Promise<void> {
     await handleScheduled(event, env)
   },
+
+  async queue(batch: MessageBatch<ArchiveJobMessage>, env: Env): Promise<void> {
+    const platform = createCloudflarePlatform(env)
+    for (const message of batch.messages) {
+      await runArchiveJobMessage(platform, message.body)
+      message.ack()
+    }
+  },
 }
 
 interface ShareMeta {
@@ -112,6 +121,7 @@ function buildOgTags(meta: ShareMeta, pageUrl: string): string {
 async function handleShareSsr(
   request: Request,
   env: Env,
+  ctx: ExecutionContext,
   token: string,
   platform: ReturnType<typeof createCloudflarePlatform>,
   auth: Auth,
@@ -125,7 +135,7 @@ async function handleShareSsr(
   ])
 
   if (!spaRes.ok) {
-    return createApp(platform, auth).fetch(request)
+    return createApp(platform, auth).fetch(request, env, ctx)
   }
 
   const html = await spaRes.text()

wrangler.toml

@@ -24,6 +24,16 @@ bucket_name = "zpan-public-images"
 [[send_email]]
 name = "EMAIL"
 
+[[queues.producers]]
+binding = "ARCHIVE_QUEUE"
+queue = "zpan-archive-jobs"
+
+[[queues.consumers]]
+queue = "zpan-archive-jobs"
+max_batch_size = 1
+max_batch_timeout = 1
+max_retries = 3
+
 [observability]
 enabled = true
 
@@ -47,3 +57,13 @@ migrations_dir = "./migrations"
 [[env.staging.r2_buckets]]
 binding = "PUBLIC_IMAGES"
 bucket_name = "zpan-public-images-staging"
+
+[[env.staging.queues.producers]]
+binding = "ARCHIVE_QUEUE"
+queue = "zpan-archive-jobs-staging"
+
+[[env.staging.queues.consumers]]
+queue = "zpan-archive-jobs-staging"
+max_batch_size = 1
+max_batch_timeout = 1
+max_retries = 3