diff --git a/BackEnd/src/common/services/bulkhead.service.spec.ts b/BackEnd/src/common/services/bulkhead.service.spec.ts index f17e7eb7e..822d76d5e 100644 --- a/BackEnd/src/common/services/bulkhead.service.spec.ts +++ b/BackEnd/src/common/services/bulkhead.service.spec.ts @@ -7,22 +7,33 @@ describe('BulkheadService', () => { let active = 0; let maxActive = 0; - const first = service.runWithBulkhead('payouts', async () => { - active += 1; - maxActive = Math.max(maxActive, active); - await new Promise((resolve) => setTimeout(resolve, 25)); - active -= 1; - return 'first'; - }, { maxConcurrent: 1, maxQueueSize: 5 }); - - const second = service.runWithBulkhead('payouts', async () => { - active += 1; - maxActive = Math.max(maxActive, active); - active -= 1; - return 'second'; - }, { maxConcurrent: 1, maxQueueSize: 5 }); - - await expect(Promise.all([first, second])).resolves.toEqual(['first', 'second']); + const first = service.runWithBulkhead( + 'payouts', + async () => { + active += 1; + maxActive = Math.max(maxActive, active); + await new Promise((resolve) => setTimeout(resolve, 25)); + active -= 1; + return 'first'; + }, + { maxConcurrent: 1, maxQueueSize: 5 }, + ); + + const second = service.runWithBulkhead( + 'payouts', + async () => { + active += 1; + maxActive = Math.max(maxActive, active); + active -= 1; + return 'second'; + }, + { maxConcurrent: 1, maxQueueSize: 5 }, + ); + + await expect(Promise.all([first, second])).resolves.toEqual([ + 'first', + 'second', + ]); expect(maxActive).toBe(1); }); @@ -30,22 +41,30 @@ describe('BulkheadService', () => { const service = new BulkheadService(); let release!: () => void; - const first = service.runWithBulkhead('webhooks', async () => { - await new Promise((resolve) => { - release = resolve; - }); - return 'first'; - }, { maxConcurrent: 1, maxQueueSize: 1 }); + const first = service.runWithBulkhead( + 'webhooks', + async () => { + await new Promise((resolve) => { + release = resolve; + }); + return 'first'; + }, + { maxConcurrent: 1, maxQueueSize: 1 }, + ); const blocked = service.runWithBulkhead('webhooks', async () => 'blocked', { maxConcurrent: 1, maxQueueSize: 1, }); - const rejected = service.runWithBulkhead('webhooks', async () => 'rejected', { - maxConcurrent: 1, - maxQueueSize: 1, - }); + const rejected = service.runWithBulkhead( + 'webhooks', + async () => 'rejected', + { + maxConcurrent: 1, + maxQueueSize: 1, + }, + ); release(); diff --git a/BackEnd/src/common/services/bulkhead.service.ts b/BackEnd/src/common/services/bulkhead.service.ts index ce36a16b8..cca2ce686 100644 --- a/BackEnd/src/common/services/bulkhead.service.ts +++ b/BackEnd/src/common/services/bulkhead.service.ts @@ -1,4 +1,8 @@ -import { Injectable, Logger, ServiceUnavailableException } from '@nestjs/common'; +import { + Injectable, + Logger, + ServiceUnavailableException, +} from '@nestjs/common'; interface BulkheadOptions { maxConcurrent: number; @@ -14,7 +18,10 @@ interface QueuedTask { @Injectable() export class BulkheadService { private readonly logger = new Logger(BulkheadService.name); - private readonly state = new Map[] }>(); + private readonly state = new Map< + string, + { running: number; queue: QueuedTask[] } + >(); async runWithBulkhead( key: string, @@ -57,7 +64,10 @@ export class BulkheadService { return entry; } - private async drainQueue(key: string, options: BulkheadOptions): Promise { + private async drainQueue( + key: string, + options: BulkheadOptions, + ): Promise { const state = this.getOrCreateState(key); while (state.running < options.maxConcurrent && state.queue.length > 0) { const next = state.queue.shift(); diff --git a/BackEnd/src/common/upload/file-scan.interceptor.spec.ts b/BackEnd/src/common/upload/file-scan.interceptor.spec.ts index d94ea8cfd..e814698e5 100644 --- a/BackEnd/src/common/upload/file-scan.interceptor.spec.ts +++ b/BackEnd/src/common/upload/file-scan.interceptor.spec.ts @@ -48,12 +48,16 @@ describe('FileScanInterceptor', () => { }; it('passes a valid, clean file through to the handler', async () => { - const virusScan = { assertFileClean: jest.fn().mockResolvedValue({ clean: true }) }; - const Interceptor = FileScanInterceptor({ allowedMimeTypes: ['image/png'] }); + const virusScan = { + assertFileClean: jest.fn().mockResolvedValue({ clean: true }), + }; + const Interceptor = FileScanInterceptor({ + allowedMimeTypes: ['image/png'], + }); - await expect(run(Interceptor, virusScan, contextWith(makeFile()))).resolves.toBe( - 'handler-result', - ); + await expect( + run(Interceptor, virusScan, contextWith(makeFile())), + ).resolves.toBe('handler-result'); expect(virusScan.assertFileClean).toHaveBeenCalledTimes(1); }); @@ -76,7 +80,9 @@ describe('FileScanInterceptor', () => { .fn() .mockRejectedValue(new UnprocessableEntityException('malware')), }; - const Interceptor = FileScanInterceptor({ allowedMimeTypes: ['image/png'] }); + const Interceptor = FileScanInterceptor({ + allowedMimeTypes: ['image/png'], + }); await expect( run(Interceptor, virusScan, contextWith(makeFile())), @@ -87,8 +93,8 @@ describe('FileScanInterceptor', () => { const virusScan = { assertFileClean: jest.fn() }; const Interceptor = FileScanInterceptor({ required: false }); - await expect(run(Interceptor, virusScan, contextWith(undefined))).resolves.toBe( - 'handler-result', - ); + await expect( + run(Interceptor, virusScan, contextWith(undefined)), + ).resolves.toBe('handler-result'); }); }); diff --git a/BackEnd/src/common/upload/file-scan.interceptor.ts b/BackEnd/src/common/upload/file-scan.interceptor.ts index e8382a8f6..59665ef4f 100644 --- a/BackEnd/src/common/upload/file-scan.interceptor.ts +++ b/BackEnd/src/common/upload/file-scan.interceptor.ts @@ -87,7 +87,10 @@ export function FileScanInterceptor( } } - intercept(context: ExecutionContext, next: CallHandler): Observable { + intercept( + context: ExecutionContext, + next: CallHandler, + ): Observable { return from(this.process(context)).pipe(switchMap(() => next.handle())); } } diff --git a/BackEnd/src/common/upload/file-validation.util.spec.ts b/BackEnd/src/common/upload/file-validation.util.spec.ts index c99e6e767..326223cba 100644 --- a/BackEnd/src/common/upload/file-validation.util.spec.ts +++ b/BackEnd/src/common/upload/file-validation.util.spec.ts @@ -41,7 +41,10 @@ describe('validateUploadedFile', () => { it('rejects an empty file', () => { expect(() => - validateUploadedFile(makeFile({ size: 0, buffer: Buffer.alloc(0) }), options), + validateUploadedFile( + makeFile({ size: 0, buffer: Buffer.alloc(0) }), + options, + ), ).toThrow(BadRequestException); }); @@ -85,10 +88,14 @@ describe('hasMatchingMagicBytes', () => { }); it('returns false for mismatched bytes', () => { - expect(hasMatchingMagicBytes(Buffer.from('hello'), 'image/png')).toBe(false); + expect(hasMatchingMagicBytes(Buffer.from('hello'), 'image/png')).toBe( + false, + ); }); it('returns true for types without a known signature', () => { - expect(hasMatchingMagicBytes(Buffer.from('plain'), 'text/plain')).toBe(true); + expect(hasMatchingMagicBytes(Buffer.from('plain'), 'text/plain')).toBe( + true, + ); }); }); diff --git a/BackEnd/src/common/upload/file-validation.util.ts b/BackEnd/src/common/upload/file-validation.util.ts index 266db0fd5..3f407b12d 100644 --- a/BackEnd/src/common/upload/file-validation.util.ts +++ b/BackEnd/src/common/upload/file-validation.util.ts @@ -92,7 +92,10 @@ export const validateUploadedFile = ( ); } - if (options.verifyMagicBytes && !hasMatchingMagicBytes(file.buffer, file.mimetype)) { + if ( + options.verifyMagicBytes && + !hasMatchingMagicBytes(file.buffer, file.mimetype) + ) { throw new UnsupportedMediaTypeException( `File content does not match its declared type "${file.mimetype}"`, ); diff --git a/BackEnd/src/common/upload/scanners/clamav-virus-scanner.ts b/BackEnd/src/common/upload/scanners/clamav-virus-scanner.ts index cb9bfa9c7..762b191f8 100644 --- a/BackEnd/src/common/upload/scanners/clamav-virus-scanner.ts +++ b/BackEnd/src/common/upload/scanners/clamav-virus-scanner.ts @@ -45,8 +45,15 @@ export class ClamAvVirusScanner implements VirusScanner { socket.on('connect', () => { socket.write('zINSTREAM\0'); - for (let offset = 0; offset < buffer.length; offset += this.options.chunkSize) { - const slice = buffer.subarray(offset, offset + this.options.chunkSize); + for ( + let offset = 0; + offset < buffer.length; + offset += this.options.chunkSize + ) { + const slice = buffer.subarray( + offset, + offset + this.options.chunkSize, + ); const size = Buffer.alloc(4); size.writeUInt32BE(slice.length, 0); socket.write(size); diff --git a/BackEnd/src/common/upload/virus-scan.service.spec.ts b/BackEnd/src/common/upload/virus-scan.service.spec.ts index d5f5d38d1..c683406aa 100644 --- a/BackEnd/src/common/upload/virus-scan.service.spec.ts +++ b/BackEnd/src/common/upload/virus-scan.service.spec.ts @@ -21,11 +21,13 @@ describe('VirusScanService', () => { }; const service = new VirusScanService(scanner, makeConfig(false)); - await expect(service.assertBufferClean(Buffer.from('ok'))).resolves.toEqual({ - clean: true, - virus: null, - scanner: 'fake', - }); + await expect(service.assertBufferClean(Buffer.from('ok'))).resolves.toEqual( + { + clean: true, + virus: null, + scanner: 'fake', + }, + ); }); it('rejects an infected buffer', async () => { @@ -39,9 +41,9 @@ describe('VirusScanService', () => { }; const service = new VirusScanService(scanner, makeConfig(false)); - await expect(service.assertBufferClean(Buffer.from('bad'))).rejects.toBeInstanceOf( - UnprocessableEntityException, - ); + await expect( + service.assertBufferClean(Buffer.from('bad')), + ).rejects.toBeInstanceOf(UnprocessableEntityException); }); it('fails open when the scanner errors and fail-closed is off', async () => { @@ -63,8 +65,8 @@ describe('VirusScanService', () => { }; const service = new VirusScanService(scanner, makeConfig(true)); - await expect(service.assertBufferClean(Buffer.from('x'))).rejects.toBeInstanceOf( - UnprocessableEntityException, - ); + await expect( + service.assertBufferClean(Buffer.from('x')), + ).rejects.toBeInstanceOf(UnprocessableEntityException); }); }); diff --git a/BackEnd/src/common/upload/virus-scan.service.ts b/BackEnd/src/common/upload/virus-scan.service.ts index 18e28af6e..cb86df875 100644 --- a/BackEnd/src/common/upload/virus-scan.service.ts +++ b/BackEnd/src/common/upload/virus-scan.service.ts @@ -47,7 +47,11 @@ export class VirusScanService { ); } this.logger.warn(`Virus scan failed (fail-open): ${message}`); - return { clean: true, virus: null, scanner: `${this.scanner.name}:degraded` }; + return { + clean: true, + virus: null, + scanner: `${this.scanner.name}:degraded`, + }; } if (!result.clean) { diff --git a/BackEnd/src/config/__tests__/versioning.config.spec.ts b/BackEnd/src/config/__tests__/versioning.config.spec.ts index 55f7c466b..b72b7e5ed 100644 --- a/BackEnd/src/config/__tests__/versioning.config.spec.ts +++ b/BackEnd/src/config/__tests__/versioning.config.spec.ts @@ -68,7 +68,9 @@ describe('versioning.config', () => { describe('isVersionDeprecated', () => { it('returns false for the current active version', () => { - expect(isVersionDeprecated(API_VERSION_CONFIG.defaultVersion)).toBe(false); + expect(isVersionDeprecated(API_VERSION_CONFIG.defaultVersion)).toBe( + false, + ); }); }); diff --git a/BackEnd/src/modules/health/health.controller.ts b/BackEnd/src/modules/health/health.controller.ts index 0cc35436a..cea58ba77 100644 --- a/BackEnd/src/modules/health/health.controller.ts +++ b/BackEnd/src/modules/health/health.controller.ts @@ -92,9 +92,11 @@ export class HealthController { }) @ApiResponse({ status: 200, description: 'Stellar Horizon is healthy' }) @ApiResponse({ status: 503, description: 'Stellar Horizon is unreachable' }) - async stellar( - @Res({ passthrough: true }) res: Response, - ): Promise<{ status: ServiceStatus; timestamp: string; service: ServiceHealth }> { + async stellar(@Res({ passthrough: true }) res: Response): Promise<{ + status: ServiceStatus; + timestamp: string; + service: ServiceHealth; + }> { const result = await this.externalHealth.checkStellar(); res.status(result.status === 'down' ? 503 : 200); @@ -114,9 +116,11 @@ export class HealthController { }) @ApiResponse({ status: 200, description: 'Redis is healthy' }) @ApiResponse({ status: 503, description: 'Redis is unreachable' }) - async redis( - @Res({ passthrough: true }) res: Response, - ): Promise<{ status: ServiceStatus; timestamp: string; service: ServiceHealth }> { + async redis(@Res({ passthrough: true }) res: Response): Promise<{ + status: ServiceStatus; + timestamp: string; + service: ServiceHealth; + }> { const result = await this.cacheHealth.check(); res.status(result.status === 'down' ? 503 : 200); @@ -284,10 +288,7 @@ export class HealthController { ): ServiceStatus { const [databaseResult, cacheResult, externalResult] = results; - if ( - databaseResult?.status === 'down' || - cacheResult?.status === 'down' - ) { + if (databaseResult?.status === 'down' || cacheResult?.status === 'down') { return 'down'; } diff --git a/BackEnd/src/modules/jobs/jobs.module.ts b/BackEnd/src/modules/jobs/jobs.module.ts index 84e345dd8..272a8d79a 100644 --- a/BackEnd/src/modules/jobs/jobs.module.ts +++ b/BackEnd/src/modules/jobs/jobs.module.ts @@ -6,6 +6,7 @@ import { JobsService } from './jobs.service'; import { JobsController } from './jobs.controller'; import { JobLogService } from './services/job-log.service'; import { JobSchedulerService } from './services/job-scheduler.service'; +import { JobIdempotencyService } from './services/job-idempotency.service'; import { PayoutProcessor } from './processors/payout.processor'; import { PayoutReconciliationProcessor } from './processors/payout-reconciliation.processor'; import { EmailProcessor } from './processors/email.processor'; @@ -33,6 +34,10 @@ import { DependencyFreshnessService } from '../../common/services/dependency-fre import { EventStore } from '../../events/entities/event-store.entity'; import { User } from '../users/entities/user.entity'; import { EmailModule } from '../email/email.module'; +// Import the IdempotencyKey entity and IdempotencyService from the payouts +// module so that job-level idempotency can reuse the same persistence layer. +import { IdempotencyKey } from '../payouts/entities/idempotency-key.entity'; +import { IdempotencyService } from '../payouts/services/idempotency.service'; @Module({ imports: [ @@ -47,6 +52,8 @@ import { EmailModule } from '../email/email.module'; Submission, EventStore, User, + // Needed for IdempotencyService which is used by JobIdempotencyService + IdempotencyKey, ]), EventEmitterModule, HttpClientModule, @@ -58,6 +65,10 @@ import { EmailModule } from '../email/email.module'; JobsService, JobLogService, JobSchedulerService, + // Idempotency — shared entity/service wired directly so JobsModule has no + // circular dependency on PayoutsModule. + IdempotencyService, + JobIdempotencyService, PayoutProcessor, PayoutReconciliationProcessor, EmailProcessor, @@ -76,6 +87,7 @@ import { EmailModule } from '../email/email.module'; JobsService, JobLogService, JobSchedulerService, + JobIdempotencyService, PayoutProcessor, PayoutReconciliationProcessor, EmailProcessor, diff --git a/BackEnd/src/modules/jobs/jobs.service.ts b/BackEnd/src/modules/jobs/jobs.service.ts index 4b9ac44b5..1f9244973 100644 --- a/BackEnd/src/modules/jobs/jobs.service.ts +++ b/BackEnd/src/modules/jobs/jobs.service.ts @@ -13,6 +13,7 @@ import { } from './job-retry-policy'; import { JobType } from './job.types'; import { DataExportProcessor } from './processors/export.processor'; +import { PayoutProcessor } from './processors/payout.processor'; import { TracingService, TraceContext, @@ -48,6 +49,7 @@ export class JobsService implements OnModuleInit, OnModuleDestroy { constructor( private readonly tracing: TracingService, private readonly dataExportProcessor?: DataExportProcessor, + private readonly payoutProcessor?: PayoutProcessor, ) {} registerEmailProcessor( @@ -76,12 +78,16 @@ export class JobsService implements OnModuleInit, OnModuleDestroy { ); this.queues[QUEUES.EMAIL] = new Queue(QUEUES.EMAIL, redisConnection()); this.queues[QUEUES.EXPORTS] = new Queue(QUEUES.EXPORTS, redisConnection()); + // ── Payouts queue — registered here so PAYOUT_PROCESS / PAYOUT_SETTLE + // jobs can be enqueued via JobsService.addJob(QUEUES.PAYOUTS, ...). + this.queues[QUEUES.PAYOUTS] = new Queue(QUEUES.PAYOUTS, redisConnection()); this.createWorker(QUEUES.NOTIFICATIONS, this.handleNotification.bind(this)); this.createWorker(QUEUES.ANALYTICS, this.handleAnalytics.bind(this)); this.createWorker(QUEUES.CLEANUP, this.handleCleanup.bind(this)); this.createWorker(QUEUES.SCHEDULED, this.handleScheduled.bind(this)); this.createWorker(QUEUES.EMAIL, this.handleEmail.bind(this)); + if ( this.dataExportProcessor && typeof this.dataExportProcessor.processExport === 'function' @@ -91,6 +97,19 @@ export class JobsService implements OnModuleInit, OnModuleDestroy { this.dataExportProcessor.processExport.bind(this.dataExportProcessor), ); } + + // ── Wire the PayoutProcessor to the PAYOUTS queue ───────────────────── + // This ensures each payout job is processed through the idempotency-aware + // PayoutProcessor regardless of how it was enqueued. + if ( + this.payoutProcessor && + typeof this.payoutProcessor.process === 'function' + ) { + this.createWorker( + QUEUES.PAYOUTS, + this.payoutProcessor.process.bind(this.payoutProcessor), + ); + } } async onModuleDestroy() { diff --git a/BackEnd/src/modules/jobs/processors/payout.processor.ts b/BackEnd/src/modules/jobs/processors/payout.processor.ts index 5f37a78b3..7f492b491 100644 --- a/BackEnd/src/modules/jobs/processors/payout.processor.ts +++ b/BackEnd/src/modules/jobs/processors/payout.processor.ts @@ -1,24 +1,87 @@ import { Injectable, Logger } from '@nestjs/common'; import { Job } from 'bullmq'; -import { PayoutProcessPayload, JobResult } from '../job.types'; +import { PayoutProcessPayload, JobResult, JobType } from '../job.types'; import { JobLogService } from '../services/job-log.service'; +import { JobIdempotencyService } from '../services/job-idempotency.service'; /** * Payout Processor - * Handles payout processing jobs - validates and executes Stellar payment transactions + * + * Handles payout processing jobs — validates and executes Stellar payment + * transactions. + * + * Idempotency + * ─────────── + * Each payout job is guarded by a deterministic idempotency key of the form: + * `payout-job:{payoutId}:payout:process` + * + * This guarantees that even if BullMQ retries the same job (or the scheduler + * enqueues it more than once), the actual Stellar payment is submitted only + * once and subsequent executions return the cached result immediately. + * + * Lifecycle: + * 1. `checkAndLock` — acquire idempotency lock or detect a duplicate. + * 2. Process — perform validation + Stellar transaction. + * 3. `complete` — persist the result and unlock. + * 4. `release` — on unrecoverable failure, remove the lock so the + * next genuine BullMQ retry can re-acquire it. */ @Injectable() export class PayoutProcessor { private readonly logger = new Logger(PayoutProcessor.name); - constructor(private readonly jobLogService: JobLogService) {} + constructor( + private readonly jobLogService: JobLogService, + private readonly jobIdempotencyService: JobIdempotencyService, + ) {} /** - * Process payout job + * Process a payout job. + * + * Returns immediately with the cached result if the same payoutId has + * already been processed successfully. Skips gracefully if another worker + * currently holds the lock (in-flight duplicate). */ async process(job: Job): Promise { const { payoutId, organizationId, amount, recipientAddress } = job.data; + // ── 1. Idempotency check ──────────────────────────────────────────────── + const idempotencyKey = this.jobIdempotencyService.buildPayoutJobKey( + payoutId, + JobType.PAYOUT_PROCESS, + ); + + const idempotencyCheck = + await this.jobIdempotencyService.checkAndLock(idempotencyKey); + + if (idempotencyCheck.alreadyProcessed) { + this.logger.log( + `Payout job ${job.id} (payoutId=${payoutId}) already processed — ` + + `returning cached result`, + ); + // Return the previously recorded result directly. + return ( + (idempotencyCheck.result as unknown as JobResult) ?? { + success: true, + data: { payoutId, cachedAt: new Date(), alreadyProcessed: true }, + duration: 0, + } + ); + } + + if (idempotencyCheck.locked) { + this.logger.warn( + `Payout job ${job.id} (payoutId=${payoutId}) is already in-flight — ` + + `skipping duplicate execution`, + ); + return { + success: true, + data: { payoutId, skippedAt: new Date(), inFlight: true }, + duration: 0, + }; + } + + // ── 2. Process the payout ─────────────────────────────────────────────── try { await job.updateProgress(10); this.logger.log( @@ -27,10 +90,13 @@ export class PayoutProcessor { // Validation if (!payoutId || !organizationId || !amount || !recipientAddress) { + // Release the lock so a corrected re-submission can proceed. + await this.jobIdempotencyService.release(idempotencyKey); throw new Error('Missing required payout fields'); } if (amount <= 0) { + await this.jobIdempotencyService.release(idempotencyKey); throw new Error('Payout amount must be greater than zero'); } @@ -38,6 +104,7 @@ export class PayoutProcessor { // Validate Stellar address format (simplified check) if (!recipientAddress.startsWith('G') || recipientAddress.length !== 56) { + await this.jobIdempotencyService.release(idempotencyKey); throw new Error('Invalid Stellar recipient address'); } @@ -51,8 +118,8 @@ export class PayoutProcessor { // 4. Submit to network // 5. Wait for confirmation - // Simulate payout processing - const transactionHash = `tx_${Date.now()}_${Math.random().toString(36).substring(7)}`; + // Simulate payout processing (replace with real Stellar SDK call) + const transactionHash = `tx_${payoutId.substring(0, 8)}_${organizationId.substring(0, 4)}`; await job.updateProgress(75); @@ -77,6 +144,12 @@ export class PayoutProcessor { duration: Date.now() - job.timestamp, }; + // ── 3. Persist the result and release the lock ────────────────────── + await this.jobIdempotencyService.complete( + idempotencyKey, + result as unknown as Record, + ); + this.logger.log(`Payout processed successfully: ${payoutId}`); return result; } catch (error) { @@ -85,6 +158,19 @@ export class PayoutProcessor { error.stack, ); + // ── 4. On unrecoverable error, release the lock ───────────────────── + // This allows BullMQ's built-in retry logic to re-acquire the lock on + // the next attempt. Do not release on validation errors (already done + // above before throwing), only on unexpected runtime errors. + try { + await this.jobIdempotencyService.release(idempotencyKey); + } catch (releaseError) { + this.logger.warn( + `Failed to release idempotency lock for ${payoutId}: ` + + `${releaseError.message}`, + ); + } + throw error; } } diff --git a/BackEnd/src/modules/jobs/services/job-idempotency.service.ts b/BackEnd/src/modules/jobs/services/job-idempotency.service.ts new file mode 100644 index 000000000..6ed0a9e7c --- /dev/null +++ b/BackEnd/src/modules/jobs/services/job-idempotency.service.ts @@ -0,0 +1,170 @@ +import { Injectable, Logger } from '@nestjs/common'; +import { IdempotencyService } from '../../payouts/services/idempotency.service'; + +/** + * Idempotency result returned when checking a job key. + * + * - `alreadyProcessed: true` → a completed record exists; `result` holds the + * previously recorded output so the caller can return it directly. + * - `locked: true` → another worker is currently processing this + * job; the caller should skip/abort this attempt. + * - both false → a fresh lock was acquired; the caller may + * proceed and should call `complete()` (or `release()` on failure) when done. + */ +export interface JobIdempotencyCheck { + alreadyProcessed: boolean; + locked: boolean; + result?: Record | null; +} + +/** + * JobIdempotencyService + * + * Wraps the existing {@link IdempotencyService} with job-queue–specific + * semantics so that payout (and other payment-triggering) jobs become + * idempotent end-to-end, regardless of how many times BullMQ schedules them. + * + * Key schema: `payout-job:{payoutId}:{jobType}` + * + * Lifecycle + * ───────── + * 1. `checkAndLock(key)` — check for a completed or in-progress record. + * • If completed → returns `{ alreadyProcessed: true, result }`. + * • If locked → returns `{ locked: true }`. + * • Otherwise → inserts a locked record and returns `{}` (proceed). + * 2. `complete(key, result)` — mark as completed; stores the job result so + * subsequent duplicates can replay the same output without re-executing. + * 3. `release(key)` — on unrecoverable failure, remove the lock so + * the next genuine retry can re-acquire it. + */ +@Injectable() +export class JobIdempotencyService { + private readonly logger = new Logger(JobIdempotencyService.name); + + /** + * TTL for job idempotency records. + * 7 days is generous enough to cover all realistic retry windows while + * still allowing the `cleanupExpired` job to reclaim the rows. + */ + private readonly JOB_TTL_HOURS = 7 * 24; + + constructor(private readonly idempotencyService: IdempotencyService) {} + + // ── Key Generation ─────────────────────────────────────────────────────── + + /** + * Builds the canonical idempotency key for a payout job. + * + * @param payoutId The internal payout record UUID. + * @param jobType e.g. `'payout:process'` or `'payout:settle'`. + */ + buildPayoutJobKey(payoutId: string, jobType: string): string { + return `payout-job:${payoutId}:${jobType}`; + } + + // ── Core Operations ────────────────────────────────────────────────────── + + /** + * Check whether a job was already processed (or is currently in flight) + * and, if neither, atomically acquire the processing lock. + * + * @param idempotencyKey The unique key for this job execution attempt. + * @returns {@link JobIdempotencyCheck} — see interface docs. + */ + async checkAndLock(idempotencyKey: string): Promise { + this.logger.debug(`Checking job idempotency key: ${idempotencyKey}`); + + // ── 1. Look for an existing (non-expired) record ───────────────────── + const existing = await this.idempotencyService.findByKey(idempotencyKey); + + if (existing) { + if (existing.completedAt) { + // Already ran successfully — return the cached result. + this.logger.log( + `Job idempotency key already completed: ${idempotencyKey}`, + ); + return { + alreadyProcessed: true, + locked: false, + result: existing.responseBody, + }; + } + + if (existing.locked) { + // Another worker holds the lock — signal the caller to skip. + this.logger.warn( + `Job idempotency key is locked (in-flight): ${idempotencyKey}`, + ); + return { alreadyProcessed: false, locked: true }; + } + } + + // ── 2. Try to acquire the lock ──────────────────────────────────────── + const fingerprint = this.idempotencyService.computeFingerprint( + 'JOB', + idempotencyKey, + {}, + ); + + const acquireResult = await this.idempotencyService.tryAcquire( + idempotencyKey, + fingerprint, + 'JOB', // requestMethod — repurposed as "origin" discriminator + idempotencyKey, // requestPath — stores the key itself for traceability + '', // bodyHash — no HTTP body for job payloads + ); + + if (!acquireResult.acquired) { + // Race condition: another instance inserted between our read and write. + const rec = acquireResult.existing; + if (rec?.completedAt) { + this.logger.log( + `Job idempotency key already completed (race): ${idempotencyKey}`, + ); + return { + alreadyProcessed: true, + locked: false, + result: rec.responseBody, + }; + } + this.logger.warn( + `Job idempotency key locked after race: ${idempotencyKey}`, + ); + return { alreadyProcessed: false, locked: true }; + } + + this.logger.debug(`Job idempotency lock acquired: ${idempotencyKey}`); + return { alreadyProcessed: false, locked: false }; + } + + /** + * Mark the job as successfully completed and persist the result so that + * future duplicate executions can return the same output without re-running. + * + * @param idempotencyKey Key that was previously acquired via `checkAndLock`. + * @param result The serialisable job result to cache. + */ + async complete( + idempotencyKey: string, + result: Record, + ): Promise { + // HTTP status code 200 is used as a sentinel for "success" in the + // existing IdempotencyService schema. + await this.idempotencyService.complete(idempotencyKey, 200, result); + this.logger.debug(`Job idempotency key marked complete: ${idempotencyKey}`); + } + + /** + * Release (delete) the idempotency lock on a permanent failure so that + * the next genuine retry can re-acquire it. This should **not** be called + * for transient errors that BullMQ will retry automatically. + * + * @param idempotencyKey Key that was previously acquired via `checkAndLock`. + */ + async release(idempotencyKey: string): Promise { + await this.idempotencyService.remove(idempotencyKey); + this.logger.debug( + `Job idempotency key released (lock removed): ${idempotencyKey}`, + ); + } +} diff --git a/BackEnd/src/modules/jobs/utils/worker-concurrency.util.ts b/BackEnd/src/modules/jobs/utils/worker-concurrency.util.ts index 6316f27c5..96de52a44 100644 --- a/BackEnd/src/modules/jobs/utils/worker-concurrency.util.ts +++ b/BackEnd/src/modules/jobs/utils/worker-concurrency.util.ts @@ -29,7 +29,10 @@ export function workerConcurrencyEnvKey(queue: string): string { } function clampConcurrency(value: number): number { - return Math.min(MAX_WORKER_CONCURRENCY, Math.max(MIN_WORKER_CONCURRENCY, value)); + return Math.min( + MAX_WORKER_CONCURRENCY, + Math.max(MIN_WORKER_CONCURRENCY, value), + ); } /** diff --git a/BackEnd/src/modules/payouts/CHANGELOG.md b/BackEnd/src/modules/payouts/CHANGELOG.md index 27e80d838..f4eaae775 100644 --- a/BackEnd/src/modules/payouts/CHANGELOG.md +++ b/BackEnd/src/modules/payouts/CHANGELOG.md @@ -5,3 +5,6 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/) and this module adheres to [Semantic Versioning](https://semver.org/). ## [Unreleased] + +### Changed +- Code formatting and improved readability in PayoutsService error handling diff --git a/BackEnd/src/modules/payouts/payouts.service.ts b/BackEnd/src/modules/payouts/payouts.service.ts index 31eb8b4f6..937a8dffc 100644 --- a/BackEnd/src/modules/payouts/payouts.service.ts +++ b/BackEnd/src/modules/payouts/payouts.service.ts @@ -163,7 +163,9 @@ export class PayoutsService { if (!settlement.isFinal) { payout.status = PayoutStatus.PROCESSING; - payout.nextRetryAt = new Date(Date.now() + this.settlementRetryDelayMs); + payout.nextRetryAt = new Date( + Date.now() + this.settlementRetryDelayMs, + ); await this.payoutRepository.save(payout); this.logger.log( `Payout ${payoutId} submitted and waiting for settlement finality (${settlement.confirmations}/${settlement.requiredConfirmations} confirmations)`, @@ -181,7 +183,9 @@ export class PayoutsService { payout.status = PayoutStatus.PROCESSING; payout.failureReason = error.message || 'Settlement confirmation failed'; - payout.nextRetryAt = new Date(Date.now() + this.settlementRetryDelayMs); + payout.nextRetryAt = new Date( + Date.now() + this.settlementRetryDelayMs, + ); await this.payoutRepository.save(payout); this.logger.warn( `Payout ${payout.id} transaction submitted but settlement confirmation is unavailable; retry scheduled`, @@ -191,7 +195,11 @@ export class PayoutsService { this.eventEmitter.emit( 'payout.failed', - new PayoutFailedEvent(payout.id, payout.stellarAddress, error.message), + new PayoutFailedEvent( + payout.id, + payout.stellarAddress, + error.message, + ), ); await this.handlePayoutFailure(payout, error); } diff --git a/BackEnd/src/modules/webhooks/CHANGELOG.md b/BackEnd/src/modules/webhooks/CHANGELOG.md index 050866286..0edcf7ab4 100644 --- a/BackEnd/src/modules/webhooks/CHANGELOG.md +++ b/BackEnd/src/modules/webhooks/CHANGELOG.md @@ -5,3 +5,6 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/) and this module adheres to [Semantic Versioning](https://semver.org/). ## [Unreleased] + +### Changed +- Improved error logging formatting in WebhooksService diff --git a/BackEnd/src/modules/webhooks/webhooks.service.ts b/BackEnd/src/modules/webhooks/webhooks.service.ts index 178b4c207..2a10c700e 100644 --- a/BackEnd/src/modules/webhooks/webhooks.service.ts +++ b/BackEnd/src/modules/webhooks/webhooks.service.ts @@ -103,7 +103,10 @@ export class WebhooksService { traceId: currentTraceId(), }; } catch (error) { - this.logger.error(`Failed to process webhook ${event.id}:`, error.stack); + this.logger.error( + `Failed to process webhook ${event.id}:`, + error.stack, + ); return { success: false, eventId: event.id, diff --git a/BackEnd/test/common/dto-whitelisting.spec.ts b/BackEnd/test/common/dto-whitelisting.spec.ts index 05f396179..bdcd605bc 100644 --- a/BackEnd/test/common/dto-whitelisting.spec.ts +++ b/BackEnd/test/common/dto-whitelisting.spec.ts @@ -82,7 +82,7 @@ class AssignRoleDto { // ── Helper ──────────────────────────────────────────────────────────────────── const bodyMeta = (metatype: any) => - ({ type: 'body', metatype, data: '' } as any); + ({ type: 'body', metatype, data: '' }) as any; // ── Tests ────────────────────────────────────────────────────────────────────── @@ -101,7 +101,7 @@ describe('CustomValidationPipe — strict DTO whitelisting', () => { username: 'alice', email: 'alice@example.com', __proto__: { admin: true }, // common prototype-pollution attempt (stripped by JSON.parse but we test the pipe) - isAdmin: true, // plain unknown field + isAdmin: true, // plain unknown field }; await expect( @@ -239,7 +239,10 @@ describe('CustomValidationPipe — strict DTO whitelisting', () => { }); it('should return the raw value when no metatype is provided', async () => { - const result = await pipe.transform({ any: 'thing' }, bodyMeta(undefined)); + const result = await pipe.transform( + { any: 'thing' }, + bodyMeta(undefined), + ); expect(result).toEqual({ any: 'thing' }); }); }); diff --git a/BackEnd/test/health/health.e2e-spec.ts b/BackEnd/test/health/health.e2e-spec.ts index 48c4aa830..72c26b0e7 100644 --- a/BackEnd/test/health/health.e2e-spec.ts +++ b/BackEnd/test/health/health.e2e-spec.ts @@ -176,9 +176,7 @@ describe('Health (e2e)', () => { mockDatabaseHealth.check.mockResolvedValue(okResult); mockCacheHealth.check.mockResolvedValue(okResult); - await request(app.getHttpServer()) - .get('/api/health/ready') - .expect(200); + await request(app.getHttpServer()).get('/api/health/ready').expect(200); // Both should have been called expect(mockDatabaseHealth.check).toHaveBeenCalledTimes(1); diff --git a/BackEnd/test/jobs/job-idempotency.service.spec.ts b/BackEnd/test/jobs/job-idempotency.service.spec.ts new file mode 100644 index 000000000..1dfefa3ff --- /dev/null +++ b/BackEnd/test/jobs/job-idempotency.service.spec.ts @@ -0,0 +1,215 @@ +import { Test, TestingModule } from '@nestjs/testing'; +import { JobIdempotencyService } from 'src/modules/jobs/services/job-idempotency.service'; +import { IdempotencyService } from 'src/modules/payouts/services/idempotency.service'; +import { JobType } from 'src/modules/jobs/job.types'; + +describe('JobIdempotencyService', () => { + let service: JobIdempotencyService; + let idempotencyService: jest.Mocked; + + beforeEach(async () => { + const mockIdempotencyService: jest.Mocked< + Pick< + IdempotencyService, + | 'findByKey' + | 'computeFingerprint' + | 'tryAcquire' + | 'complete' + | 'remove' + > + > = { + findByKey: jest.fn(), + computeFingerprint: jest.fn().mockReturnValue('fingerprint-abc'), + tryAcquire: jest.fn(), + complete: jest.fn(), + remove: jest.fn(), + }; + + const module: TestingModule = await Test.createTestingModule({ + providers: [ + JobIdempotencyService, + { + provide: IdempotencyService, + useValue: mockIdempotencyService, + }, + ], + }).compile(); + + service = module.get(JobIdempotencyService); + idempotencyService = module.get(IdempotencyService); + }); + + // ── buildPayoutJobKey ──────────────────────────────────────────────────── + + describe('buildPayoutJobKey', () => { + it('should build the canonical key for a payout job', () => { + const key = service.buildPayoutJobKey( + 'payout-uuid-123', + JobType.PAYOUT_PROCESS, + ); + expect(key).toBe('payout-job:payout-uuid-123:payout:process'); + }); + + it('should build different keys for different job types', () => { + const processKey = service.buildPayoutJobKey( + 'id', + JobType.PAYOUT_PROCESS, + ); + const settleKey = service.buildPayoutJobKey('id', JobType.PAYOUT_SETTLE); + expect(processKey).not.toEqual(settleKey); + }); + + it('should build different keys for different payoutIds', () => { + const key1 = service.buildPayoutJobKey( + 'payout-1', + JobType.PAYOUT_PROCESS, + ); + const key2 = service.buildPayoutJobKey( + 'payout-2', + JobType.PAYOUT_PROCESS, + ); + expect(key1).not.toEqual(key2); + }); + }); + + // ── checkAndLock — fresh job ────────────────────────────────────────────── + + describe('checkAndLock — fresh job', () => { + it('should acquire lock and return alreadyProcessed=false, locked=false when no record exists', async () => { + idempotencyService.findByKey.mockResolvedValue(null); + idempotencyService.tryAcquire.mockResolvedValue({ acquired: true }); + + const result = await service.checkAndLock('some-key'); + + expect(result.alreadyProcessed).toBe(false); + expect(result.locked).toBe(false); + expect(idempotencyService.tryAcquire).toHaveBeenCalledWith( + 'some-key', + 'fingerprint-abc', + 'JOB', + 'some-key', + '', + ); + }); + }); + + // ── checkAndLock — already completed ──────────────────────────────────── + + describe('checkAndLock — already completed', () => { + it('should return alreadyProcessed=true with cached result when record has completedAt', async () => { + const cachedResult = { + success: true, + data: { transactionHash: 'tx_abc' }, + }; + idempotencyService.findByKey.mockResolvedValue({ + id: 'rec-1', + key: 'some-key', + completedAt: new Date(), + locked: false, + responseBody: cachedResult, + } as any); + + const result = await service.checkAndLock('some-key'); + + expect(result.alreadyProcessed).toBe(true); + expect(result.locked).toBe(false); + expect(result.result).toEqual(cachedResult); + // Should not attempt to acquire the lock + expect(idempotencyService.tryAcquire).not.toHaveBeenCalled(); + }); + }); + + // ── checkAndLock — in-flight (locked) ─────────────────────────────────── + + describe('checkAndLock — in-flight (locked)', () => { + it('should return locked=true when existing record is locked and has no completedAt', async () => { + idempotencyService.findByKey.mockResolvedValue({ + id: 'rec-2', + key: 'some-key', + completedAt: null, + locked: true, + responseBody: null, + } as any); + + const result = await service.checkAndLock('some-key'); + + expect(result.alreadyProcessed).toBe(false); + expect(result.locked).toBe(true); + expect(idempotencyService.tryAcquire).not.toHaveBeenCalled(); + }); + }); + + // ── checkAndLock — race condition (tryAcquire fails) ───────────────────── + + describe('checkAndLock — race condition', () => { + it('should return locked=true when tryAcquire is not acquired and existing record is locked', async () => { + idempotencyService.findByKey.mockResolvedValue(null); + idempotencyService.tryAcquire.mockResolvedValue({ + acquired: false, + existing: { + key: 'some-key', + fingerprint: 'fp', + locked: true, + completedAt: null, + responseBody: null, + responseStatusCode: null, + }, + }); + + const result = await service.checkAndLock('some-key'); + + expect(result.alreadyProcessed).toBe(false); + expect(result.locked).toBe(true); + }); + + it('should return alreadyProcessed=true when tryAcquire fails and existing record is completed', async () => { + const cachedResult = { success: true }; + idempotencyService.findByKey.mockResolvedValue(null); + idempotencyService.tryAcquire.mockResolvedValue({ + acquired: false, + existing: { + key: 'some-key', + fingerprint: 'fp', + locked: false, + completedAt: new Date(), + responseBody: cachedResult, + responseStatusCode: 200, + }, + }); + + const result = await service.checkAndLock('some-key'); + + expect(result.alreadyProcessed).toBe(true); + expect(result.result).toEqual(cachedResult); + }); + }); + + // ── complete ──────────────────────────────────────────────────────────── + + describe('complete', () => { + it('should call idempotencyService.complete with status 200 and the result', async () => { + idempotencyService.complete.mockResolvedValue(undefined); + const result = { success: true, data: { payoutId: 'p-1' } }; + + await service.complete('some-key', result as any); + + expect(idempotencyService.complete).toHaveBeenCalledWith( + 'some-key', + 200, + result, + ); + }); + }); + + // ── release ──────────────────────────────────────────────────────────── + + describe('release', () => { + it('should call idempotencyService.remove with the key', async () => { + idempotencyService.remove.mockResolvedValue(undefined); + + await service.release('some-key'); + + expect(idempotencyService.remove).toHaveBeenCalledWith('some-key'); + }); + }); +}); diff --git a/BackEnd/test/jobs/job-processors.spec.ts b/BackEnd/test/jobs/job-processors.spec.ts index ad9715d23..7a7b24576 100644 --- a/BackEnd/test/jobs/job-processors.spec.ts +++ b/BackEnd/test/jobs/job-processors.spec.ts @@ -7,6 +7,7 @@ import { WebhookProcessor } from 'src/modules/jobs/processors/webhook.processor' import { AnalyticsProcessor } from 'src/modules/jobs/processors/analytics.processor'; import { QuestProcessor } from 'src/modules/jobs/processors/quest.processor'; import { JobLogService } from 'src/modules/jobs/services/job-log.service'; +import { JobIdempotencyService } from 'src/modules/jobs/services/job-idempotency.service'; import { PayoutProcessPayload, EmailSendPayload, @@ -125,6 +126,23 @@ describe('Job Processors', () => { .mockResolvedValue({ id: 'report-1', status: 'COMPLETED' }), }, }, + // Default mock for JobIdempotencyService: always allow processing + { + provide: JobIdempotencyService, + useValue: { + buildPayoutJobKey: jest + .fn() + .mockImplementation( + (payoutId: string, jobType: string) => + `payout-job:${payoutId}:${jobType}`, + ), + checkAndLock: jest + .fn() + .mockResolvedValue({ alreadyProcessed: false, locked: false }), + complete: jest.fn().mockResolvedValue(undefined), + release: jest.fn().mockResolvedValue(undefined), + }, + }, PayoutProcessor, EmailProcessor, DataExportProcessor, diff --git a/BackEnd/test/jobs/worker-concurrency.util.spec.ts b/BackEnd/test/jobs/worker-concurrency.util.spec.ts index 66526238d..e7b94b72a 100644 --- a/BackEnd/test/jobs/worker-concurrency.util.spec.ts +++ b/BackEnd/test/jobs/worker-concurrency.util.spec.ts @@ -10,7 +10,9 @@ import { JOB_QUEUE_CONFIG, QUEUES } from '#src/modules/jobs/jobs.constants'; describe('worker concurrency tuning', () => { describe('workerConcurrencyEnvKey', () => { it('builds an upper-cased QUEUE__CONCURRENCY key', () => { - expect(workerConcurrencyEnvKey('payouts')).toBe('QUEUE_PAYOUTS_CONCURRENCY'); + expect(workerConcurrencyEnvKey('payouts')).toBe( + 'QUEUE_PAYOUTS_CONCURRENCY', + ); expect(workerConcurrencyEnvKey(QUEUES.MAINTENANCE)).toBe( 'QUEUE_MAINTENANCE_CONCURRENCY', ); @@ -40,7 +42,9 @@ describe('worker concurrency tuning', () => { it('clamps an override above the maximum down to MAX_WORKER_CONCURRENCY', () => { const env = { QUEUE_EMAIL_CONCURRENCY: '10000' }; - expect(resolveWorkerConcurrency(QUEUES.EMAIL, env)).toBe(MAX_WORKER_CONCURRENCY); + expect(resolveWorkerConcurrency(QUEUES.EMAIL, env)).toBe( + MAX_WORKER_CONCURRENCY, + ); }); it('rejects an override below the minimum and uses the configured default', () => { diff --git a/BackEnd/test/load-tests/benchmark-config.spec.ts b/BackEnd/test/load-tests/benchmark-config.spec.ts index 2ea628c6d..c59497cfd 100644 --- a/BackEnd/test/load-tests/benchmark-config.spec.ts +++ b/BackEnd/test/load-tests/benchmark-config.spec.ts @@ -1,14 +1,19 @@ -import { createBenchmarkConfig, buildRoute } from '../../load-tests/benchmark-config'; +import { + createBenchmarkConfig, + buildRoute, +} from '../../load-tests/benchmark-config'; describe('benchmark config', () => { it('builds route templates with dynamic parameters', () => { expect(buildRoute('/api/quests/:id', { id: 'quest-1' })).toBe( '/api/quests/quest-1', ); - expect(buildRoute('/api/quests/:questId/submissions/:id', { - questId: 'q-1', - id: 's-1', - })).toBe('/api/quests/q-1/submissions/s-1'); + expect( + buildRoute('/api/quests/:questId/submissions/:id', { + questId: 'q-1', + id: 's-1', + }), + ).toBe('/api/quests/q-1/submissions/s-1'); }); it('includes the critical endpoint scenarios and thresholds', () => {