From 2dcd5ed073fbddf491608cd2b8d7dba273cd441f Mon Sep 17 00:00:00 2001 From: Olusegun Kehinde Date: Mon, 27 Jul 2026 05:19:29 +0100 Subject: [PATCH 1/7] Fix toPublicVault to use DB column names Adjust toPublicVault to accept actual Knex/DB vault row shape and map real column names. Added a VaultDbRow interface, removed the old Vault import, and mapped creator -> creator and end_date -> endTimestamp (instead of creator_address and deadline). Also updated TODO.md to note the fix and checklist steps. --- TODO.md | 24 +++++++----------------- src/utils/mappers.ts | 28 ++++++++++++++++++++++------ 2 files changed, 29 insertions(+), 23 deletions(-) diff --git a/TODO.md b/TODO.md index 71e7937a..9555b0b3 100644 --- a/TODO.md +++ b/TODO.md @@ -1,20 +1,10 @@ -# TODO - -## Plan confirmation (pre-edit) -- Implement admin suspend + reinstate endpoints (POST /api/admin/verifiers/:userId/suspend and .../reinstate) -- Wire suspend/reinstate to verifier status transitions (suspended <-> pending/approved as "prior active") -- Ensure suspended verifiers are rejected by validateMilestoneMultiVerifier() during multi-verifier milestone approvals -- Add audit logging for both lifecycle transitions and blocked approval attempts -- Add tests covering: - - suspend sets deactivated/suspended status correctly and writes audit log - - reinstate restores prior state - - suspended verifier cannot approve milestones; historical votes remain +# Fix: toPublicVault throws on real DB vault rows ## Steps -1. Inspect current milestone multi-approval flow and where validateMilestoneMultiVerifier is used. -2. Update services/milestones.ts to block suspended/deactivated verifiers (query verifier status). -3. Update adminVerifiers routes to add reinstate path and correct lifecycle transitions (and audit). -4. Implement service helpers in services/verifiers.ts for reinstate "prior active state". -5. Add/modify tests in tests/ for lifecycle + multi-verifier approval blocking. -6. Run test suite. + +- [x] Step 0: Analyze issue and gather context (completed) +- [x] Step 1: Confirm plan with user (completed) +- [ ] Step 2: Update `src/utils/mappers.ts` — fix `toPublicVault` to read DB column names (`creator`, `end_date`) +- [ ] Step 3: Update `src/tests/mappers.test.ts` — fix `makeVault` fixture and assertions to match DB row shape +- [ ] Step 4: Run tests to verify the fix diff --git a/src/utils/mappers.ts b/src/utils/mappers.ts index e24d968b..339ab25f 100644 --- a/src/utils/mappers.ts +++ b/src/utils/mappers.ts @@ -1,5 +1,5 @@ import { Milestone } from '../types/horizonSync.js'; -import { Vault, VaultStatus as InternalVaultStatus } from '../types/vault.js'; +import { VaultStatus as InternalVaultStatus } from '../types/vault.js'; import { EnterpriseVault, EnterpriseMilestone, VaultStatus as PublicVaultStatus } from '../types/enterprise.js'; const STATUS_MAP: Record = { @@ -11,17 +11,33 @@ const STATUS_MAP: Record = { }; /** - * Maps an internal Vault model to a public EnterpriseVault DTO. - * Explicitly omits internal fields like 'created_at'. + * Shape of a vault row as returned by Knex from the `vaults` table. + * Uses the real database column names (snake_case). */ -export function toPublicVault(vault: Vault): EnterpriseVault { +interface VaultDbRow { + id: string; + creator: string; + amount: string; + status: InternalVaultStatus; + created_at: Date; + end_date: Date; + success_destination: string; + failure_destination: string; + organization_id?: string; +} + +/** + * Maps a database vault row (from Knex) to a public EnterpriseVault DTO. + * Explicitly omits internal fields like 'created_at' and 'organization_id'. + */ +export function toPublicVault(vault: VaultDbRow): EnterpriseVault { return { id: vault.id, - creator: vault.creator_address, + creator: vault.creator, amount: vault.amount, status: STATUS_MAP[vault.status], startTimestamp: vault.created_at.toISOString(), - endTimestamp: vault.deadline.toISOString(), + endTimestamp: vault.end_date.toISOString(), successDestination: vault.success_destination, failureDestination: vault.failure_destination, }; From d1315c8064dcdd779e8278a1ac6fd2badd949913 Mon Sep 17 00:00:00 2001 From: Olusegun Kehinde Date: Mon, 27 Jul 2026 05:39:11 +0100 Subject: [PATCH 2/7] fix(tests): update vault tests to align with DB schema changes --- TODO.md | 12 ++++++-- src/tests/enterpriseExposure.test.ts | 36 ++++++++++++++-------- src/tests/mappers.test.ts | 46 ++++++++++++++++------------ 3 files changed, 59 insertions(+), 35 deletions(-) diff --git a/TODO.md b/TODO.md index 9555b0b3..12fb4736 100644 --- a/TODO.md +++ b/TODO.md @@ -4,7 +4,13 @@ - [x] Step 0: Analyze issue and gather context (completed) - [x] Step 1: Confirm plan with user (completed) -- [ ] Step 2: Update `src/utils/mappers.ts` — fix `toPublicVault` to read DB column names (`creator`, `end_date`) -- [ ] Step 3: Update `src/tests/mappers.test.ts` — fix `makeVault` fixture and assertions to match DB row shape -- [ ] Step 4: Run tests to verify the fix +- [x] Step 2: Update `src/utils/mappers.ts` — fix `toPublicVault` to read DB column names (`creator`, `end_date`) +- [x] Step 3: Update `src/tests/mappers.test.ts` — fix `makeVault` fixture and assertions to match DB row shape +- [x] Step 4: Update `src/tests/enterpriseExposure.test.ts` — fix mock vault to use DB row shape +- [ ] Step 5: Run tests to verify the fix (blocked — npm install in progress) +**Note:** `npm install` is still running in the terminal. Once it completes, run: +``` +cd c:/Users/dj/Documents/Revora-Contract/Disciplr-backend; npm test +``` +to verify the changes. diff --git a/src/tests/enterpriseExposure.test.ts b/src/tests/enterpriseExposure.test.ts index b26c9705..58956086 100644 --- a/src/tests/enterpriseExposure.test.ts +++ b/src/tests/enterpriseExposure.test.ts @@ -1,22 +1,34 @@ import { toPublicVault, toPublicMilestone } from '../utils/mappers.js'; import { maskPii } from '../utils/privacy.js'; import { Milestone } from '../types/horizonSync.js'; -import { Vault, VaultStatus } from '../types/vault.js'; +import { VaultStatus } from '../types/vault.js'; + +/** + * Shape of a vault row as returned by Knex from the `vaults` table. + * Matches the VaultDbRow interface in mappers.ts. + */ +interface VaultDbRow { + id: string; + creator: string; + amount: string; + status: VaultStatus; + created_at: Date; + end_date: Date; + success_destination: string; + failure_destination: string; + organization_id?: string; +} describe('Enterprise API Exposure Audit', () => { - const mockInternalVault: Vault = { + const mockInternalVault: VaultDbRow = { id: 'vault_123', - contract_id: 'C123', - creator_address: 'GA7QYNF7SOWQ3GLR2BGMZEHXAVIRZA4KVWLTJJFC7MGXUA74P7UJVSGZ', + creator: 'GA7QYNF7SOWQ3GLR2BGMZEHXAVIRZA4KVWLTJJFC7MGXUA74P7UJVSGZ', amount: '1000.0000000', - deadline: new Date('2024-12-31T23:59:59Z'), - milestone_hash: 'hash', - verifier_address: 'GVERIFIER', + end_date: new Date('2024-12-31T23:59:59Z'), success_destination: 'GBBM6BKZPEHWYO3E3YKREDPQXMS4VK35YLNU7NFBRI26RAN7GI5POFBB', failure_destination: 'GDTNXRLOJD2YEBPKK7KCMR7J33AAG5VZXHAJTHIG736D6LVEFLLLKPDL', status: VaultStatus.ACTIVE, created_at: new Date('2024-01-01T00:00:00Z'), - updated_at: new Date() }; test('toPublicVault should omit internal fields', () => { @@ -24,13 +36,13 @@ describe('Enterprise API Exposure Audit', () => { // Verify expected fields are present expect(result.id).toBe(mockInternalVault.id); - expect(result.creator).toBe(mockInternalVault.creator_address); + expect(result.creator).toBe(mockInternalVault.creator); expect(result.amount).toBe(mockInternalVault.amount); // Verify internal fields are strictly omitted expect(result).not.toHaveProperty('created_at'); - expect(result).not.toHaveProperty('updated_at'); - + expect(result).not.toHaveProperty('end_date'); + // Verify date format conversion expect(typeof result.startTimestamp).toBe('string'); expect(result.endTimestamp).toBe('2024-12-31T23:59:59.000Z'); @@ -97,4 +109,4 @@ describe('Enterprise API Exposure Audit', () => { expect(result).not.toHaveProperty('internal_notes'); expect(result.status).toBe('pending'); }); -}); \ No newline at end of file +}); diff --git a/src/tests/mappers.test.ts b/src/tests/mappers.test.ts index aa3ec4ba..3eead38c 100644 --- a/src/tests/mappers.test.ts +++ b/src/tests/mappers.test.ts @@ -1,25 +1,36 @@ import { describe, expect, it } from '@jest/globals' import { toPublicVault, toPublicMilestone } from '../utils/mappers.js' import { VaultStatus } from '../types/vault.js' -import type { Vault } from '../types/vault.js' import type { Milestone } from '../types/horizonSync.js' // ── Fixtures ────────────────────────────────────────────────────────────── -function makeVault(overrides: Partial = {}): Vault { +/** + * Shape of a vault row as returned by Knex from the `vaults` table. + * Matches the VaultDbRow interface in mappers.ts. + */ +interface VaultDbRow { + id: string; + creator: string; + amount: string; + status: VaultStatus; + created_at: Date; + end_date: Date; + success_destination: string; + failure_destination: string; + organization_id?: string; +} + +function makeVault(overrides: Partial = {}): VaultDbRow { return { id: 'vault-001', - contract_id: 'CONTRACT-ABC', - creator_address: 'GCREATOR123', + creator: 'GCREATOR123', amount: '1000', - milestone_hash: 'hash-abc', - verifier_address: 'GVERIFIER456', - success_destination: 'GSUCCESS789', - failure_destination: 'GFAILURE000', status: VaultStatus.ACTIVE, - deadline: new Date('2026-12-31T23:59:59.000Z'), created_at: new Date('2026-01-01T00:00:00.000Z'), - updated_at: new Date('2026-01-01T00:00:00.000Z'), + end_date: new Date('2026-12-31T23:59:59.000Z'), + success_destination: 'GSUCCESS789', + failure_destination: 'GFAILURE000', ...overrides, } } @@ -61,8 +72,8 @@ describe('toPublicVault', () => { expect(dto.startTimestamp).toBe('2026-01-01T00:00:00.000Z') }) - it('converts deadline to ISO 8601 UTC string for endTimestamp', () => { - const vault = makeVault({ deadline: new Date('2026-12-31T23:59:59.000Z') }) + it('converts end_date to ISO 8601 UTC string for endTimestamp', () => { + const vault = makeVault({ end_date: new Date('2026-12-31T23:59:59.000Z') }) const dto = toPublicVault(vault) expect(dto.endTimestamp).toBe('2026-12-31T23:59:59.000Z') }) @@ -76,13 +87,8 @@ describe('toPublicVault', () => { it('does not leak internal fields into the DTO', () => { const dto = toPublicVault(makeVault()) as Record expect(dto).not.toHaveProperty('created_at') - expect(dto).not.toHaveProperty('updated_at') - expect(dto).not.toHaveProperty('contract_id') - expect(dto).not.toHaveProperty('milestone_hash') - expect(dto).not.toHaveProperty('verifier_address') + expect(dto).not.toHaveProperty('end_date') expect(dto).not.toHaveProperty('organization_id') - expect(dto).not.toHaveProperty('creator_address') - expect(dto).not.toHaveProperty('deadline') }) it('preserves exact amount string without coercion', () => { @@ -120,7 +126,7 @@ describe('toPublicVault', () => { const vault = makeVault() const dto = toPublicVault(vault) expect(dto.id).toBe(vault.id) - expect(dto.creator).toBe(vault.creator_address) + expect(dto.creator).toBe(vault.creator) expect(dto.amount).toBe(vault.amount) expect(dto.successDestination).toBe(vault.success_destination) expect(dto.failureDestination).toBe(vault.failure_destination) @@ -218,4 +224,4 @@ describe('toPublicMilestone', () => { expect(dto.targetAmount).toBe('12345.67') expect(dto.currentAmount).toBe('0.01') }) -}) \ No newline at end of file +}) From 962cdf8e1209ba5004e6a0fa29f1feed566caa16 Mon Sep 17 00:00:00 2001 From: Olusegun Kehinde Date: Mon, 31 Aug 2026 19:58:11 +0100 Subject: [PATCH 3/7] feat(exports): add bounded streaming and concurrency primitives --- src/services/exportBounds.ts | 73 ++++++++++++++++++++++++++++++++++++ 1 file changed, 73 insertions(+) create mode 100644 src/services/exportBounds.ts diff --git a/src/services/exportBounds.ts b/src/services/exportBounds.ts new file mode 100644 index 00000000..c0bff153 --- /dev/null +++ b/src/services/exportBounds.ts @@ -0,0 +1,73 @@ +import { Readable } from 'node:stream' + +export const EXPORT_BOUNDS = { + MAX_COLUMN_FILTER_BYTES: 16 * 1024, + MAX_COLUMNS_PER_SECTION: 25, + DOWNLOAD_CHUNK_BYTES: 512 * 1024, + MAX_CONCURRENT_REQUESTS_PER_ORG: 2, + CONCURRENCY_RETRY_AFTER_SECONDS: 1, +} as const + +/** + * Process-local admission gate for export creation requests. + * + * The daily quota remains the authoritative cross-process limit. This gate is + * deliberately smaller and short-lived: it protects the HTTP boundary from a + * burst of simultaneous enqueue work without pretending to be a distributed + * concurrency controller. + */ +export class ExportRequestGate { + private readonly activeRequests = new Map() + + tryAcquire(key: string, limit = EXPORT_BOUNDS.MAX_CONCURRENT_REQUESTS_PER_ORG): boolean { + const active = this.activeRequests.get(key) ?? 0 + if (active >= limit) return false + this.activeRequests.set(key, active + 1) + return true + } + + release(key: string): void { + const active = this.activeRequests.get(key) ?? 0 + if (active <= 1) { + this.activeRequests.delete(key) + return + } + this.activeRequests.set(key, active - 1) + } + + active(key: string): number { + return this.activeRequests.get(key) ?? 0 + } + + reset(): void { + this.activeRequests.clear() + } +} + +export const exportRequestGate = new ExportRequestGate() + +/** + * Stream an already-materialized export without handing the whole Buffer to + * Express in one write. Readable handles backpressure and the generator keeps + * only one bounded chunk live at a time. + */ +export const streamExportBuffer = ( + res: NodeJS.WritableStream, + buffer: Buffer, + chunkSize = EXPORT_BOUNDS.DOWNLOAD_CHUNK_BYTES, +): void => { + if (!Number.isInteger(chunkSize) || chunkSize <= 0) { + throw new RangeError('chunkSize must be a positive integer') + } + + function* chunks(): Generator { + for (let offset = 0; offset < buffer.length; offset += chunkSize) { + yield buffer.subarray(offset, Math.min(offset + chunkSize, buffer.length)) + } + } + + Readable.from(chunks()).pipe(res) +} + +export const isWithinByteLimit = (value: string, maxBytes: number): boolean => + Buffer.byteLength(value, 'utf8') <= maxBytes From afa684b681f156fcb6511112edfe938d53d8d3fc Mon Sep 17 00:00:00 2001 From: Olusegun Kehinde Date: Mon, 31 Aug 2026 19:58:18 +0100 Subject: [PATCH 4/7] test(exports): cover request and streaming bounds --- src/services/exportBounds.test.ts | 83 +++++++++++++++++++++++++++++++ 1 file changed, 83 insertions(+) create mode 100644 src/services/exportBounds.test.ts diff --git a/src/services/exportBounds.test.ts b/src/services/exportBounds.test.ts new file mode 100644 index 00000000..2fdf04ce --- /dev/null +++ b/src/services/exportBounds.test.ts @@ -0,0 +1,83 @@ +import { describe, expect, it, beforeEach } from '@jest/globals' +import { Writable } from 'node:stream' +import { + EXPORT_BOUNDS, + ExportRequestGate, + isWithinByteLimit, + streamExportBuffer, +} from './exportBounds.js' + +describe('ExportRequestGate', () => { + let gate: ExportRequestGate + + beforeEach(() => { + gate = new ExportRequestGate() + }) + + it('admits requests up to the configured concurrency bound', () => { + expect(gate.tryAcquire('org-a')).toBe(true) + expect(gate.tryAcquire('org-a')).toBe(true) + expect(gate.tryAcquire('org-a')).toBe(false) + expect(gate.active('org-a')).toBe(EXPORT_BOUNDS.MAX_CONCURRENT_REQUESTS_PER_ORG) + }) + + it('isolates concurrency between organizations', () => { + expect(gate.tryAcquire('org-a')).toBe(true) + expect(gate.tryAcquire('org-a')).toBe(true) + expect(gate.tryAcquire('org-b')).toBe(true) + expect(gate.active('org-a')).toBe(2) + expect(gate.active('org-b')).toBe(1) + }) + + it('release is idempotent at zero and allows recovery', () => { + gate.release('missing') + expect(gate.active('org-a')).toBe(0) + + expect(gate.tryAcquire('org-a')).toBe(true) + expect(gate.tryAcquire('org-a')).toBe(true) + gate.release('org-a') + expect(gate.active('org-a')).toBe(1) + expect(gate.tryAcquire('org-a')).toBe(true) + }) + + it('supports explicit limits for adversarial boundary tests', () => { + expect(gate.tryAcquire('org-a', 0)).toBe(false) + expect(gate.tryAcquire('org-a', 1)).toBe(true) + expect(gate.tryAcquire('org-a', 1)).toBe(false) + }) +}) + +describe('export input byte bounds', () => { + it('counts UTF-8 bytes rather than JavaScript code units', () => { + expect(isWithinByteLimit('a'.repeat(16), 16)).toBe(true) + expect(isWithinByteLimit('€'.repeat(6), 16)).toBe(false) + }) +}) + +describe('streamExportBuffer', () => { + it('writes the complete payload in bounded chunks', async () => { + const payload = Buffer.from('abcdefghij') + const chunks: Buffer[] = [] + const sink = new Writable({ + write(chunk, _encoding, callback) { + chunks.push(Buffer.from(chunk)) + callback() + }, + }) + + streamExportBuffer(sink, payload, 3) + await new Promise((resolve, reject) => { + sink.once('finish', resolve) + sink.once('error', reject) + }) + + expect(chunks.map(chunk => chunk.length)).toEqual([3, 3, 3, 1]) + expect(Buffer.concat(chunks)).toEqual(payload) + }) + + it('rejects invalid chunk sizes', () => { + const sink = new Writable({ write(_chunk, _encoding, callback) { callback() } }) + expect(() => streamExportBuffer(sink, Buffer.from('x'), 0)).toThrow(RangeError) + expect(() => streamExportBuffer(sink, Buffer.from('x'), 1.5)).toThrow(RangeError) + }) +}) From d5a38ddffb42d411f731b49c4e3a6e9bf4294fde Mon Sep 17 00:00:00 2001 From: Olusegun Kehinde Date: Mon, 31 Aug 2026 19:59:00 +0100 Subject: [PATCH 5/7] fix(exports): support lightweight response fallback --- src/services/exportBounds.ts | 23 +++++++++-------------- 1 file changed, 9 insertions(+), 14 deletions(-) diff --git a/src/services/exportBounds.ts b/src/services/exportBounds.ts index c0bff153..23c3097f 100644 --- a/src/services/exportBounds.ts +++ b/src/services/exportBounds.ts @@ -8,14 +8,6 @@ export const EXPORT_BOUNDS = { CONCURRENCY_RETRY_AFTER_SECONDS: 1, } as const -/** - * Process-local admission gate for export creation requests. - * - * The daily quota remains the authoritative cross-process limit. This gate is - * deliberately smaller and short-lived: it protects the HTTP boundary from a - * burst of simultaneous enqueue work without pretending to be a distributed - * concurrency controller. - */ export class ExportRequestGate { private readonly activeRequests = new Map() @@ -46,13 +38,8 @@ export class ExportRequestGate { export const exportRequestGate = new ExportRequestGate() -/** - * Stream an already-materialized export without handing the whole Buffer to - * Express in one write. Readable handles backpressure and the generator keeps - * only one bounded chunk live at a time. - */ export const streamExportBuffer = ( - res: NodeJS.WritableStream, + res: NodeJS.WritableStream & { send?: (body: Buffer) => unknown }, buffer: Buffer, chunkSize = EXPORT_BOUNDS.DOWNLOAD_CHUNK_BYTES, ): void => { @@ -60,6 +47,14 @@ export const streamExportBuffer = ( throw new RangeError('chunkSize must be a positive integer') } + if (typeof res.write !== 'function') { + if (typeof res.send === 'function') { + res.send(buffer) + return + } + throw new TypeError('response must support write() or send()') + } + function* chunks(): Generator { for (let offset = 0; offset < buffer.length; offset += chunkSize) { yield buffer.subarray(offset, Math.min(offset + chunkSize, buffer.length)) From d271c94ba4c74d1ebd41aaac43c0aad0d86cbac5 Mon Sep 17 00:00:00 2001 From: Olusegun Kehinde Date: Mon, 31 Aug 2026 19:59:14 +0100 Subject: [PATCH 6/7] feat(exports): bound requests, enforce quota identity, stream downloads --- src/routes/exports.ts | 226 ++++++++++++++++++++++++++---------------- 1 file changed, 140 insertions(+), 86 deletions(-) diff --git a/src/routes/exports.ts b/src/routes/exports.ts index 57fac47c..38f5ae11 100644 --- a/src/routes/exports.ts +++ b/src/routes/exports.ts @@ -3,7 +3,6 @@ import type { BackgroundJobSystem } from '../jobs/system.js' import { authenticate, requireAdmin, - signDownloadToken, verifyDownloadToken, type AuthenticatedRequest, } from '../middleware/auth.js' @@ -22,9 +21,27 @@ import { getEnv } from '../config/index.js' import { resolveS3Config, getExportSignedUrl } from '../services/exportS3.js' import { createAuditLog } from '../lib/audit-logs.js' import { isOrgMember } from '../models/organizations.js' - -const resolveOrgId = (req: AuthenticatedRequest): string => - (req as any).orgId as string | undefined ?? (req.query.orgId as string | undefined) ?? (req.headers['x-organization-id'] as string | undefined) ?? (req.user as any)?.orgId ?? req.user!.userId +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): void => { + console.info(JSON.stringify({ + level: 'info', + event, + ...payload, + timestamp: new Date().toISOString(), + })) +} const enforceExportQuota = async ( req: AuthenticatedRequest, @@ -38,6 +55,7 @@ const enforceExportQuota = async ( error: 'Export quota exceeded. Try again tomorrow.', retryAfter: result.retryAfter, }) + logExportEvent('exports.quota_rejected', { orgId, retryAfter: result.retryAfter }) return false } return true @@ -74,9 +92,7 @@ export function createExportRouter(jobSystem: BackgroundJobSystem): Router { const columnsParam = req.query.columns as string | undefined const validScopes = ['vaults', 'transactions', 'analytics', 'all'] - if (!validScopes.includes(scope)) { - return null - } + if (!validScopes.includes(scope)) return null const result: ParseOptionsResult = { format, @@ -84,21 +100,17 @@ export function createExportRouter(jobSystem: BackgroundJobSystem): Router { } if (columnsParam) { + if (!isWithinByteLimit(columnsParam, EXPORT_BOUNDS.MAX_COLUMN_FILTER_BYTES)) return null + try { - const parsedColumns: Record = typeof columnsParam === 'string' - ? JSON.parse(columnsParam) - : columnsParam - result.columns = {} + const parsedColumns: unknown = JSON.parse(columnsParam) + if (!parsedColumns || typeof parsedColumns !== 'object' || Array.isArray(parsedColumns)) return null - for (const [section, cols] of Object.entries(parsedColumns)) { + result.columns = {} + for (const [section, cols] of Object.entries(parsedColumns as Record)) { const allowed = ALLOWED_COLUMNS[section as keyof typeof ALLOWED_COLUMNS] - if (!allowed) { - return null - } - - if (!Array.isArray(cols) || !cols.every(col => allowed.includes(col))) { - return null - } + 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 result.columns[section as keyof typeof ALLOWED_COLUMNS] = cols } } catch { @@ -113,8 +125,26 @@ export function createExportRouter(jobSystem: BackgroundJobSystem): Router { 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) { @@ -122,28 +152,47 @@ export function createExportRouter(jobSystem: BackgroundJobSystem): Router { return } - if (!await enforceExportQuota(req, res)) return + const orgId = acquireExportRequest(req, res) + if (!orgId) return + const startedAt = Date.now() try { - const job = await enqueueExportJob(jobSystem, { - userId: req.user!.userId, - orgId: resolveOrgId(req), - isAdmin: false, - scope: options.scope, - format: options.format, - columns: options.columns as any, - idempotencyKey: req.header('idempotency-key') ?? undefined, - }) + if (!await enforceExportQuota(req, res)) return - 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) } }) @@ -154,32 +203,49 @@ export function createExportRouter(jobSystem: BackgroundJobSystem): Router { return } - if (!await enforceExportQuota(req, res)) return - - const targetUserId = - typeof req.query.targetUserId === 'string' ? req.query.targetUserId : undefined + const targetUserId = typeof req.query.targetUserId === 'string' ? req.query.targetUserId : undefined + const orgId = acquireExportRequest(req, res) + if (!orgId) return + const startedAt = Date.now() try { - const job = await enqueueExportJob(jobSystem, { - userId: req.user!.userId, - orgId: resolveOrgId(req), - isAdmin: true, - targetUserId, - scope: options.scope, - format: options.format, - columns: options.columns as any, - idempotencyKey: req.header('idempotency-key') ?? undefined, - }) + if (!await enforceExportQuota(req, res)) return - 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) } }) @@ -232,25 +298,19 @@ export function createExportRouter(jobSystem: BackgroundJobSystem): Router { const mimeType = job.format === 'csv' ? 'text/csv; charset=utf-8' + : job.format === 'ndjson' + ? 'application/x-ndjson; charset=utf-8' : '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) => { @@ -287,16 +347,10 @@ export function createExportRouter(jobSystem: BackgroundJobSystem): Router { 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) { @@ -322,9 +376,9 @@ export function createExportRouter(jobSystem: BackgroundJobSystem): Router { : '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 From 35f3850e23c4fe62685af522d7ed81f88fbe0119 Mon Sep 17 00:00:00 2001 From: Olusegun Kehinde Date: Mon, 31 Aug 2026 19:59:23 +0100 Subject: [PATCH 7/7] docs(exports): document bounded export contract --- docs/EXPORT_STREAMING_AND_QUOTAS.md | 38 +++++++++++++++++++++++++++++ 1 file changed, 38 insertions(+) create mode 100644 docs/EXPORT_STREAMING_AND_QUOTAS.md diff --git a/docs/EXPORT_STREAMING_AND_QUOTAS.md b/docs/EXPORT_STREAMING_AND_QUOTAS.md new file mode 100644 index 00000000..0db41aea --- /dev/null +++ b/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 +```