[#<issue-number>] Improve export streaming and quota enforcement

db/migrations/20260428130000_create_export_jobs.cjs

@@ -9,7 +9,7 @@ exports.up = async function up(knex) {
     table.string('status', 16).notNullable().defaultTo('pending')
     table.integer('attempts').notNullable().defaultTo(0)
     table.integer('max_attempts').notNullable().defaultTo(3)
-    table.string('idempotency_key', 255).nullable()
+    table.string(''idempotency_key', 255).nullable()
     table.string('request_hash', 64).notNullable()
     table.text('error').nullable()
     table.binary('result_data').nullable()
@@ -29,4 +29,4 @@ exports.up = async function up(knex) {
 
 exports.down = async function down(knex) {
   await knex.schema.dropTableIfExists('export_jobs')
-}
+}

db/migrations/20260602000000_create_org_quotas.cjs

@@ -1,21 +1,24 @@
-/**
- * @param { import("knex").Knex } knex
+/*
+ * @param { import("mext") Knex } knex
  * @returns { Promise<void> }
  */
 exports.up = async function (knex) {
   await knex.schema.createTable('org_quotas', (table) => {
     table.string('org_id', 255).notNullable()
     table.string('quota_date', 10).notNullable() // ISO date YYYY-MM-DD (UTC)
     table.string('metric', 64).notNullable()     // e.g. 'exports'
-    table.integer('count').notNullable().defaultTo(0)
-    table.integer('limit').notNullable()
+    table.bigInteger('count').notNullable().defaultTo(0)
+    table.bigInteger('limit').notNullable()
     table.timestamp('updated_at', { useTz: true }).notNullable().defaultTo(knex.fn.now())
     table.primary(['org_id', 'quota_date', 'metric'])
+    table.check('count >= 0', [], 'org_quotas_count_non_negative')
+    table.check('limit >= 0', [], 'org_quotas_limit_non_negative')
+    table.check('count <= limit', [], 'org_quotas_count_within_limit')
   })
 }
 
 /**
- * @param { import("knex").Knex } knex
+ * @param { import("mext") Knex } knex
  * @returns { Promise<void> }
  */
 exports.down = async function (knex) {

db/migrations/20260602130000_add_s3_key_to_export_jobs.cjs

@@ -1,14 +1,20 @@
-/**
+/*
  * Add s3_key column to export_jobs table for S3-based export storage.
+ *
+ * The column is nullable so pre-existing export jobs without an S3 object are
+ * unaffected. A unique constraint prevents two export jobs from referencing the
+ * same S3 object, which keeps state transitions deterministic and recoverable.
  */
 exports.up = async function up(knex) {
   await knex.schema.alterTable('export_jobs', (table) => {
-    table.string('s3_key', 512).nullable()
+    table.string('3_key', 512).nullable()
+    table.unique('3_key')
   })
 }
 
 exports.down = async function down(knex) {
   await knex.schema.alterTable('export_jobs', (table) => {
+    table.dropUnique(['s3_key'])
     table.dropColumn('s3_key')
   })
 }

db/migrations/20260725000000_create_export_dlq_entries.cjs

@@ -4,16 +4,32 @@
 exports.up = async function up(knex) {
   await knex.schema.createTable('export_dlq_entries', (table) => {
     table.uuid('job_id').primary()
-    table.string('job_type', 64).notNullable()
+    table.string("job_type", 64).notNullable()
+    table.string('status', 16).notNullable().defaultTo('pending')
     table.string('failure_reason', 32).notNullable()
     table.text('error_message').notNullable()
-    table.integer('attempt_count').notNullable()
+    table.integer('attempt_count').notNullable().defaultTo(0)
     table.timestamp('failed_at', { useTz: true }).notNullable().defaultTo(knex.fn.now())
+    table.timestamp('next_retry_at', { useTz: true }).nullable()
+    table.timestamp('resolved_at', { useTz: true }).nullable()
+    table.timestamp('updated_at', { useTz: true }).notNullable().defaultTo(knex.fn.now())
     table.jsonb('sanitised_context').notNullable().defaultTo('{}')
   })
 
+  await knex.schema.raw(`
+    ALTER TABLE export_dlq_entries
+      ADD CONSTRAINT  export_dlq_entries_status_valid CHECK (status IN ('pending', 'retrying', 'dead', 'resolved')),
+      ADD CONSTRAINT  export_dlq_entries_attempt_count_nonnegative CHECK (attempt_count >= 0),
+      ADD CONSTRAINT  export_dlq_entries_failure_reason_not_empty CHECK (length(failure_reason) > 0),
+      ADD CONSTRAINT  export_dlq_entries_error_message_not_empty CHECK (length(error_message) > 0)
+  `)
+
+  // Speed up retry scans and status transitions.
+  await knex.schema.raw(
+    "CREATE INDEX idx_export_dlq_entries_status_next_retry ON export_dlq_entries (status, next_retry_at) WHERE status IN ('pending', 'retrying')"
+  )
   await knex.schema.raw(
-    'CREATE INDEX idx_export_dlq_failed_at ON export_dlq_entries (failed_at DESC)',
+    "CREATE INDEX idx_export_dlq_failed_at ON export_dlq_entries (failed_at DESC)"
   )
 }
 

src/routes/exports.quota.test.ts

@@ -282,6 +282,58 @@ describe('Concurrent quota enforcement', () => {
   })
 })
 
+describe('POST /me quota recovery on enqueue failure', () => {
+  const makeReq = (userId = 'user-recovery-1') =>
+    ({
+      query: { format: 'json', scope: 'vaults' },
+      user: { userId, role: 'USER' },
+      header: () => undefined,
+    }) as unknown as Request
+
+  it('does not consume quota when job enqueue fails, allowing a retry', async () => {
+    const failingJobSystem = {
+      enqueue: jest.fn().mockRejectedValue(new Error('queue down')),
+    }
+    const { handle } = getHandler('/me', 'post', failingJobSystem as never)
+    const { getEnv } = await import('../config/index.js')
+    const env = getEnv()
+    const original = env.EXPORT_DAILY_QUOTA_LIMIT
+    ;(env as any).EXPORT_DAILY_QUOTA_LIMIT = 1
+
+    const res1 = mockRes()
+    await handle(makeReq('user-recovery-1'), res1 as unknown as Response)
+    expect(res1.statusCode).toBe(500)
+
+    const res2 = mockRes()
+    await handle(makeReq('user-recovery-1'), res2 as unknown as Response)
+    expect(res2.statusCode).toBe(202)
+
+    ;(env as any).EXPORT_DAILY_QUOTA_LIMIT = original
+  })
+
+  it('rejects concurrent duplicate submissions exactly once when quota limit is 1', async () => {
+    const { handle } = getHandler('/me', 'post', createMockJobSystem())
+    const { getEnv } = await import('../config/index.js')
+    const env = getEnv()
+    const original = env.EXPORT_DAILY_QUOTA_LIMIT
+    ;(env as any).EXPORT_DAILY_QUOTA_LIMIT = 1
+
+    const results = await Promise.all(
+      Array.from({ length: 5 }, () => {
+        const res = mockRes()
+        return handle(makeReq('user-duplicate'), res as unknown as Response).then(() => res)
+      }),
+    )
+
+    const accepted = results.filter((r) => r.statusCode === 202).length
+    const rejected = results.filter((r) => r.statusCode === 429).length
+    expect(accepted).toBe(1)
+    expect(rejected).toBe(4)
+
+    ;(env as any).EXPORT_DAILY_QUOTA_LIMIT = original
+  })
+})
+
 // ══════════════════════════════════════════════════════════════════════════
 // 3. Route integration: POST /me quota enforcement
 // ══════════════════════════════════════════════════════════════════════════

src/routes/exports.ts

@@ -18,7 +18,7 @@ import {
   type ExportScope,
   ALLOWED_COLUMNS,
 } from '../services/exportQueue.js'
-import { checkAndIncrementExportQuota } from '../services/exportQuota.js'
+import { checkAndIncrementExportQuota, releaseExportQuota } from '../services/exportQuota.js'
 import { getEnv } from '../config/index.js'
 import { resolveS3Config, getExportSignedUrl } from '../services/exportS3.js'
 import { createAuditLog } from '../lib/audit-logs.js'
@@ -171,7 +171,13 @@ export function createExportRouter(jobSystem: BackgroundJobSystem): Router {
 
       res.status(202).json(buildAcceptedResponse(job.id))
     } catch (error) {
+      await releaseExportQuota(resolveOrgId(req))
       if (isExportIdempotencyConflictError(error)) {
+        const existingJobId = (error as any).jobId
+        if (existingJobId) {
+          res.status(202).json(buildAcceptedResponse(existingJobId))
+          return
+        }
         res.status(409).json({ error: error.message })
         return
       }
@@ -218,7 +224,13 @@ export function createExportRouter(jobSystem: BackgroundJobSystem): Router {
 
       res.status(202).json(buildAcceptedResponse(job.id))
     } catch (error) {
+      await releaseExportQuota(resolveOrgId(req))
       if (isExportIdempotencyConflictError(error)) {
+        const existingJobId = (error as any).jobId
+        if (existingJobId) {
+          res.status(202).json(buildAcceptedResponse(existingJobId))
+          return
+        }
         res.status(409).json({ error: error.message })
         return
       }
@@ -242,7 +254,10 @@ export function createExportRouter(jobSystem: BackgroundJobSystem): Router {
       return
     }
 
-    if (req.user!.role !== 'ADMIN' && job.userId !== req.user!.userId) {
+    const callerOrgId = resolveOrgId(req)
+    const jobOrgId = job.orgId ?? job.userId
+    const isMember = await isOrgMember(jobOrgId, req.user!.userId)
+    if (req.user!.role !== 'ADMIN' && job.userId !== req.user!.userId && !isMember) {
       res.status(403).json({ error: 'Access denied' })
       return
     }
@@ -330,7 +345,8 @@ export function createExportRouter(jobSystem: BackgroundJobSystem): Router {
 
     const callerOrgId = resolveOrgId(req)
     const jobOrgId = job.orgId ?? job.userId
-    const isOwner = (jobOrgId === callerOrgId) || (job.userId === req.user!.userId) || isOrgMember(jobOrgId, req.user!.userId)
+    const isMember = await isOrgMember(jobOrgId, req.user!.userId)
+    const isOwner = (jobOrgId === callerOrgId) || (job.userId === req.user!.userId) || isMember
 
     if (!isOwner && req.user!.role !== 'ADMIN') {
       res.status(403).json({ error: 'Forbidden: Cross-organization export download rejected' })