feat(backend): add cleanup old jobs cron and worker functionality

Author: Lasim

services/backend/src/cron/index.ts

@@ -2,6 +2,7 @@ import type { FastifyBaseLogger } from 'fastify';
 import type { JobQueueService } from '../services/jobQueueService';
 import { CronManager } from './cronManager';
 import { createMcpClientActivityMetricsCleanupJob } from './jobs/mcpClientActivityMetricsCleanup';
+import { createCleanupOldJobsJob } from './jobs/cleanupOldJobs';
 // import { createExampleCronJob } from './jobs/exampleJob';
 
 /**
@@ -26,6 +27,9 @@ export function initializeCronJobs(
   // MCP client activity metrics cleanup (every 30 minutes)
   cronManager.register(createMcpClientActivityMetricsCleanupJob(jobQueueService));
 
+  // Cleanup old queue jobs (every 6 hours)
+  cronManager.register(createCleanupOldJobsJob(jobQueueService));
+
   // Example cron job - commented out, uncomment to test
   // cronManager.register(createExampleCronJob(jobQueueService));
 

services/backend/src/cron/jobs/cleanupOldJobs.ts

@@ -0,0 +1,15 @@
+import type { CronJob } from '../cronManager';
+import type { JobQueueService } from '../../services/jobQueueService';
+
+export function createCleanupOldJobsJob(jobQueueService: JobQueueService): CronJob {
+  return {
+    name: 'cleanup-old-jobs',
+    schedule: '0 */6 * * *', // Every 6 hours
+    
+    task: async () => {
+      await jobQueueService.createJob('cleanup_old_jobs', {
+        olderThanDays: 14
+      });
+    }
+  };
+}

services/backend/src/workers/cleanupOldJobsWorker.ts

@@ -0,0 +1,87 @@
+import type { AnyDatabase } from '../db';
+import type { FastifyBaseLogger } from 'fastify';
+import type { Worker, WorkerResult } from './types';
+import { queueJobs, queueJobBatches } from '../db/schema.sqlite';
+import { lt, sql } from 'drizzle-orm';
+
+interface CleanupPayload {
+  olderThanDays: number;
+}
+
+export class CleanupOldJobsWorker implements Worker {
+  constructor(
+    private readonly db: AnyDatabase,
+    private readonly logger: FastifyBaseLogger
+  ) {}
+
+  async execute(payload: unknown, jobId: string): Promise<WorkerResult> {
+    if (!this.isValidPayload(payload)) {
+      return {
+        success: false,
+        message: 'Invalid payload format'
+      };
+    }
+
+    const { olderThanDays } = payload as CleanupPayload;
+
+    this.logger.info({ 
+      jobId, 
+      olderThanDays,
+      operation: 'cleanup_old_jobs'
+    }, 'Starting cleanup of old queue jobs');
+
+    try {
+      const cutoffDate = new Date();
+      cutoffDate.setDate(cutoffDate.getDate() - olderThanDays);
+
+      // Delete old jobs
+      const jobsResult = await this.db
+        .delete(queueJobs)
+        .where(lt(queueJobs.created_at, cutoffDate));
+
+      const jobsDeleted = (jobsResult.changes || jobsResult.rowsAffected || 0);
+
+      // Delete orphaned batches (batches with no jobs)
+      const batchesResult = await this.db
+        .delete(queueJobBatches)
+        .where(
+          sql`${queueJobBatches.id} NOT IN (SELECT DISTINCT ${queueJobs.batch_id} FROM ${queueJobs} WHERE ${queueJobs.batch_id} IS NOT NULL)`
+        );
+
+      const batchesDeleted = (batchesResult.changes || batchesResult.rowsAffected || 0);
+
+      this.logger.info({ 
+        jobId,
+        jobsDeleted,
+        batchesDeleted,
+        cutoffDate: cutoffDate.toISOString(),
+        operation: 'cleanup_old_jobs'
+      }, 'Cleanup completed successfully');
+
+      return {
+        success: true,
+        message: `Deleted ${jobsDeleted} jobs and ${batchesDeleted} orphaned batches older than ${olderThanDays} days`,
+        data: {
+          jobsDeleted,
+          batchesDeleted,
+          cutoffDate: cutoffDate.toISOString()
+        }
+      };
+    } catch (error) {
+      this.logger.error({ 
+        jobId, 
+        error,
+        operation: 'cleanup_old_jobs'
+      }, 'Cleanup job failed');
+      
+      throw error;
+    }
+  }
+
+  private isValidPayload(payload: unknown): payload is CleanupPayload {
+    if (typeof payload !== 'object' || payload === null) return false;
+    // eslint-disable-next-line @typescript-eslint/no-explicit-any
+    const p = payload as any;
+    return typeof p.olderThanDays === 'number' && p.olderThanDays > 0;
+  }
+}

services/backend/src/workers/index.ts

@@ -5,6 +5,7 @@ import { McpServerSyncWorker } from './mcpServerSyncWorker';
 import { RegistryCoordinatorWorker } from './registryCoordinatorWorker';
 import { EmailWorker } from './emailWorker';
 import { McpClientActivityMetricsCleanupWorker } from './mcpClientActivityMetricsCleanupWorker';
+import { CleanupOldJobsWorker } from './cleanupOldJobsWorker';
 
 /**
  * Register all workers with the job processor
@@ -56,6 +57,12 @@ export function registerWorkers(
     new McpClientActivityMetricsCleanupWorker(db, logger)
   );
 
+  // Register Cleanup Old Jobs Worker
+  processor.registerWorker(
+    'cleanup_old_jobs',
+    new CleanupOldJobsWorker(db, logger)
+  );
+
   // Log all registered workers dynamically
   const registeredWorkers = processor.getRegisteredWorkerTypes();
   logger.info({