feat(jobs): add idempotency keys for payout-triggering jobs (BE-103)

BackEnd/src/modules/jobs/jobs.module.ts

@@ -5,6 +5,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';
@@ -32,6 +33,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: [
@@ -46,6 +51,8 @@ import { EmailModule } from '../email/email.module';
       Submission,
       EventStore,
       User,
+      // Needed for IdempotencyService which is used by JobIdempotencyService
+      IdempotencyKey,
     ]),
     EventEmitterModule,
     StellarModule,
@@ -56,6 +63,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,
@@ -74,6 +85,7 @@ import { EmailModule } from '../email/email.module';
     JobsService,
     JobLogService,
     JobSchedulerService,
+    JobIdempotencyService,
     PayoutProcessor,
     PayoutReconciliationProcessor,
     EmailProcessor,

BackEnd/src/modules/jobs/jobs.service.ts

@@ -69,12 +69,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'
@@ -84,6 +88,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() {

BackEnd/src/modules/jobs/processors/payout.processor.ts

@@ -1,24 +1,85 @@
 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<PayoutProcessPayload>): Promise<JobResult> {
     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,17 +88,21 @@ 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');
       }
 
       await job.updateProgress(25);
 
       // 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 +116,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 +142,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<string, unknown>,
+      );
+
       this.logger.log(`Payout processed successfully: ${payoutId}`);
       return result;
     } catch (error) {
@@ -85,6 +156,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;
     }
   }