pn-new-crm/src/lib/queue/workers/maintenance.ts

import { Worker, type Job } from 'bullmq';
import { and, eq, lt } from 'drizzle-orm';

import type { ConnectionOptions } from 'bullmq';
import { db } from '@/lib/db';
import { formSubmissions } from '@/lib/db/schema/documents';
import { logger } from '@/lib/logger';
import { QUEUE_CONFIGS } from '@/lib/queue';

export const maintenanceWorker = new Worker(
  'maintenance',
  async (job: Job) => {
    logger.info({ jobId: job.id, jobName: job.name }, 'Processing maintenance job');
    switch (job.name) {
      case 'currency-refresh': {
        const { refreshRates } = await import('@/lib/services/currency');
        await refreshRates();
        break;
      }
      case 'form-expiry-check': {
        const result = await db
          .update(formSubmissions)
          .set({ status: 'expired' })
          .where(
            and(eq(formSubmissions.status, 'pending'), lt(formSubmissions.expiresAt, new Date())),
          )
          .returning({ id: formSubmissions.id });
        logger.info({ expired: result.length }, 'Form expiry check complete');
        break;
      }
      case 'alerts-evaluate': {
        const { runAlertEngine } = await import('@/lib/services/alert-engine');
        const summary = await runAlertEngine();
        logger.info(summary, 'Alert engine sweep complete');
        break;
      }
      case 'analytics-refresh': {
        const { ports } = await import('@/lib/db/schema/ports');
        const { refreshSnapshotsForPort } = await import('@/lib/services/analytics.service');
        const allPorts = await db.select({ id: ports.id }).from(ports);
        for (const p of allPorts) {
          try {
            await refreshSnapshotsForPort(p.id);
          } catch (err) {
            logger.warn({ portId: p.id, err }, 'Analytics refresh failed for port');
          }
        }
        logger.info({ count: allPorts.length }, 'Analytics snapshot refresh complete');
        break;
      }
      case 'expense-dedup-scan': {
        const { expenseId } = job.data as { expenseId: string };
        if (!expenseId) {
          logger.warn({ jobId: job.id }, 'expense-dedup-scan missing expenseId');
          break;
        }
        const { markBestDuplicate } = await import('@/lib/services/expense-dedup.service');
        const matchedId = await markBestDuplicate(expenseId);
        logger.info({ expenseId, matchedId: matchedId ?? null }, 'expense-dedup-scan complete');
        break;
      }
      default:
        logger.warn({ jobName: job.name }, 'Unknown maintenance job');
    }
  },
  {
    connection: { url: process.env.REDIS_URL! } as ConnectionOptions,
    concurrency: QUEUE_CONFIGS.maintenance.concurrency,
  },
);

maintenanceWorker.on('failed', (job, err) => {
  logger.error({ jobId: job?.id, jobName: job?.name, err }, 'Maintenance job failed');
});