perf(files): stream workspace archives instead of buffering them

The bulk download route materialized every selected file before writing a byte,
so peak memory tracked the size of the selection. Ordinary files are now
appended as lazy streams: each opens its storage read only when the archiver
reaches it, so one entry is resident at a time rather than the whole archive.

Generated documents still resolve to buffers first. They are the only entries
whose bytes decide anything, and every status this route returns comes from
them — once the first byte is written the status code is committed, so those
decisions have to happen before the archive starts. The per-entry allowance and
byte budget therefore govern documents only.

archiver processes appended entries through a sequential queue; lazystream
defers each storage read until that entry's turn, since handing the archiver an
open stream per entry would hold more connections than the storage client pools.
nodeReadableToWebStream moves out of input-validation.server.ts into a shared
util rather than being written twice: Readable.toWeb throws ERR_INVALID_STATE
when a consumer cancels while the source is still flowing, which is exactly what
a cancelled download does.

Trade: a storage read failing mid-archive truncates the response rather than
returning 500, since the status is already sent. Documents cannot hit this. The
table export route already behaves this way.

Author: waleedlatif1

apps/sim/app/api/workspaces/[id]/files/download/route.test.ts

@@ -0,0 +1,272 @@
+/**
+ * @vitest-environment node
+ */
+import { Readable } from 'stream'
+import { createMockRequest } from '@sim/testing'
+import JSZip from 'jszip'
+import { beforeEach, describe, expect, it, vi } from 'vitest'
+
+const {
+  mockGetSession,
+  mockVerifyWorkspaceMembership,
+  mockListWorkspaceFiles,
+  mockListWorkspaceFileFolders,
+  mockFetchServableWorkspaceFileBuffer,
+  mockDownloadFileStream,
+} = vi.hoisted(() => ({
+  mockGetSession: vi.fn(),
+  mockVerifyWorkspaceMembership: vi.fn(),
+  mockListWorkspaceFiles: vi.fn(),
+  mockListWorkspaceFileFolders: vi.fn(),
+  mockFetchServableWorkspaceFileBuffer: vi.fn(),
+  mockDownloadFileStream: vi.fn(),
+}))
+
+vi.mock('@/lib/auth', () => ({
+  auth: { api: { getSession: vi.fn() } },
+  getSession: mockGetSession,
+}))
+
+vi.mock('@/app/api/workflows/utils', () => ({
+  verifyWorkspaceMembership: mockVerifyWorkspaceMembership,
+}))
+
+vi.mock('@/lib/uploads/contexts/workspace', () => ({
+  listWorkspaceFiles: mockListWorkspaceFiles,
+  listWorkspaceFileFolders: mockListWorkspaceFileFolders,
+  buildWorkspaceFileFolderPathMap: (folders: Array<{ id: string; name: string }>) =>
+    new Map(folders.map((folder) => [folder.id, folder.name])),
+  fetchServableWorkspaceFileBuffer: mockFetchServableWorkspaceFileBuffer,
+}))
+
+vi.mock('@/lib/uploads/core/storage-service', () => ({
+  downloadFileStream: mockDownloadFileStream,
+}))
+
+vi.mock('@sim/audit', () => ({
+  recordAudit: vi.fn(),
+  AuditAction: { FILE_DOWNLOADED: 'file.downloaded' },
+  AuditResourceType: { FILE: 'file' },
+}))
+
+vi.mock('@/lib/posthog/server', () => ({ captureServerEvent: vi.fn() }))
+
+import { DocCompileUserError } from '@/lib/copilot/tools/server/files/doc-compile'
+import { PayloadSizeLimitError } from '@/lib/core/utils/stream-limits'
+import { GET } from '@/app/api/workspaces/[id]/files/download/route'
+
+const WORKSPACE_ID = 'ws-1'
+const context = { params: Promise.resolve({ id: WORKSPACE_ID }) }
+const MB = 1024 * 1024
+
+function workspaceFile(id: string, name: string, folderId: string | null = 'folder-1') {
+  return {
+    id,
+    name,
+    key: `workspace/${WORKSPACE_ID}/${id}`,
+    path: `/serve/${id}`,
+    size: 100,
+    type: 'application/octet-stream',
+    folderId,
+  }
+}
+
+function requestFor(query: string) {
+  return createMockRequest(
+    'GET',
+    undefined,
+    {},
+    `http://localhost:3000/api/workspaces/${WORKSPACE_ID}/files/download?${query}`
+  )
+}
+
+async function zipFrom(response: Response) {
+  return JSZip.loadAsync(Buffer.from(await response.arrayBuffer()))
+}
+
+describe('workspace files download route', () => {
+  beforeEach(() => {
+    vi.clearAllMocks()
+    mockGetSession.mockResolvedValue({ user: { id: 'user-1' } })
+    mockVerifyWorkspaceMembership.mockResolvedValue({ role: 'member' })
+    mockListWorkspaceFileFolders.mockResolvedValue([
+      { id: 'folder-1', name: 'Reports', parentId: null },
+    ])
+    mockDownloadFileStream.mockImplementation(async () => Readable.from([Buffer.from('plain')]))
+  })
+
+  it('zips the rendered bytes for a generated doc, not its stored source', async () => {
+    mockListWorkspaceFiles.mockResolvedValue([workspaceFile('f1', 'overview.docx')])
+    // A real .docx is a ZIP; the stored source would be plain JS text.
+    const rendered = Buffer.from('PKrendered-docx')
+    mockFetchServableWorkspaceFileBuffer.mockResolvedValue({
+      buffer: rendered,
+      contentType: 'application/vnd.openxmlformats-officedocument.wordprocessingml.document',
+    })
+
+    const response = await GET(requestFor('fileIds=f1'), context)
+
+    expect(response.status).toBe(200)
+    const entry = (await zipFrom(response)).file('Reports/overview.docx')
+    expect(entry).not.toBeNull()
+    expect(Buffer.from(await entry!.async('uint8array'))).toEqual(rendered)
+  })
+
+  it('streams ordinary files instead of materializing them', async () => {
+    mockListWorkspaceFiles.mockResolvedValue([workspaceFile('f1', 'clip.mp4')])
+
+    const response = await GET(requestFor('fileIds=f1'), context)
+
+    expect(response.status).toBe(200)
+    // Nothing has been read yet: the entry opens its storage read only once the
+    // consumer pulls the archive, which is what keeps peak memory to one entry.
+    expect(mockDownloadFileStream).not.toHaveBeenCalled()
+
+    const zip = await zipFrom(response)
+
+    expect(mockDownloadFileStream).toHaveBeenCalledTimes(1)
+    // Never routed through the buffering document reader.
+    expect(mockFetchServableWorkspaceFileBuffer).not.toHaveBeenCalled()
+
+    const entry = zip.file('Reports/clip.mp4')
+    expect(entry).not.toBeNull()
+    expect(await entry!.async('string')).toBe('plain')
+  })
+
+  it('preserves nested folder paths across both entry kinds', async () => {
+    mockListWorkspaceFileFolders.mockResolvedValue([
+      { id: 'folder-1', name: 'Reports', parentId: null },
+      { id: 'folder-2', name: 'visuals', parentId: 'folder-1' },
+    ])
+    mockListWorkspaceFiles.mockResolvedValue([
+      workspaceFile('f1', 'summary.docx', 'folder-1'),
+      workspaceFile('f2', 'hero.png', 'folder-2'),
+    ])
+    mockFetchServableWorkspaceFileBuffer.mockResolvedValue({
+      buffer: Buffer.from('PKdoc'),
+      contentType: 'application/vnd.openxmlformats-officedocument.wordprocessingml.document',
+    })
+
+    const zip = await zipFrom(await GET(requestFor('fileIds=f1&fileIds=f2'), context))
+
+    expect(zip.file('Reports/summary.docx')).not.toBeNull()
+    expect(zip.file('visuals/hero.png')).not.toBeNull()
+  })
+
+  it('returns 409 naming the documents whose artifacts are still compiling', async () => {
+    mockListWorkspaceFiles.mockResolvedValue([
+      workspaceFile('f1', 'ready.docx'),
+      workspaceFile('f2', 'pending.docx'),
+    ])
+    mockFetchServableWorkspaceFileBuffer.mockImplementation(async (file: { name: string }) => {
+      if (file.name === 'pending.docx')
+        throw new DocCompileUserError('Document is still being generated')
+      return { buffer: Buffer.from('PKok'), contentType: 'application/octet-stream' }
+    })
+
+    const response = await GET(requestFor('fileIds=f1&fileIds=f2'), context)
+
+    expect(response.status).toBe(409)
+    const body = await response.json()
+    expect(body.error).toContain('pending.docx')
+    expect(body.error).not.toContain('ready.docx')
+  })
+
+  it('rejects with 400, not 500, when a document blows its own allowance', async () => {
+    mockListWorkspaceFiles.mockResolvedValue([workspaceFile('f1', 'huge.docx')])
+    mockFetchServableWorkspaceFileBuffer.mockRejectedValue(
+      new PayloadSizeLimitError({ label: 'servable file download', maxBytes: 1 })
+    )
+
+    const response = await GET(requestFor('fileIds=f1'), context)
+
+    expect(response.status).toBe(400)
+    const body = await response.json()
+    expect(body.error).toContain('huge.docx')
+    expect(body.error).not.toContain('Selected files total')
+  })
+
+  it('blames the shared budget once earlier documents have consumed it', async () => {
+    mockListWorkspaceFiles.mockResolvedValue([
+      workspaceFile('f1', 'first.docx'),
+      workspaceFile('f2', 'second.docx'),
+    ])
+    mockFetchServableWorkspaceFileBuffer.mockImplementation(async (file: { name: string }) => {
+      // The first document eats the whole budget, so the second's cap is the remainder.
+      if (file.name === 'first.docx') {
+        return { buffer: Buffer.alloc(240 * MB), contentType: 'application/octet-stream' }
+      }
+      throw new PayloadSizeLimitError({ label: 'servable file download', maxBytes: 1 })
+    })
+
+    const response = await GET(requestFor('fileIds=f1&fileIds=f2'), context)
+
+    expect(response.status).toBe(400)
+    const body = await response.json()
+    expect(body.error).toContain('Selected files total')
+    expect(body.error).not.toContain('second.docx')
+  })
+
+  it('lets an uploaded office file larger than the render headroom through', async () => {
+    const big = { ...workspaceFile('f1', 'deck.pptx'), size: 80 * MB }
+    mockListWorkspaceFiles.mockResolvedValue([big])
+    mockFetchServableWorkspaceFileBuffer.mockResolvedValue({
+      buffer: Buffer.from('PKdeck'),
+      contentType: 'application/vnd.openxmlformats-officedocument.presentationml.presentation',
+    })
+
+    const response = await GET(requestFor('fileIds=f1'), context)
+
+    expect(response.status).toBe(200)
+    // Capped at the declared size, not the smaller render headroom.
+    expect(mockFetchServableWorkspaceFileBuffer.mock.calls[0][1].maxBytes).toBe(80 * MB)
+  })
+
+  it('surfaces a storage failure as a 500 even when another document is pending', async () => {
+    mockListWorkspaceFiles.mockResolvedValue([
+      workspaceFile('f1', 'pending.docx'),
+      workspaceFile('f2', 'broken.docx'),
+    ])
+    mockFetchServableWorkspaceFileBuffer.mockImplementation(async (file: { name: string }) => {
+      if (file.name === 'pending.docx')
+        throw new DocCompileUserError('Document is still being generated')
+      throw new Error('storage down')
+    })
+
+    const response = await GET(requestFor('fileIds=f1&fileIds=f2'), context)
+
+    // A 409 would tell the client to retry something that can never succeed.
+    expect(response.status).toBe(500)
+  })
+
+  it('stops resolving documents once one hard-fails', async () => {
+    const files = Array.from({ length: 20 }, (_, index) =>
+      workspaceFile(`f${index}`, `doc${index}.docx`)
+    )
+    mockListWorkspaceFiles.mockResolvedValue(files)
+    mockFetchServableWorkspaceFileBuffer.mockImplementation(async (file: { name: string }) => {
+      if (file.name === 'doc0.docx') throw new Error('storage down')
+      return { buffer: Buffer.from('PKok'), contentType: 'application/octet-stream' }
+    })
+
+    const response = await GET(
+      requestFor(files.map((file) => `fileIds=${file.id}`).join('&')),
+      context
+    )
+
+    expect(response.status).toBe(500)
+    expect(mockFetchServableWorkspaceFileBuffer.mock.calls.length).toBeLessThan(files.length)
+  })
+
+  it('rejects a selection whose declared sizes already exceed the limit', async () => {
+    mockListWorkspaceFiles.mockResolvedValue([
+      { ...workspaceFile('f1', 'a.mp4'), size: 200 * MB },
+      { ...workspaceFile('f2', 'b.mp4'), size: 200 * MB },
+    ])
+
+    const response = await GET(requestFor('fileIds=f1&fileIds=f2'), context)
+
+    expect(response.status).toBe(400)
+    expect(mockDownloadFileStream).not.toHaveBeenCalled()
+  })
+})