Feat/1537 export bounds quota

docs/EXPORT_STREAMING_AND_QUOTAS.md

@@ -0,0 +1,38 @@
+# Export streaming and quota contract
+
+Issue: #1537
+
+## Invariants
+
+- Export quota accounting is keyed only by the authenticated principal. Client-supplied `orgId` query parameters and `x-organization-id` headers are not trusted.
+- The daily quota repository performs the check and increment atomically, so concurrent requests cannot push a stored counter above its configured limit.
+- Export column filters are bounded to 16 KiB of UTF-8 input and at most 25 columns per section.
+- Export creation admits at most two simultaneous HTTP enqueue operations per quota key in a process. The daily quota remains the authoritative cross-process control.
+- Download responses are emitted in 512 KiB chunks through a backpressure-aware stream when the response supports `write()`/`end()`.
+- Export polling metadata exposes a one-second minimum interval and a 300-attempt client-side ceiling (five minutes at the advertised interval).
+- Operational events contain bounded error text and export metadata, not credentials or signed URLs.
+
+## Degraded behavior
+
+- Daily quota exhaustion returns `429` with `Retry-After` set to the remaining UTC-day duration.
+- Process-local concurrency saturation returns `429` with a one-second `Retry-After` hint.
+- Invalid or oversized column filters return `400` before quota consumption or job enqueue.
+- Unauthorized export status/download access returns `403` without revealing export contents.
+
+## Tradeoffs and limitations
+
+The concurrency gate is intentionally process-local. It protects the HTTP enqueue boundary from bursts but is not a replacement for a distributed semaphore. Deployments with multiple application instances must rely on the atomic persistent quota and job-system worker capacity for cross-process enforcement.
+
+Export generation still materializes the completed result in the existing job representation. This change bounds request parsing and network delivery and avoids one-shot response writes; it does not redesign the storage/worker representation of generated results.
+
+## Validation
+
+Focused validation:
+
+```bash
+npm test -- src/services/exportBounds.test.ts
+npm test -- src/routes/exports.quota.test.ts
+npm test -- src/routes/exports.test.ts
+npm run build
+npm run lint
+```

src/routes/exports.ts

@@ -4,7 +4,6 @@
 import {
   authenticate,
   requireAdmin,
-  signDownloadToken,
   verifyDownloadToken,
   type AuthenticatedRequest,
 } from '../middleware/auth.js'
@@ -23,6 +22,27 @@
 import { resolveS3Config, getExportSignedUrl } from '../services/exportS3.js'
 import { createAuditLog } from '../lib/audit-logs.js'
 import { isOrgMember } from '../models/organizations.js'
