Bunch of updates to scheduling

api/src/services/poll-auto-finalize-queue.service.ts

@@ -0,0 +1,132 @@
+import { Queue, Worker, type Job } from 'bullmq';
+import { env } from '../config/env';
+import { prisma } from '../config/database';
+import { logger } from '../utils/logger';
+
+interface PollAutoFinalizeJobData {
+  pollId: string;
+}
+
+class PollAutoFinalizeQueueService {
+  private queue: Queue;
+  private worker: Worker | null = null;
+
+  constructor() {
+    this.queue = new Queue('poll-auto-finalize', {
+      connection: { url: env.REDIS_URL },
+      defaultJobOptions: {
+        attempts: 3,
+        backoff: { type: 'exponential', delay: 5000 },
+        removeOnComplete: { age: 7 * 24 * 60 * 60, count: 500 },
+        removeOnFail: { age: 30 * 24 * 60 * 60 },
+      },
+    });
+  }
+
+  startWorker() {
+    this.worker = new Worker(
+      'poll-auto-finalize',
+      async (job: Job<PollAutoFinalizeJobData>) => {
+        const { pollId } = job.data;
+        logger.info(`Processing poll auto-finalize job ${job.id}`, { pollId });
+
+        // Dynamic import to avoid circular dependency
+        const { meetingPlannerService } = await import(
+          '../modules/meeting-planner/meeting-planner.service'
+        );
+        await meetingPlannerService.processAutoFinalize(pollId);
+      },
+      {
+        connection: { url: env.REDIS_URL },
+        concurrency: 1,
+      }
+    );
+
+    this.worker.on('completed', (job) => {
+      logger.info(`Poll auto-finalize job ${job.id} completed`);
+    });
+
+    this.worker.on('failed', (job, err) => {
+      logger.error(`Poll auto-finalize job ${job?.id} failed: ${err.message}`);
+    });
+
+    logger.info('Poll auto-finalize queue worker started');
+
+    // Startup recovery: process past-due polls and re-schedule future ones
+    this.recoverOnStartup().catch((err) =>
+      logger.error('Poll auto-finalize startup recovery failed', { error: err })
+    );
+  }
+
+  async scheduleJob(pollId: string, deadline: Date): Promise<string | null> {
+    const delay = deadline.getTime() - Date.now();
+    if (delay <= 0) {
+      // Already past deadline — process immediately
+      const job = await this.queue.add(`finalize-${pollId}`, { pollId }, {
+        jobId: `poll-finalize-${pollId}`,
+      });
+      return job.id ?? null;
+    }
+
+    const job = await this.queue.add(`finalize-${pollId}`, { pollId }, {
+      delay,
+      jobId: `poll-finalize-${pollId}`,
+    });
+    logger.info(`Scheduled poll auto-finalize for ${deadline.toISOString()}`, {
+      pollId,
+      jobId: job.id,
+      delayMs: delay,
+    });
+    return job.id ?? null;
+  }
+
+  async cancelJob(pollId: string): Promise<void> {
+    try {
+      const jobs = await this.queue.getJobs(['delayed', 'waiting']);
+      for (const job of jobs) {
+        if (job.data.pollId === pollId) {
+          await job.remove();
+          logger.info(`Cancelled auto-finalize job for poll ${pollId}`, { jobId: job.id });
+        }
+      }
+    } catch (error) {
+      logger.error('Failed to cancel poll auto-finalize job', { error, pollId });
+    }
+  }
+
+  private async recoverOnStartup() {
+    const openPolls = await prisma.schedulingPoll.findMany({
+      where: {
+        autoFinalize: true,
+        status: 'OPEN',
+        votingDeadline: { not: null },
+      },
+      select: { id: true, votingDeadline: true },
+    });
+
+    for (const poll of openPolls) {
+      if (!poll.votingDeadline) continue;
+      const jobId = await this.scheduleJob(poll.id, poll.votingDeadline);
+      if (jobId) {
+        await prisma.schedulingPoll.update({
+          where: { id: poll.id },
+          data: { autoFinalizeJobId: jobId },
+        }).catch(() => {}); // Best-effort
+      }
+    }
+
+    if (openPolls.length > 0) {
+      logger.info(`Recovered ${openPolls.length} poll auto-finalize jobs on startup`);
+    }
+  }
+
+  async close() {
+    if (this.worker) {
+      await this.worker.close();
+    }
+    await this.queue.close();
+    logger.info('Poll auto-finalize queue closed');
+  }
+}
+
+export const pollAutoFinalizeQueueService = new PollAutoFinalizeQueueService();

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

@@ -14,7 +14,8 @@ type ScheduledJobType =
   | 'cleanup-verification-tokens'
   | 'listmonk-full-sync'
   | 'validate-mkdocs-exports'
-  | 'cleanup-docs-collab-states';
+  | 'cleanup-docs-collab-states'
+  | 'purge-expired-participant-needs';
 
 interface ScheduledJobData {
   type: ScheduledJobType;
@@ -33,6 +34,7 @@ const JOB_DEFINITIONS: Array<{ type: ScheduledJobType; every: number; conditiona
   { 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> {
@@ -89,6 +91,11 @@ async function executeJob(type: ScheduledJobType): Promise<void> {
       await docsCollabService.cleanupStaleStates();
       break;
     }
+    case 'purge-expired-participant-needs': {
+      const { participantNeedsService } = await import('../modules/people/participant-needs.service');
+      await participantNeedsService.purgeExpired();
+      break;
+    }
   }
 }
 
@@ -145,7 +152,7 @@ class ScheduledJobsQueueService {
       logger.error(`Scheduled job ${job?.name} failed: ${err.message}`);
     });
 
-    logger.info('Scheduled jobs queue worker started (10 job types)');
+    logger.info('Scheduled jobs queue worker started (11 job types)');
   }
 
   async close() {