feat: add leader election and fencing for singleton background jobs

src/audit/services/audit-retention.service.ts

@@ -3,6 +3,7 @@ import { Injectable, Logger } from '@nestjs/common';
 import { Cron, CronExpression } from '@nestjs/schedule';
 import { ConfigService } from '@nestjs/config';
 import { AuditTrailService } from './audit-trail.service';
+import { LeaseService } from '../../common/lease/lease.service';
 import { GracefulShutdownService } from '../../common/lifecycle/graceful-shutdown.service';
 
 export interface RetentionExecutionResult {
@@ -13,6 +14,8 @@ export interface RetentionExecutionResult {
   timestamp: string;
 }
 
+const SINGLETON_JOB_NAME = 'audit-log-retention';
+
 @Injectable()
 export class AuditRetentionService {
   private readonly logger = new Logger(AuditRetentionService.name);
@@ -22,6 +25,7 @@ export class AuditRetentionService {
   constructor(
     private readonly auditTrailService: AuditTrailService,
     private readonly configService: ConfigService,
+    private readonly leaseService: LeaseService,
     private readonly gracefulShutdown: GracefulShutdownService,
   ) {
     this.daysToKeep = this.resolveRetentionDays();
@@ -32,33 +36,53 @@ export class AuditRetentionService {
     getRuntimeEnvironment().AUDIT_LOG_RETENTION_CRON ||
       CronExpression.EVERY_DAY_AT_MIDNIGHT,
     {
-      name: 'audit-log-retention',
+      name: SINGLETON_JOB_NAME,
       timeZone: 'UTC',
     },
   )
-  async enforceRetentionAndPrivacyPolicies(): Promise<RetentionExecutionResult> {
-    return this.gracefulShutdown.runTracked('audit-retention', async () => {
-      this.logger.log(
-        'Starting scheduled data retention and privacy controls execution',
+  async enforceRetentionAndPrivacyPolicies(): Promise<RetentionExecutionResult | null> {
+    const lease = await this.leaseService.tryAcquire(SINGLETON_JOB_NAME);
+    if (!lease.isLeader) {
+      this.logger.debug(
+        `Skipping retention run; not leader for ${SINGLETON_JOB_NAME} (fencing token ${lease.fencingToken})`,
       );
+      return null;
+    }
 
-      const purgedLogsCount = await this.purgeOldAuditLogs();
-      const scrubbedPiiCount = await this.scrubOldPii();
-
-      const result: RetentionExecutionResult = {
-        purgedLogsCount,
-        scrubbedPiiCount,
-        retentionDays: this.daysToKeep,
-        piiRetentionDays: this.piiDaysToKeep,
-        timestamp: new Date().toISOString(),
-      };
+    this.logger.log(
+      `Starting scheduled data retention and privacy controls execution (fencing token ${lease.fencingToken})`,
+    );
 
-      this.logger.log(
-        `Retention and privacy execution complete: purged ${purgedLogsCount} expired records, scrubbed PII from ${scrubbedPiiCount} records`,
+    try {
+      return await this.gracefulShutdown.runTracked('audit-retention', async () => {
+        const purgedLogsCount = await this.purgeOldAuditLogs();
+        await this.leaseService.assertLeadership(lease);
+
+        const scrubbedPiiCount = await this.scrubOldPii();
+        await this.leaseService.assertLeadership(lease);
+
+        const result: RetentionExecutionResult = {
+          purgedLogsCount,
+          scrubbedPiiCount,
+          retentionDays: this.daysToKeep,
+          piiRetentionDays: this.piiDaysToKeep,
+          timestamp: new Date().toISOString(),
+        };
+
+        this.logger.log(
+          `Retention and privacy execution complete: purged ${purgedLogsCount} expired records, scrubbed PII from ${scrubbedPiiCount} records`,
+        );
+
+        return result;
+      });
+    } catch (error) {
+      this.logger.warn(
+        `Retention run aborted for ${SINGLETON_JOB_NAME}: ${error instanceof Error ? error.message : String(error)}`,
       );
-
-      return result;
-    });
+      throw error;
+    } finally {
+      await this.leaseService.release(lease);
+    }
   }
 
   async purgeOldAuditLogs(): Promise<number> {
@@ -79,7 +103,7 @@ export class AuditRetentionService {
     );
 
     this.logger.log(
-      `Privacy controls scrubbed PII from ${scrubbedCount} records older than ${this.piiDaysToKeep} days`,
+      `Privacy controls scrubbed PIF from ${scrubbedCount} records older than ${this.piiDaysToKeep} days`,
     );
 
     return scrubbedCount;
@@ -96,4 +120,4 @@ export class AuditRetentionService {
     const parsedDays = parseInt(rawDays ?? '', 10);
     return Number.isNaN(parsedDays) || parsedDays <= 0 ? 30 : parsedDays;
   }
-}
+}