+import {
+  EXPORT_BOUNDS,
+  exportRequestGate,
+  isWithinByteLimit,
+  streamExportBuffer,
+} from '../services/exportBounds.js'
+
+/**
+ * The authenticated principal is the only trusted quota/access-control key.
+ * Client-supplied orgId query/header values are deliberately ignored.
+ */
+const resolveOrgId = (req: AuthenticatedRequest): string => req.user!.userId
+
+const logExportEvent = (event: string, payload: Record<string, unknown>): void => {
+  console.info(JSON.stringify({
+    level: 'info',
+    event,
+    ...payload,
+    timestamp: new Date().toISOString(),
+  }))
+}
 
 /**
  * Derive the org identifier used for quota and access-control decisions.
@@ -51,6 +71,7 @@
       error: 'Export quota exceeded. Try again tomorrow.',
       retryAfter: result.retryAfter,
     })
+    logExportEvent('exports.quota_rejected', { orgId, retryAfter: result.retryAfter })
     return false
   }
   return true
@@ -91,6 +112,8 @@
     const format = negotiateFormat(req, formatStr)
     const scopeStr = typeof req.query.scope === 'string' ? req.query.scope : 'all'
 
+    const validScopes = ['vaults', 'transactions', 'analytics', 'all']
+    if (!validScopes.includes(scope)) return null
     const ScopeSchema = z.enum(['vaults', 'transactions', 'analytics', 'all'])
     const scopeParse = ScopeSchema.safeParse(scopeStr)
     if (!scopeParse.success) {
@@ -105,7 +128,17 @@
 
     const columnsParam = req.query.columns
     if (columnsParam) {
+      if (!isWithinByteLimit(columnsParam, EXPORT_BOUNDS.MAX_COLUMN_FILTER_BYTES)) return null
+
       try {
+        const parsedColumns: unknown = JSON.parse(columnsParam)
+        if (!parsedColumns || typeof parsedColumns !== 'object' || Array.isArray(parsedColumns)) return null
+
+        result.columns = {}
+        for (const [section, cols] of Object.entries(parsedColumns as Record<string, unknown>)) {
+          const allowed = ALLOWED_COLUMNS[section as keyof typeof ALLOWED_COLUMNS]
+          if (!allowed || !Array.isArray(cols) || cols.length > EXPORT_BOUNDS.MAX_COLUMNS_PER_SECTION) return null
+          if (!cols.every(col => typeof col === 'string' && allowed.includes(col))) return null
         const parsedColumns: unknown = typeof columnsParam === 'string'
           ? JSON.parse(columnsParam)
           : columnsParam
@@ -129,27 +162,51 @@
           }
           result.columns[section as keyof typeof ALLOWED_COLUMNS] = cols
         }
       } catch {
         return null
       }
     }
 
     return result
   }
 
   const buildAcceptedResponse = (jobId: string) => ({
     jobId,
     statusUrl: `/api/exports/status/${jobId}`,
     pollIntervalMs: 1000,
+    maxPollAttempts: 300,
   })
 
+  const acquireExportRequest = (req: AuthenticatedRequest, res: Response): string | null => {
+    const orgId = resolveOrgId(req)
+    if (exportRequestGate.tryAcquire(orgId)) return orgId
+
+    res.setHeader('Retry-After', String(EXPORT_BOUNDS.CONCURRENCY_RETRY_AFTER_SECONDS))
+    res.status(429).json({
+      error: 'Too many export requests in progress. Retry shortly.',
+      retryAfter: EXPORT_BOUNDS.CONCURRENCY_RETRY_AFTER_SECONDS,
+    })
+    logExportEvent('exports.concurrency_rejected', {
+      orgId,
+      activeRequests: exportRequestGate.active(orgId),
+      limit: EXPORT_BOUNDS.MAX_CONCURRENT_REQUESTS_PER_ORG,
+    })
+    return null
+  }
+
   router.post('/me', authenticate, requireScopes(ApiScope.ReadAnalytics, ApiScope.ReadVaults), async (req: AuthenticatedRequest, res: Response) => {
     const options = parseOptions(req)
     if (!options) {
       res.status(400).json({ error: 'Invalid format, scope, or columns parameter' })
       return
     }
 
+    const orgId = acquireExportRequest(req, res)
+    if (!orgId) return
+    const startedAt = Date.now()
+
+    try {
+      if (!await enforceExportQuota(req, res)) return
     const idemParse = IdempotencyKeySchema.safeParse(req.header('idempotency-key'))
     if (!idemParse.success) {
       res.status(400).json({ error: 'Invalid idempotency-key header' })
@@ -169,17 +226,42 @@
         idempotencyKey: idemParse.data,
       })
 
-      res.status(202).json(buildAcceptedResponse(job.id))
-    } catch (error) {
-      if (isExportIdempotencyConflictError(error)) {
-        res.status(409).json({ error: error.message })
-        return
+      try {
+        const job = await enqueueExportJob(jobSystem, {
+          userId: req.user!.userId,
+          orgId,
+          isAdmin: false,
+          scope: options.scope,
+          format: options.format,
+          columns: options.columns as any,
+          idempotencyKey: req.header('idempotency-key') ?? undefined,
+        })
+
+        logExportEvent('exports.enqueue_accepted', {
+          orgId,
+          jobId: job.id,
+          format: options.format,
+          scope: options.scope,
+          latencyMs: Date.now() - startedAt,
+        })
+        res.status(202).json(buildAcceptedResponse(job.id))
+      } catch (error) {
+        if (isExportIdempotencyConflictError(error)) {
+          res.status(409).json({ error: error.message })
+          return
+        }
+        const message = error instanceof Error ? error.message : 'Failed to enqueue export job'
+        logExportEvent('exports.enqueue_failed', {
+          orgId,
+          latencyMs: Date.now() - startedAt,
+          error: message.slice(0, 200),
+        })
+        res.status(500).json({ error: message })
       }
-
-      const message = error instanceof Error ? error.message : 'Failed to enqueue export job'
-      res.status(500).json({ error: message })
+    } finally {
+      exportRequestGate.release(orgId)
     }
   })
 
   router.post('/admin', authenticate, requireAdmin, requireScopes(ApiScope.ReadAnalytics, ApiScope.ReadVaults), async (req: AuthenticatedRequest, res: Response) => {
     const options = parseOptions(req)
@@ -188,6 +270,13 @@
       return
     }
 
+    const targetUserId = typeof req.query.targetUserId === 'string' ? req.query.targetUserId : undefined
+    const orgId = acquireExportRequest(req, res)
+    if (!orgId) return
+    const startedAt = Date.now()
+
+    try {
+      if (!await enforceExportQuota(req, res)) return
     const idemParse = IdempotencyKeySchema.safeParse(req.header('idempotency-key'))
     if (!idemParse.success) {
       res.status(400).json({ error: 'Invalid idempotency-key header' })
@@ -216,17 +305,43 @@
         idempotencyKey: idemParse.data,
       })
 
-      res.status(202).json(buildAcceptedResponse(job.id))
-    } catch (error) {
-      if (isExportIdempotencyConflictError(error)) {
-        res.status(409).json({ error: error.message })
-        return
+      try {
+        const job = await enqueueExportJob(jobSystem, {
+          userId: req.user!.userId,
+          orgId,
+          isAdmin: true,
+          targetUserId,
+          scope: options.scope,
+          format: options.format,
+          columns: options.columns as any,
+          idempotencyKey: req.header('idempotency-key') ?? undefined,
+        })
+
+        logExportEvent('exports.enqueue_accepted', {
+          orgId,
+          jobId: job.id,
+          format: options.format,
+          scope: options.scope,
+          latencyMs: Date.now() - startedAt,
+        })
+        res.status(202).json(buildAcceptedResponse(job.id))
+      } catch (error) {
+        if (isExportIdempotencyConflictError(error)) {
+          res.status(409).json({ error: error.message })
+          return
+        }
+        const message = error instanceof Error ? error.message : 'Failed to enqueue export job'
+        logExportEvent('exports.enqueue_failed', {
+          orgId,
+          latencyMs: Date.now() - startedAt,
+          error: message.slice(0, 200),
+        })
+        res.status(500).json({ error: message })
       }
-
-      const message = error instanceof Error ? error.message : 'Failed to enqueue export job'
-      res.status(500).json({ error: message })
+    } finally {
+      exportRequestGate.release(orgId)
     }
   })
 
   router.get('/status/:jobId', authenticate, async (req: AuthenticatedRequest, res: Response) => {
     const parseResult = JobIdSchema.safeParse(req.params.jobId)
@@ -296,22 +411,14 @@
       : 'application/json; charset=utf-8'
 
     res.setHeader('Content-Type', mimeType)
-    res.setHeader('Content-Disposition', `attachment; filename="${job.filename}"`)
+    res.setHeader('Content-Disposition', `attachment; filename=\"${job.filename}\"`)
     res.setHeader('Content-Length', job.result.length)
-
-    console.info(
-      JSON.stringify({
-        level: 'info',
-        event: 'exports.download_served',
-        jobId: job.id,
-        format: job.format,
-        bytes: job.result.length,
-        filename: job.filename,
-        timestamp: new Date().toISOString(),
-      }),
-    )
-
-    res.send(job.result)
+    logExportEvent('exports.download_served', {
+      jobId: job.id,
+      format: job.format,
+      bytes: job.result.length,
+    })
+    streamExportBuffer(res, job.result)
   })
 
   router.get('/:id/download', authenticate, async (req: AuthenticatedRequest, res: Response) => {
@@ -355,16 +462,10 @@
       console.warn('Failed to record audit log for export download:', err)
     }
 
-    console.info(
-      JSON.stringify({
-        level: 'info',
-        event: 'exports.download_served',
-        jobId: job.id,
-        principal: req.user!.userId,
-        org: callerOrgId,
-        timestamp: new Date().toISOString(),
-      }),
-    )
+    logExportEvent('exports.download_requested', {
+      jobId: job.id,
+      storage: job.s3Key ? 's3' : 'local',
+    })
 
     const s3Config = resolveS3Config()
     if (s3Config && job.s3Key) {
@@ -390,10 +491,10 @@
       : 'application/x-ndjson'
 
     res.setHeader('Content-Type', mimeType)
-    res.setHeader('Content-Disposition', `attachment; filename="${job.filename ?? 'export'}"`)
+    res.setHeader('Content-Disposition', `attachment; filename=\"${job.filename ?? 'export'}\"`)
     res.setHeader('Content-Length', job.result.length)
-    res.send(job.result)
+    streamExportBuffer(res, job.result)
   })
 
   return router
 }