diff --git a/db/migrations/20260428130000_create_export_jobs.cjs b/db/migrations/20260428130000_create_export_jobs.cjs index e3065b3b..58fba5ff 100644 --- a/db/migrations/20260428130000_create_export_jobs.cjs +++ b/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') -} +} \ No newline at end of file diff --git a/db/migrations/20260602000000_create_org_quotas.cjs b/db/migrations/20260602000000_create_org_quotas.cjs index a42acbe9..b4025273 100644 --- a/db/migrations/20260602000000_create_org_quotas.cjs +++ b/db/migrations/20260602000000_create_org_quotas.cjs @@ -1,5 +1,5 @@ -/** - * @param { import("knex").Knex } knex +/* + * @param { import("mext") Knex } knex * @returns { Promise } */ exports.up = async function (knex) { @@ -7,15 +7,18 @@ exports.up = async function (knex) { 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 } */ exports.down = async function (knex) { diff --git a/db/migrations/20260602130000_add_s3_key_to_export_jobs.cjs b/db/migrations/20260602130000_add_s3_key_to_export_jobs.cjs index 69db40ee..c12b4710 100644 --- a/db/migrations/20260602130000_add_s3_key_to_export_jobs.cjs +++ b/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') }) } diff --git a/db/migrations/20260725000000_create_export_dlq_entries.cjs b/db/migrations/20260725000000_create_export_dlq_entries.cjs index cfc4783d..cc0be86a 100644 --- a/db/migrations/20260725000000_create_export_dlq_entries.cjs +++ b/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)" ) } diff --git a/src/routes/exports.quota.test.ts b/src/routes/exports.quota.test.ts index 45cb8a09..758a4953 100644 --- a/src/routes/exports.quota.test.ts +++ b/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 // ══════════════════════════════════════════════════════════════════════════ diff --git a/src/routes/exports.ts b/src/routes/exports.ts index 8426e1c2..f1dd5e71 100644 --- a/src/routes/exports.ts +++ b/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' }) diff --git a/src/services/exportService.ts b/src/services/exportService.ts new file mode 100644 index 00000000..9c81fc38 --- /dev/null +++ b/src/services/exportService.ts @@ -0,0 +1,335 @@ +import { createHash } from 'crypto'; +import type { Knex } from 'knex'; +import { enforceExportQuota } from './quotaService'; + +export type ExportScope = 'self' | 'org' | 'admin'; +export type ExportFormat = 'csv' | 'json' | 'xlsx'; +export type ExportStatus = 'pending' | 'processing' | 'completed' | 'failed' | 'cancelled'; + +export interface ExportJob { + id: string; + requester_user_id: string; + requester_is_admin: boolean; + target_user_id: string | null; + scope: ExportScope; + format: ExportFormat; + status: ExportStatus; + attempts: number; + max_attempts: number; + idempotency_key: string | null; + request_hash: string; + error: string | null; + result_data: Buffer | null; + filename: string | null; + s3_key: string | null; + created_at: Date; + completed_at: Date | null; +} + +export interface CreateExportJobParams { + requesterUserId: string; + requesterIsAdmin?: boolean; + targetUserId?: string | null; + scope: ExportScope; + format: ExportFormat; + orgId?: string | null; + quotaLimit?: number | null; + idempotencyKey?: string | null; + requestHash?: string | null; + maxAttempts?: number; +} + +export class ExportJobError extends Error { + constructor(message: string, public statusCode = 400) { + super(message); + this.name = 'ExportJobError'; + } +} + +export class ExportJobNotFoundError extends ExportJobError { + constructor() { + super('Export job not found', 404); + this.name = 'ExportJobNotFoundError'; + } +} + +export class ExportJobConflictError extends ExportJobError { + constructor(message: string) { + super(message, 409); + this.name = 'ExportJobConflictError'; + } +} + +function computeRequestHash(data: Record): string { + return createHash('sha256').update(JSON.stringify(data)).digest('hex'); +} + +async function lockIdempotencyKey( + trx: Knex.Transaction, + requesterUserId: string, + idempotencyKey: string, +): Promise { + await trx.raw('SELECT pg_advisory_xact_lock(hashtextextended(?, 0))', [ + `export_job:${requesterUserId}:${idempotencyKey}`, + ]); +} + +async function transitionJob( + knex: Knex, + jobId: string, + fromStatuses: ExportStatus[], + toStatus: ExportStatus, + additionalData: Record = {}, +): Promise { + const trx = await knex.transaction(); + try { + const [job] = await trx('export_jobs') + .where({ id: jobId }) + .forUpdate() + .select('*'); + if (!job) { + throw new ExportJobNotFoundError(); + } + if (!fromStatuses.includes(job.status)) { + throw new ExportJobConflictError( + Cannot transition job from ${job.status} to ${toStatus}, + ); + } + + const updateData = { + status: toStatus, + updated_at: trx.fn.now(), + ...additionalData, + }; + + const [updated] = await trx('export_jobs') + .where({ id: jobId, status: job.status }) + .update(updateData) + .returning('*'); + + await trx.commit(); + return updated as ExportJob; + } catch (err) { + await trx.rollback(); + throw err; + } +} + +export async function createExportJob( + knex: Knex, + params: CreateExportJobParams, +): Promise { + const { + requesterUserId, + requesterIsAdmin = false, + targetUserId = null, + scope, + format, + orgId = null, + quotaLimit = null, + idempotencyKey = null, + requestHash = null, + maxAttempts = 3, + } = params; + + const effectiveRequestHash = + requestHash ?? computeRequestHash({ + requesterUserId, + targetUserId, + scope, + format, + }); + + const trx = await knex.transaction(); + + try { + if (idempotencyKey) { + // Serialize all create attempts for this idempotency key. + await lockIdempotencyKey(trx, requesterUserId, idempotencyKey); + + const existing = await trx('export_jobs') + .where({ + requester_user_id: requesterUserId, + idempotency_key: idempotencyKey, + }) + .forUpdate() + .first(); + + if (existing) { + if (existing.status === 'pending' || existing.status === 'processing') { + await trx.commit(); + return existing as ExportJob; + } + + if (existing.status === 'completed') { + await trx.commit(); + return existing as ExportJob; + } + + if (existing.status === 'failed') { + if (existing.attempts < existing.max_attempts) { + const [updated] = await trx('export_jobs') + .where({ id: existing.id, status: 'failed' }) + .update({ + status: 'pending', + attempts: existing.attempts + 1, + error: null, + updated_at: trx.fn.now(), + }) + .returning('*'); + await trx.commit(); + return updated as ExportJob; + } + throw new ExportJobConflictError( + 'Export job has exhausted retry attempts', + ); + } + + if (existing.status === 'cancelled') { + throw new ExportJobConflictError('Export job has been cancelled'); + } + } + } + + // Enforce quota before creating a new job. + if (orgId && quotaLimit !== null) { + const today = new Date().toISOString().slice(0, 10); + await enforceExportQuota(trx, orgId, today, quotaLimit); + } + + const [job] = await trx('export_jobs') + .insert({ + requester_user_id: requesterUserId, + requester_is_admin: requesterIsAdmin, + target_user_id: targetUserId, + scope, + format, + status: 'pending', + attempts: 0, + max_attempts: maxAttempts, + idempotency_key: idempotencyKey, + request_hash: effectiveRequestHash, + error: null, + result_data: null, + filename: null, + s3_key: null, + }) + .returning('*'); + + await trx.commit(); + return job as ExportJob; + } catch (err) { + await trx.rollback(); + throw err; + } +} + +export async function getExportJob( + knex: Knex, + jobId: string, + userId: string, + isAdmin = false, +): Promise { + const query = knex('export_jobs').where({ id: jobId }).first(); + if (!isAdmin) { + query.andWhere({ requester_user_id: userId }); + } + const job = await query; + if (!job) { + throw new ExportJobNotFoundError(); + } + return job as ExportJob; +} + +export async function listExportJobs( + knex: Knex, + userId: string, + isAdmin = false, + status?: ExportStatus, + limit = 50, + offset = 0, +): Promise<{ jobs: ExportJob[]; total: number }> { + const query = knex('export_jobs'); + if (!isAdmin) { + query.where({ requester_user_id: userId }); + } + if (status) { + query.where({ status }); + } + const totalQuery = query.clone().count<{ count: string }=[]>('* as count'); + const [{ count }] = await totalQuery; + const jobs = await query.orderBy('created_at', 'desc').limit(limit).offset(offset); + return { jobs: jobs as ExportJob[], total: Number(count) }; +} + +export async function cancelExportJob( + knex: Knex, + jobId: string, + userId: string, + isAdmin = false, +): Promise { + const trx = await knex.transaction(); + try { + const query = trx('export_jobs').where({ id: jobId }).forUpdate(); + if (!isAdmin) { + query.andWhere({ requester_user_id: userId }); + } + const job = await query.first(); + if (!job) { + throw new ExportJobNotFoundError(); + } + if (job.status === 'pending' || job.status === 'processing') { + const [updated] = await trx('export_jobs') + .where({ id: jobId, status: job.status }) + .update({ + status: 'cancelled', + updated_at: trx.fn.now(), + }) + .returning('*'); + await trx.commit(); + return updated as ExportJob; + } + throw new ExportJobConflictError( + Cannot cancel job in state ${job.status}, + ); + } catch (err) { + await trx.rollback(); + throw err; + } +} + +export async function markJobProcessing( + knex: Knex, + jobId: string, +): Promise { + return transitionJob(knex, jobId, ['pending'], 'processing'); +} + +export async function completeExportJob( + knex: Knex, + jobId: string, + data: { + resultData?: Buffer | null; + filename?: string | null; + s3Key?: string | null; + }, +): Promise { + const updateData: Record = { + completed_at: new Date(), + }; + if ('resultData' in data) updateData.result_data = data.resultData ?? null; + if ('filename' in data) updateData.filename = data.filename ?? null; + if ('s3Key' in data) updateData.s3_key = data.s3Key ?? null; + return transitionJob(knex, jobId, ['processing'], 'completed', updateData); +} + +export async function failExportJob( + knex: Knex, + jobId: string, + error: Error | string, +): Promise { + const message = typeof error === 'string' ? error : error.message; + return transitionJob(knex, jobId, ['processing'], 'failed', { + error: message, + }); +} diff --git a/src/services/quotaService.ts b/src/services/quotaService.ts new file mode 100644 index 00000000..2daf71cb --- /dev/null +++ b/src/services/quotaService.ts @@ -0,0 +1,61 @@ +import type { Knex } from 'knex'; + +export class QuotaExceededError extends Error { + constructor( + public orgId: string, + public quotaDate: string, + public limit: number, + ) { + super(`Export quota exceeded for org ${orgId} on ${quotaDate} (limit: ${limit})`); + this.name = 'QuotaExceededError'; + } +} + +/** + * Atomically increments the export count for the given org/date/metric. + * Throws QuotaExceededError if the count would exceed the limit. + */ +export async function enforceExportQuota( + trx: Knex.Transaction, + orgId: string, + quotaDate: string, + limit: number, +): Promise { + if (limit <= 0) { + throw new QuotaExceededError(orgId, quotaDate, limit); + } + + // Serialize quota increments for the same org/date/metric. + await trx.raw('SELECT pg_advisory_xact_lock(hashtextextended(?, 0))', [ + `${orgId}:${quotaDate}:exports`, + ]); + + const existing = await trx('org_quotas') + .where({ + org_id: orgId, + quota_date: quotaDate, + metric: 'exports', + }) + .forUpdate() + .first(); + + if (existing) { + if (existing.count >= limit) { + throw new QuotaExceededError(orgId, quotaDate, limit); + } + await trx('org_quotas') + .where({ org_id: orgId, quota_date: quotaDate, metric: 'exports' }) + .update({ + count: existing.count + 1, + updated_at: trx.fn.now(), + }); + } else { + await trx('org_quotas').insert({ + org_id: orgId, + quota_date: quotaDate, + metric: 'exports', + count: 1, + limit: + }); + } +}