Fix/part49 leader elected scheduling

backend/src/index.ts

@@ -15,6 +15,10 @@ dotenv.config();
 
 const server = createServer(app);
 
+// Part-49 job handles — assigned on server start, cleaned up on shutdown
+let usageSnapshotJob: { stop(): void };
+let integrityCheckJob: { stop(): void };
+
 // Initialize Socket.IO
 initializeSocket(server);
 
@@ -38,8 +42,9 @@ server.listen(PORT, () => {
   logger.info('ContractEventIndexer initialized');
 
   // Part 49 — daily quota snapshots + nightly audit-chain integrity
-  scheduleDailyUsageSnapshots();
-  scheduleNightlyIntegrityCheck();
+  // (leader-elected via Postgres advisory lock; cron at midnight UTC)
+  usageSnapshotJob = scheduleDailyUsageSnapshots();
+  integrityCheckJob = scheduleNightlyIntegrityCheck();
   logger.info('Part-49 jobs scheduled (usage snapshots + audit integrity)');
 
   // Part 45 — cleanup expired audit cache every hour
@@ -81,6 +86,10 @@ const shutdown = () => {
 
   liquidityAlertChecker.stop();
 
+  // Stop Part-49 cron jobs
+  usageSnapshotJob?.stop();
+  integrityCheckJob?.stop();
+
   // Stop the contract event indexer
   contractEventIndexer.stop();
 

backend/src/jobs/__tests__/part49Jobs.test.ts

@@ -0,0 +1,189 @@
+/**
+ * Tests for Part-49 leader-elected scheduling (part49Jobs.ts).
+ *
+ * Verifies:
+ *  - node-cron is scheduled at midnight UTC (0 0 * * *)
+ *  - Postgres advisory lock ensures only one pod executes per tick
+ *  - Startup catch-up run fires immediately (no lock)
+ *  - stop() halts the cron task
+ */
+
+import { describe, it, expect, jest, beforeEach, afterEach } from '@jest/globals';
+
+// ── Mocks ────────────────────────────────────────────────────────────────────
+
+const mockQuery = jest.fn() as jest.Mock;
+const mockConnect = jest.fn() as jest.Mock;
+const mockRelease = jest.fn() as jest.Mock;
+const mockCronStop = jest.fn() as jest.Mock;
+const mockCronSchedule = jest.fn() as jest.Mock;
+
+jest.mock('../../config/database.js', () => ({
+  __esModule: true,
+  default: { query: mockQuery, connect: mockConnect },
+}));
+
+jest.mock('../../services/tenantQuotaService.js', () => ({
+  tenantQuotaService: { snapshotAllTenants: jest.fn() },
+}));
+
+jest.mock('../../services/auditIntegrityService.js', () => ({
+  auditIntegrityService: { runScheduledCheck: jest.fn() },
+}));
+
+jest.mock('../../utils/logger.js', () => ({
+  __esModule: true,
+  default: {
+    info: jest.fn(),
+    debug: jest.fn(),
+    error: jest.fn(),
+    warn: jest.fn(),
+  },
+}));
+
+jest.mock('node-cron', () => ({
+  __esModule: true,
+  default: { schedule: mockCronSchedule },
+}));
+
+// ── Imports (after mocks) ────────────────────────────────────────────────────
+
+import { scheduleDailyUsageSnapshots, scheduleNightlyIntegrityCheck } from '../part49Jobs.js';
+import { tenantQuotaService } from '../../services/tenantQuotaService.js';
+import { auditIntegrityService } from '../../services/auditIntegrityService.js';
+
+// ── Helpers ──────────────────────────────────────────────────────────────────
+
+function mockAdvisoryLock(acquired: boolean) {
+  mockConnect.mockResolvedValue({
+    query: jest.fn<() => Promise<any>>().mockResolvedValue({
+      rows: [{ acquired }],
+    }),
+    release: mockRelease,
+  });
+}
+
+// ── Tests ────────────────────────────────────────────────────────────────────
+
+describe('Part-49 leader-elected scheduling', () => {
+  beforeEach(() => {
+    jest.clearAllMocks();
+    mockCronSchedule.mockReturnValue({ stop: mockCronStop });
+    mockAdvisoryLock(true);
+  });
+
+  describe('scheduleDailyUsageSnapshots', () => {
+    it('schedules a cron job at midnight UTC', () => {
+      scheduleDailyUsageSnapshots();
+
+      expect(mockCronSchedule).toHaveBeenCalledWith(
+        '0 0 * * *',
+        expect.any(Function),
+        { timezone: 'UTC' },
+      );
+    });
+
+    it('runs the snapshot immediately on startup (catch-up)', async () => {
+      const spy = tenantQuotaService.snapshotAllTenants as jest.Mock;
+      scheduleDailyUsageSnapshots();
+
+      // The startup run is fire-and-forget; let the microtask queue flush
+      await new Promise((r) => setTimeout(r, 10));
+      expect(spy).toHaveBeenCalledTimes(1);
+    });
+
+    it('returns a ScheduledJob with a working stop() method', () => {
+      const job = scheduleDailyUsageSnapshots();
+      job.stop();
+      expect(mockCronStop).toHaveBeenCalled();
+    });
+  });
+
+  describe('scheduleNightlyIntegrityCheck', () => {
+    it('schedules a cron job at midnight UTC', () => {
+      scheduleNightlyIntegrityCheck();
+
+      expect(mockCronSchedule).toHaveBeenCalledWith(
+        '0 0 * * *',
+        expect.any(Function),
+        { timezone: 'UTC' },
+      );
+    });
+
+    it('runs the integrity check immediately on startup (catch-up)', async () => {
+      const spy = auditIntegrityService.runScheduledCheck as jest.Mock;
+      scheduleNightlyIntegrityCheck();
+
+      await new Promise((r) => setTimeout(r, 10));
+      expect(spy).toHaveBeenCalledTimes(1);
+    });
+
+    it('returns a ScheduledJob with a working stop() method', () => {
+      const job = scheduleNightlyIntegrityCheck();
+      job.stop();
+      expect(mockCronStop).toHaveBeenCalled();
+    });
+  });
+
+  describe('advisory lock behaviour (cron tick)', () => {
+    it('executes the job when the advisory lock is acquired', async () => {
+      mockAdvisoryLock(true);
+      const spy = tenantQuotaService.snapshotAllTenants as jest.Mock;
+      scheduleDailyUsageSnapshots();
+
+      // The cron callback fires runWithAdvisoryLock via void (fire-and-forget)
+      const cronCallback = mockCronSchedule.mock.calls[0][1] as () => void;
+      cronCallback();
+      await new Promise((r) => setTimeout(r, 20));
+
+      expect(spy).toHaveBeenCalled(); // 1 from startup + 1 from cron tick
+      expect(mockRelease).toHaveBeenCalled();
+    });
+
+    it('skips execution when the advisory lock is not acquired', async () => {
+      const spy = tenantQuotaService.snapshotAllTenants as jest.Mock;
+      scheduleDailyUsageSnapshots();
+
+      // 1 call from startup
+      await new Promise((r) => setTimeout(r, 20));
+      const callCountAfterStartup = spy.mock.calls.length;
+
+      // Now simulate the advisory lock being held by another pod
+      mockAdvisoryLock(false);
+      const cronCallback = mockCronSchedule.mock.calls[0][1] as () => void;
+      cronCallback();
+      await new Promise((r) => setTimeout(r, 20));
+
+      // No additional execution
+      expect(spy.mock.calls.length).toBe(callCountAfterStartup);
+      expect(mockRelease).toHaveBeenCalled();
+    });
+
+    it('releases the database client even if the job throws', async () => {
+      const spy = tenantQuotaService.snapshotAllTenants as jest.Mock;
+      spy.mockRejectedValueOnce(new Error('quota boom'));
+
+      scheduleDailyUsageSnapshots();
+      const cronCallback = mockCronSchedule.mock.calls[0][1] as () => void;
+      cronCallback();
+      await new Promise((r) => setTimeout(r, 20));
+
+      expect(mockRelease).toHaveBeenCalled();
+    });
+  });
+
+  describe('both jobs use distinct advisory lock IDs', () => {
+    it('the snapshot cron callback acquires a different lock ID than the integrity check', () => {
+      scheduleDailyUsageSnapshots();
+      scheduleNightlyIntegrityCheck();
+
+      const snapshotCallback = mockCronSchedule.mock.calls[0][1] as () => void;
+      const integrityCallback = mockCronSchedule.mock.calls[1][1] as () => void;
+
+      // Both callbacks are functions — verify they exist (lock IDs tested implicitly
+      // via the connect/query mock; distinct IDs are constants in the source)
+      expect(typeof snapshotCallback).toBe('function');
+      expect(typeof integrityCallback).toBe('function');
+    });
+  });
+});

backend/src/jobs/part49Jobs.ts

@@ -1,30 +1,100 @@
+import cron from 'node-cron';
+import type { ScheduledTask } from 'node-cron';
+import pool from '../config/database.js';
 import { tenantQuotaService } from '../services/tenantQuotaService.js';
 import { auditIntegrityService } from '../services/auditIntegrityService.js';
 import logger from '../utils/logger.js';
 
+// Advisory lock IDs — arbitrary but stable; chosen to avoid collisions
+// with any other pg_advisory_lock usage in the application.
+const ADVISORY_LOCK_USAGE_SNAPSHOT = 84_901_001;
+const ADVISORY_LOCK_INTEGRITY_CHECK = 84_901_002;
+
 /**
- * Persist daily usage snapshots for every active organisation.
- * Designed to run once per day at midnight UTC via setInterval or a cron library.
+ * Daily-part-49 jobs — leader-elected, cron-scheduled.
+ *
+ * scheduleDailyUsageSnapshots() and scheduleNightlyIntegrityCheck() each:
+ *   1. Run once immediately on startup (catches missed windows after deploys).
+ *   2. Schedule a node-cron job at midnight UTC (0 0 * * *).
+ *   3. On each cron tick, attempt a Postgres advisory lock so only one pod
+ *      in the replica set actually executes the job.
  *
  * Usage in index.ts / server bootstrap:
- *   scheduleDailyUsageSnapshots();
- *   scheduleNightlyIntegrityCheck();
+ *   const usageSnapshots = scheduleDailyUsageSnapshots();
+ *   const integrityCheck = scheduleNightlyIntegrityCheck();
+ *   // on shutdown: usageSnapshots.stop(); integrityCheck.stop();
  */
-export function scheduleDailyUsageSnapshots(): NodeJS.Timeout {
-  const TWENTY_FOUR_HOURS = 24 * 60 * 60 * 1000;
 
-  // Run once immediately on startup (catches any orgs that missed yesterday's snapshot)
+export interface ScheduledJob {
+  stop(): void;
+}
+
+export function scheduleDailyUsageSnapshots(): ScheduledJob {
+  // Run once immediately on startup (non-leader-elected — safe idempotent catch-up)
   void runDailyUsageSnapshots();
 
-  return setInterval(() => void runDailyUsageSnapshots(), TWENTY_FOUR_HOURS);
-}
+  const task: ScheduledTask = cron.schedule('0 0 * * *', () => {
+    void runWithAdvisoryLock(ADVISORY_LOCK_USAGE_SNAPSHOT, 'daily-usage-snapshot', runDailyUsageSnapshots);
+  }, { timezone: 'UTC' });
 
-export function scheduleNightlyIntegrityCheck(): NodeJS.Timeout {
-  const TWENTY_FOUR_HOURS = 24 * 60 * 60 * 1000;
+  logger.info('Daily usage snapshot cron scheduled (midnight UTC, leader-elected)');
 
+  return {
+    stop() {
+      task.stop();
+      logger.info('Daily usage snapshot cron stopped');
+    },
+  };
+}
+
+export function scheduleNightlyIntegrityCheck(): ScheduledJob {
   void runNightlyIntegrityCheck();
 
-  return setInterval(() => void runNightlyIntegrityCheck(), TWENTY_FOUR_HOURS);
+  const task: ScheduledTask = cron.schedule('0 0 * * *', () => {
+    void runWithAdvisoryLock(ADVISORY_LOCK_INTEGRITY_CHECK, 'nightly-integrity-check', runNightlyIntegrityCheck);
+  }, { timezone: 'UTC' });
+
+  logger.info('Nightly integrity check cron scheduled (midnight UTC, leader-elected)');
+
+  return {
+    stop() {
+      task.stop();
+      logger.info('Nightly integrity check cron stopped');
+    },
+  };
+}
+
+/**
+ * Try to acquire a Postgres advisory lock. If this pod wins the lock,
+ * execute the job and release; otherwise skip silently.
+ *
+ * pg_try_advisory_lock is non-blocking and session-scoped: it releases
+ * automatically when the client connection returns to the pool.
+ */
+async function runWithAdvisoryLock(
+  lockId: number,
+  jobName: string,
+  job: () => Promise<void>,
+): Promise<void> {
+  const client = await pool.connect();
+  try {
+    const { rows } = await client.query<{ acquired: boolean }>(
+      'SELECT pg_try_advisory_lock($1) AS acquired',
+      [lockId],
+    );
+
+    if (!rows[0].acquired) {
+      logger.debug(`[${jobName}] Lock held by another pod — skipping`);
+      return;
+    }
+
+    logger.info(`[${jobName}] Acquired advisory lock — executing`);
+    await job();
+  } catch (err) {
+    logger.error(`[${jobName}] Error during leader-elected execution`, { err });
+  } finally {
+    client.release();
+  }
 }
 
 async function runDailyUsageSnapshots(): Promise<void> {