From 2438275ea81e0583b08c8cfe46b3e82ca22b9853 Mon Sep 17 00:00:00 2001 From: saltbo Date: Fri, 15 May 2026 09:59:08 -0400 Subject: [PATCH] feat(archive): queue streaming archive jobs --- e2e/archive.spec.ts | 111 +++++++++ e2e/global-setup.ts | 28 ++- package.json | 4 +- playwright.config.ts | 13 ++ scripts/run-cloud-e2e.mjs | 44 +++- scripts/s3-mock.mjs | 219 ++++++++++++++++++ .../background-jobs.integration.test.ts | 131 ++++++++++- server/routes/background-jobs.ts | 26 ++- server/services/archive-jobs.ts | 67 ++++++ server/services/archive-processing.test.ts | 162 +++++++++++-- server/services/archive-processing.ts | 157 ++++++++----- server/services/s3.test.ts | 103 ++------ server/services/s3.ts | 54 +++-- server/services/zip-compress.ts | 80 ++++++- server/services/zip-extract.ts | 175 +++++++++++++- src/components/files/file-manager.tsx | 4 + src/components/layout/app-sidebar.tsx | 20 +- src/routes/_authenticated/tasks/index.tsx | 1 + workers/bootstrap.ts | 18 +- wrangler.toml | 20 ++ 20 files changed, 1212 insertions(+), 225 deletions(-) create mode 100644 e2e/archive.spec.ts create mode 100644 scripts/s3-mock.mjs create mode 100644 server/services/archive-jobs.ts diff --git a/e2e/archive.spec.ts b/e2e/archive.spec.ts new file mode 100644 index 00000000..e059b53e --- /dev/null +++ b/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 { + 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 + 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') +} diff --git a/e2e/global-setup.ts b/e2e/global-setup.ts index 4404b21d..d3ea9556 100644 --- a/e2e/global-setup.ts +++ b/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 diff --git a/package.json b/package.json index 737ebc22..37c51303 100644 --- a/package.json +++ b/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" diff --git a/playwright.config.ts b/playwright.config.ts index af866091..187f7e06 100644 --- a/playwright.config.ts +++ b/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, diff --git a/scripts/run-cloud-e2e.mjs b/scripts/run-cloud-e2e.mjs index 3774d253..16b6baa4 100644 --- a/scripts/run-cloud-e2e.mjs +++ b/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'], diff --git a/scripts/s3-mock.mjs b/scripts/s3-mock.mjs new file mode 100644 index 00000000..b28b9e80 --- /dev/null +++ b/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(`${uploadId}`) +} + +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('') +} + +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)) +} diff --git a/server/routes/background-jobs.integration.test.ts b/server/routes/background-jobs.integration.test.ts index 3a8cec9b..7b2bbe2a 100644 --- a/server/routes/background-jobs.integration.test.ts +++ b/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([['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([['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 { `) } +async function waitForJob( + db: TestDb, + orgId: string, + jobId: string, + status: 'completed' | 'failed', +): Promise>> { + 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): 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 diff --git a/server/routes/background-jobs.ts b/server/routes/background-jobs.ts index 3d46bb73..59137ad2 100644 --- a/server/routes/background-jobs.ts +++ b/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() 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() 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() export default backgroundJobs -function requireOrg(c: Context): string { +function requireOrg(c: { get(key: 'orgId'): string | null }): string { const orgId = c.get('orgId') if (!orgId) throw new BackgroundJobError('not_found') return orgId diff --git a/server/services/archive-jobs.ts b/server/services/archive-jobs.ts new file mode 100644 index 00000000..6b0253df --- /dev/null +++ b/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 +} + +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 { + 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 { + const queue = platform.getBinding(ARCHIVE_QUEUE_BINDING) + if (queue) { + await queue.send(message) + return + } + + localArchiveQueue.push(platform, message) +} + +export async function runArchiveJobMessage(platform: Platform, message: ArchiveJobMessage): Promise { + await processArchiveJob(platform.db, { + orgId: message.orgId, + userId: message.userId, + request: message.request, + jobId: message.jobId, + }) +} diff --git a/server/services/archive-processing.test.ts b/server/services/archive-processing.test.ts index 2457f328..a41543ca 100644 --- a/server/services/archive-processing.test.ts +++ b/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>['db'] @@ -16,12 +16,29 @@ class MemoryS3 { objects = new Map() putKeys: string[] = [] - async getObjectBytes(_storage: unknown, key: string): Promise { + async getObjectBytes(_storage: unknown, key: string, range?: string): Promise { 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> { + 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 { 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() + + constructor( + private readonly generatedKey: string, + private readonly generatedSize: number, + ) { + super() + } + + override async getObjectStream(_storage: unknown, key: string): Promise> { + if (key === this.generatedKey) return generatedBytes(this.generatedSize) + return super.getObjectStream(_storage, key) + } + + override async putObject(_storage: unknown, key: string, body: Uint8Array | ReadableStream): Promise { + 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 { + 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 { + 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): Promise { + 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 diff --git a/server/services/archive-processing.ts b/server/services/archive-processing.ts index 59ad87db..eb65de2a 100644 --- a/server/services/archive-processing.ts +++ b/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 { + const job = await enqueueArchiveJob(db, input) + return processArchiveJob(db, { ...input, jobId: job.id }) +} + +export async function enqueueArchiveJob(db: Database, input: CreateArchiveJobInput): Promise { 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 { 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() + let outputBytes = 0 try { - const folderParents = new Map() - 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 { + 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 { @@ -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 { + 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 }), + }) +} diff --git a/server/services/s3.test.ts b/server/services/s3.test.ts index 06a63d47..71020c09 100644 --- a/server/services/s3.test.ts +++ b/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') }) }) diff --git a/server/services/s3.ts b/server/services/s3.ts index 214c05f6..10d653ea 100644 --- a/server/services/s3.ts +++ b/server/services/s3.ts @@ -89,19 +89,23 @@ export class S3Service { } async getObjectBytes(storage: Storage, key: string, range?: string): Promise { - 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 { 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> { + const body = await this.getObjectBody(storage, key, range) + return bodyToReadableStream(body) } async copyObject(srcStorage: Storage, srcKey: string, dstStorage: Storage, dstKey: string): Promise { @@ -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 { @@ -284,14 +287,13 @@ export class S3Service { } async function bodyToBytes(body: unknown): Promise { - if (body instanceof Uint8Array) return body - if (body instanceof ReadableStream) return streamToBytes(body) - const streamBody = body as { transformToByteArray?: () => Promise arrayBuffer?: () => Promise } 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 { 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 { + if (body instanceof ReadableStream) return body as ReadableStream + 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 { + return new ReadableStream({ + start(controller) { + controller.enqueue(bytes) + controller.close() + }, + }) +} + function concatBytes( left: Uint8Array, right: Uint8Array, diff --git a/server/services/zip-compress.ts b/server/services/zip-compress.ts index d64a4cff..d25b6690 100644 --- a/server/services/zip-compress.ts +++ b/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> +} + export interface CompressionPlan { files: CompressionSourceFile[] directories: CompressionSourceDirectory[] @@ -41,6 +46,9 @@ export async function collectCompressionPlan( opts: { targetFolder?: string; outputName?: string } = {}, ): Promise { 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 { + return new ReadableStream({ + 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, +): Promise { + 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, + entry: ZipDeflate, + waitForWrites: () => Promise, +): Promise { + 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() diff --git a/server/services/zip-extract.ts b/server/services/zip-extract.ts index 5080e6d9..82d14cd3 100644 --- a/server/services/zip-extract.ts +++ b/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 + size: Promise +} + +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, +): Promise { + 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, + onFile: (file: StreamingZipFile) => Promise, +): Promise { + const folders = new Set() + const tasks: Promise[] = [] + 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() + 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((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() 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): 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') diff --git a/src/components/files/file-manager.tsx b/src/components/files/file-manager.tsx index 60ab2549..81fcb8a9 100644 --- a/src/components/files/file-manager.tsx +++ b/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(['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(['background-jobs', 'active-count'], (count) => Math.max(0, (count ?? 1) - 1)) toast.error(err.message) }, }) diff --git a/src/components/layout/app-sidebar.tsx b/src/components/layout/app-sidebar.tsx index 62c376ac..0a66f39b 100644 --- a/src/components/layout/app-sidebar.tsx +++ b/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() { {t('nav.tasks')} + {activeTaskCount > 0 && ( + + {activeTaskCount > 99 ? '99+' : activeTaskCount} + + )} diff --git a/src/routes/_authenticated/tasks/index.tsx b/src/routes/_authenticated/tasks/index.tsx index 3d62c227..f38d083e 100644 --- a/src/routes/_authenticated/tasks/index.tsx +++ b/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({ diff --git a/workers/bootstrap.ts b/workers/bootstrap.ts index 9c6cd966..cb0d7ae7 100644 --- a/workers/bootstrap.ts +++ b/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 { + async fetch(request: Request, env: Env, ctx: ExecutionContext): Promise { 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 { await handleScheduled(event, env) }, + + async queue(batch: MessageBatch, env: Env): Promise { + 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, 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() diff --git a/wrangler.toml b/wrangler.toml index a95da621..db5b6f0f 100644 --- a/wrangler.toml +++ b/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