changemaker.lite/api/src/services/scheduled-jobs-queue.service.ts

import { Queue, Worker, type Job } from 'bullmq';
import { env } from '../config/env';
import { logger } from '../utils/logger';

const QUEUE_NAME = 'scheduled-jobs';

type ScheduledJobType =
  | 'reengagement-scan'
  | 'social-digest-scan'
  | 'close-abandoned-canvass-sessions'
  | 'close-stale-tracking-sessions'
  | 'cleanup-tracking-data'
  | 'cleanup-docs-analytics'
  | 'cleanup-verification-tokens'
  | 'listmonk-full-sync'
  | 'validate-mkdocs-exports'
  | 'cleanup-docs-collab-states'
  | 'purge-expired-participant-needs';

interface ScheduledJobData {
  type: ScheduledJobType;
}

const HOUR = 60 * 60 * 1000;

const JOB_DEFINITIONS: Array<{ type: ScheduledJobType; every: number; conditional?: boolean }> = [
  { type: 'reengagement-scan', every: 24 * HOUR },
  { type: 'social-digest-scan', every: 24 * HOUR },
  { type: 'close-abandoned-canvass-sessions', every: HOUR },
  { type: 'close-stale-tracking-sessions', every: HOUR },
  { type: 'cleanup-tracking-data', every: 24 * HOUR },
  { type: 'cleanup-docs-analytics', every: 24 * HOUR },
  { type: 'cleanup-verification-tokens', every: HOUR },
  { type: 'listmonk-full-sync', every: 6 * HOUR, conditional: true },
  { type: 'validate-mkdocs-exports', every: 24 * HOUR },
  { type: 'cleanup-docs-collab-states', every: 24 * HOUR },
  { type: 'purge-expired-participant-needs', every: 24 * HOUR },
];

async function executeJob(type: ScheduledJobType): Promise<void> {
  switch (type) {
    case 'reengagement-scan': {
      const { reengagementService } = await import('./reengagement.service');
      await reengagementService.scan();
      break;
    }
    case 'social-digest-scan': {
      const { socialDigestService } = await import('./social-digest.service');
      await socialDigestService.scan();
      break;
    }
    case 'close-abandoned-canvass-sessions': {
      const { canvassService } = await import('../modules/map/canvass/canvass.service');
      await canvassService.closeAbandonedSessions();
      break;
    }
    case 'close-stale-tracking-sessions': {
      const { trackingService } = await import('../modules/map/tracking/tracking.service');
      await trackingService.closeStaleTrackingSessions(120);
      break;
    }
    case 'cleanup-tracking-data': {
      const { trackingService } = await import('../modules/map/tracking/tracking.service');
      await trackingService.cleanupOldData(30);
      break;
    }
    case 'cleanup-docs-analytics': {
      const { docsAnalyticsService } = await import('../modules/docs-analytics/docs-analytics.service');
      await docsAnalyticsService.cleanupOldData(90);
      break;
    }
    case 'cleanup-verification-tokens': {
      const { verificationTokenService } = await import('./verification-token.service');
      const { passwordResetTokenService } = await import('./password-reset-token.service');
      await verificationTokenService.cleanupExpiredTokens();
      await passwordResetTokenService.cleanupExpiredTokens();
      break;
    }
    case 'listmonk-full-sync': {
      const { listmonkSyncService } = await import('./listmonk-sync.service');
      await listmonkSyncService.syncAll();
      break;
    }
    case 'validate-mkdocs-exports': {
      const { pagesService } = await import('../modules/pages/pages.service');
      await pagesService.validateExports();
      break;
    }
    case 'cleanup-docs-collab-states': {
      const { docsCollabService } = await import('../modules/docs/docs-collab.service');
      await docsCollabService.cleanupStaleStates();
      break;
    }
    case 'purge-expired-participant-needs': {
      const { participantNeedsService } = await import('../modules/people/participant-needs.service');
      await participantNeedsService.purgeExpired();
      break;
    }
  }
}

class ScheduledJobsQueueService {
  private queue: Queue;
  private worker: Worker | null = null;

  constructor() {
    this.queue = new Queue(QUEUE_NAME, {
      connection: { url: env.REDIS_URL },
      defaultJobOptions: {
        removeOnComplete: { age: 60 * 60, count: 200 },
        removeOnFail: { age: 24 * 60 * 60 },
      },
    });
  }

  startWorker() {
    // Register repeatable jobs
    for (const def of JOB_DEFINITIONS) {
      // Skip conditional jobs when their feature is disabled
      if (def.type === 'listmonk-full-sync' && env.LISTMONK_SYNC_ENABLED !== 'true') {
        continue;
      }

      this.queue.add(
        def.type,
        { type: def.type } satisfies ScheduledJobData,
        {
          repeat: { every: def.every },
          jobId: `scheduled-${def.type}`,
        }
      );
    }

    this.worker = new Worker(
      QUEUE_NAME,
      async (job: Job<ScheduledJobData>) => {
        const { type } = job.data;
        logger.debug(`Scheduled job starting: ${type}`);
        await executeJob(type);
      },
      {
        connection: { url: env.REDIS_URL },
        concurrency: 2,
      }
    );

    this.worker.on('completed', (job) => {
      logger.debug(`Scheduled job ${job.name} completed`);
    });

    this.worker.on('failed', (job, err) => {
      logger.error(`Scheduled job ${job?.name} failed: ${err.message}`);
    });

    logger.info('Scheduled jobs queue worker started (11 job types)');
  }

  async close() {
    if (this.worker) {
      await this.worker.close();
    }
    await this.queue.close();
    logger.info('Scheduled jobs queue closed');
  }
}

export const scheduledJobsQueueService = new ScheduledJobsQueueService();