Files
pn-new-crm/src/lib/queue/scheduler.ts

77 lines
3.0 KiB
TypeScript
Raw Normal View History

import { getQueue, type QueueName } from './index';
import { logger } from '@/lib/logger';
interface RecurringJobDef {
queue: QueueName;
name: string;
pattern: string;
}
/**
* Register all recurring jobs from 11-REALTIME-AND-BACKGROUND-JOBS.md Section 3.2.
* Called once on server startup.
*/
export async function registerRecurringJobs(): Promise<void> {
const recurring: RecurringJobDef[] = [
// Documenso signature fallback poll — primary is webhooks, this is safety net
feat(alerts): rule engine, recurring evaluator, socket fanout PR2 of Phase B. Wires the alert framework end-to-end: - alert-rules.ts: 10 rule evaluators implemented as pure async fns over the existing schema. reservation.no_agreement, interest.stale, document.signer_overdue, berth.under_offer_stalled, expense.duplicate, expense.unscanned, interest.high_value_silent, eoi.unsigned_long, audit.suspicious_login fire against real conditions. document.expiring_soon stays inert until the documents schema gets an expires_at column. audit.suspicious_login also stays inert until the auth layer logs 'login.failed' rows (TODO noted in the rule body). - alert-engine.ts: runAlertEngine() walks every port × every rule and calls reconcileAlertsForPort. Errors per (port, rule) are collected in the summary, not thrown — one bad evaluator can't stop the sweep. - alerts.service.ts: reconcileAlertsForPort now emits 'alert:created' socket events on insert and 'alert:resolved' on auto-resolve; dismissAlert emits 'alert:dismissed'. All scoped to port:{portId} rooms. - socket/events.ts: adds the three Server→Client alert event types. - queue/scheduler.ts: registers 'alerts-evaluate' on the maintenance queue with cron */5 * * * * (every 5 min, per spec risk register). - queue/workers/maintenance.ts: dispatches 'alerts-evaluate' to runAlertEngine; logs sweep summary. Tests: - tests/integration/alerts-engine.test.ts (6 cases): seeds reservation → fires, runs twice → no dupe, adds agreement → auto-resolves; seeds stale interest → fires; hot lead silent → critical; engine summary shape on no-data port. Socket emit module is vi.mocked. Vitest 681/681 (was 675; +6). tsc clean. Lint clean. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-04-28 14:50:55 +02:00
{ queue: 'documents', name: 'signature-poll', pattern: '0 */6 * * *' },
// Reminder checks
feat(alerts): rule engine, recurring evaluator, socket fanout PR2 of Phase B. Wires the alert framework end-to-end: - alert-rules.ts: 10 rule evaluators implemented as pure async fns over the existing schema. reservation.no_agreement, interest.stale, document.signer_overdue, berth.under_offer_stalled, expense.duplicate, expense.unscanned, interest.high_value_silent, eoi.unsigned_long, audit.suspicious_login fire against real conditions. document.expiring_soon stays inert until the documents schema gets an expires_at column. audit.suspicious_login also stays inert until the auth layer logs 'login.failed' rows (TODO noted in the rule body). - alert-engine.ts: runAlertEngine() walks every port × every rule and calls reconcileAlertsForPort. Errors per (port, rule) are collected in the summary, not thrown — one bad evaluator can't stop the sweep. - alerts.service.ts: reconcileAlertsForPort now emits 'alert:created' socket events on insert and 'alert:resolved' on auto-resolve; dismissAlert emits 'alert:dismissed'. All scoped to port:{portId} rooms. - socket/events.ts: adds the three Server→Client alert event types. - queue/scheduler.ts: registers 'alerts-evaluate' on the maintenance queue with cron */5 * * * * (every 5 min, per spec risk register). - queue/workers/maintenance.ts: dispatches 'alerts-evaluate' to runAlertEngine; logs sweep summary. Tests: - tests/integration/alerts-engine.test.ts (6 cases): seeds reservation → fires, runs twice → no dupe, adds agreement → auto-resolves; seeds stale interest → fires; hot lead silent → critical; engine summary shape on no-data port. Socket emit module is vi.mocked. Vitest 681/681 (was 675; +6). tsc clean. Lint clean. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-04-28 14:50:55 +02:00
{ queue: 'notifications', name: 'reminder-check', pattern: '0 * * * *' },
{ queue: 'notifications', name: 'reminder-overdue-check', pattern: '*/15 * * * *' },
// Google Calendar background sync
feat(alerts): rule engine, recurring evaluator, socket fanout PR2 of Phase B. Wires the alert framework end-to-end: - alert-rules.ts: 10 rule evaluators implemented as pure async fns over the existing schema. reservation.no_agreement, interest.stale, document.signer_overdue, berth.under_offer_stalled, expense.duplicate, expense.unscanned, interest.high_value_silent, eoi.unsigned_long, audit.suspicious_login fire against real conditions. document.expiring_soon stays inert until the documents schema gets an expires_at column. audit.suspicious_login also stays inert until the auth layer logs 'login.failed' rows (TODO noted in the rule body). - alert-engine.ts: runAlertEngine() walks every port × every rule and calls reconcileAlertsForPort. Errors per (port, rule) are collected in the summary, not thrown — one bad evaluator can't stop the sweep. - alerts.service.ts: reconcileAlertsForPort now emits 'alert:created' socket events on insert and 'alert:resolved' on auto-resolve; dismissAlert emits 'alert:dismissed'. All scoped to port:{portId} rooms. - socket/events.ts: adds the three Server→Client alert event types. - queue/scheduler.ts: registers 'alerts-evaluate' on the maintenance queue with cron */5 * * * * (every 5 min, per spec risk register). - queue/workers/maintenance.ts: dispatches 'alerts-evaluate' to runAlertEngine; logs sweep summary. Tests: - tests/integration/alerts-engine.test.ts (6 cases): seeds reservation → fires, runs twice → no dupe, adds agreement → auto-resolves; seeds stale interest → fires; hot lead silent → critical; engine summary shape on no-data port. Socket emit module is vi.mocked. Vitest 681/681 (was 675; +6). tsc clean. Lint clean. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-04-28 14:50:55 +02:00
{ queue: 'maintenance', name: 'calendar-sync', pattern: '*/30 * * * *' },
// Daily checks at 08:00
feat(alerts): rule engine, recurring evaluator, socket fanout PR2 of Phase B. Wires the alert framework end-to-end: - alert-rules.ts: 10 rule evaluators implemented as pure async fns over the existing schema. reservation.no_agreement, interest.stale, document.signer_overdue, berth.under_offer_stalled, expense.duplicate, expense.unscanned, interest.high_value_silent, eoi.unsigned_long, audit.suspicious_login fire against real conditions. document.expiring_soon stays inert until the documents schema gets an expires_at column. audit.suspicious_login also stays inert until the auth layer logs 'login.failed' rows (TODO noted in the rule body). - alert-engine.ts: runAlertEngine() walks every port × every rule and calls reconcileAlertsForPort. Errors per (port, rule) are collected in the summary, not thrown — one bad evaluator can't stop the sweep. - alerts.service.ts: reconcileAlertsForPort now emits 'alert:created' socket events on insert and 'alert:resolved' on auto-resolve; dismissAlert emits 'alert:dismissed'. All scoped to port:{portId} rooms. - socket/events.ts: adds the three Server→Client alert event types. - queue/scheduler.ts: registers 'alerts-evaluate' on the maintenance queue with cron */5 * * * * (every 5 min, per spec risk register). - queue/workers/maintenance.ts: dispatches 'alerts-evaluate' to runAlertEngine; logs sweep summary. Tests: - tests/integration/alerts-engine.test.ts (6 cases): seeds reservation → fires, runs twice → no dupe, adds agreement → auto-resolves; seeds stale interest → fires; hot lead silent → critical; engine summary shape on no-data port. Socket emit module is vi.mocked. Vitest 681/681 (was 675; +6). tsc clean. Lint clean. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-04-28 14:50:55 +02:00
{ queue: 'notifications', name: 'invoice-overdue-check', pattern: '0 8 * * *' },
{ queue: 'notifications', name: 'tenure-expiry-check', pattern: '0 8 * * *' },
// Exchange rate refresh every 6 hours
feat(alerts): rule engine, recurring evaluator, socket fanout PR2 of Phase B. Wires the alert framework end-to-end: - alert-rules.ts: 10 rule evaluators implemented as pure async fns over the existing schema. reservation.no_agreement, interest.stale, document.signer_overdue, berth.under_offer_stalled, expense.duplicate, expense.unscanned, interest.high_value_silent, eoi.unsigned_long, audit.suspicious_login fire against real conditions. document.expiring_soon stays inert until the documents schema gets an expires_at column. audit.suspicious_login also stays inert until the auth layer logs 'login.failed' rows (TODO noted in the rule body). - alert-engine.ts: runAlertEngine() walks every port × every rule and calls reconcileAlertsForPort. Errors per (port, rule) are collected in the summary, not thrown — one bad evaluator can't stop the sweep. - alerts.service.ts: reconcileAlertsForPort now emits 'alert:created' socket events on insert and 'alert:resolved' on auto-resolve; dismissAlert emits 'alert:dismissed'. All scoped to port:{portId} rooms. - socket/events.ts: adds the three Server→Client alert event types. - queue/scheduler.ts: registers 'alerts-evaluate' on the maintenance queue with cron */5 * * * * (every 5 min, per spec risk register). - queue/workers/maintenance.ts: dispatches 'alerts-evaluate' to runAlertEngine; logs sweep summary. Tests: - tests/integration/alerts-engine.test.ts (6 cases): seeds reservation → fires, runs twice → no dupe, adds agreement → auto-resolves; seeds stale interest → fires; hot lead silent → critical; engine summary shape on no-data port. Socket emit module is vi.mocked. Vitest 681/681 (was 675; +6). tsc clean. Lint clean. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-04-28 14:50:55 +02:00
{ queue: 'maintenance', name: 'currency-refresh', pattern: '0 */6 * * *' },
// Database backup / cleanup
feat(alerts): rule engine, recurring evaluator, socket fanout PR2 of Phase B. Wires the alert framework end-to-end: - alert-rules.ts: 10 rule evaluators implemented as pure async fns over the existing schema. reservation.no_agreement, interest.stale, document.signer_overdue, berth.under_offer_stalled, expense.duplicate, expense.unscanned, interest.high_value_silent, eoi.unsigned_long, audit.suspicious_login fire against real conditions. document.expiring_soon stays inert until the documents schema gets an expires_at column. audit.suspicious_login also stays inert until the auth layer logs 'login.failed' rows (TODO noted in the rule body). - alert-engine.ts: runAlertEngine() walks every port × every rule and calls reconcileAlertsForPort. Errors per (port, rule) are collected in the summary, not thrown — one bad evaluator can't stop the sweep. - alerts.service.ts: reconcileAlertsForPort now emits 'alert:created' socket events on insert and 'alert:resolved' on auto-resolve; dismissAlert emits 'alert:dismissed'. All scoped to port:{portId} rooms. - socket/events.ts: adds the three Server→Client alert event types. - queue/scheduler.ts: registers 'alerts-evaluate' on the maintenance queue with cron */5 * * * * (every 5 min, per spec risk register). - queue/workers/maintenance.ts: dispatches 'alerts-evaluate' to runAlertEngine; logs sweep summary. Tests: - tests/integration/alerts-engine.test.ts (6 cases): seeds reservation → fires, runs twice → no dupe, adds agreement → auto-resolves; seeds stale interest → fires; hot lead silent → critical; engine summary shape on no-data port. Socket emit module is vi.mocked. Vitest 681/681 (was 675; +6). tsc clean. Lint clean. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-04-28 14:50:55 +02:00
{ queue: 'maintenance', name: 'database-backup', pattern: '0 2 * * *' },
{ queue: 'maintenance', name: 'backup-cleanup', pattern: '0 3 * * 0' }, // Sunday 03:00
// Session cleanup
feat(alerts): rule engine, recurring evaluator, socket fanout PR2 of Phase B. Wires the alert framework end-to-end: - alert-rules.ts: 10 rule evaluators implemented as pure async fns over the existing schema. reservation.no_agreement, interest.stale, document.signer_overdue, berth.under_offer_stalled, expense.duplicate, expense.unscanned, interest.high_value_silent, eoi.unsigned_long, audit.suspicious_login fire against real conditions. document.expiring_soon stays inert until the documents schema gets an expires_at column. audit.suspicious_login also stays inert until the auth layer logs 'login.failed' rows (TODO noted in the rule body). - alert-engine.ts: runAlertEngine() walks every port × every rule and calls reconcileAlertsForPort. Errors per (port, rule) are collected in the summary, not thrown — one bad evaluator can't stop the sweep. - alerts.service.ts: reconcileAlertsForPort now emits 'alert:created' socket events on insert and 'alert:resolved' on auto-resolve; dismissAlert emits 'alert:dismissed'. All scoped to port:{portId} rooms. - socket/events.ts: adds the three Server→Client alert event types. - queue/scheduler.ts: registers 'alerts-evaluate' on the maintenance queue with cron */5 * * * * (every 5 min, per spec risk register). - queue/workers/maintenance.ts: dispatches 'alerts-evaluate' to runAlertEngine; logs sweep summary. Tests: - tests/integration/alerts-engine.test.ts (6 cases): seeds reservation → fires, runs twice → no dupe, adds agreement → auto-resolves; seeds stale interest → fires; hot lead silent → critical; engine summary shape on no-data port. Socket emit module is vi.mocked. Vitest 681/681 (was 675; +6). tsc clean. Lint clean. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-04-28 14:50:55 +02:00
{ queue: 'maintenance', name: 'session-cleanup', pattern: '0 4 * * *' },
// Report scheduler — checks every minute for reports due to run
feat(alerts): rule engine, recurring evaluator, socket fanout PR2 of Phase B. Wires the alert framework end-to-end: - alert-rules.ts: 10 rule evaluators implemented as pure async fns over the existing schema. reservation.no_agreement, interest.stale, document.signer_overdue, berth.under_offer_stalled, expense.duplicate, expense.unscanned, interest.high_value_silent, eoi.unsigned_long, audit.suspicious_login fire against real conditions. document.expiring_soon stays inert until the documents schema gets an expires_at column. audit.suspicious_login also stays inert until the auth layer logs 'login.failed' rows (TODO noted in the rule body). - alert-engine.ts: runAlertEngine() walks every port × every rule and calls reconcileAlertsForPort. Errors per (port, rule) are collected in the summary, not thrown — one bad evaluator can't stop the sweep. - alerts.service.ts: reconcileAlertsForPort now emits 'alert:created' socket events on insert and 'alert:resolved' on auto-resolve; dismissAlert emits 'alert:dismissed'. All scoped to port:{portId} rooms. - socket/events.ts: adds the three Server→Client alert event types. - queue/scheduler.ts: registers 'alerts-evaluate' on the maintenance queue with cron */5 * * * * (every 5 min, per spec risk register). - queue/workers/maintenance.ts: dispatches 'alerts-evaluate' to runAlertEngine; logs sweep summary. Tests: - tests/integration/alerts-engine.test.ts (6 cases): seeds reservation → fires, runs twice → no dupe, adds agreement → auto-resolves; seeds stale interest → fires; hot lead silent → critical; engine summary shape on no-data port. Socket emit module is vi.mocked. Vitest 681/681 (was 675; +6). tsc clean. Lint clean. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-04-28 14:50:55 +02:00
{ queue: 'reports', name: 'report-scheduler', pattern: '* * * * *' },
// Notification digest — configurable per user; placeholder fires hourly
// TODO(L2): make per-user schedule configurable (read from user_settings)
feat(alerts): rule engine, recurring evaluator, socket fanout PR2 of Phase B. Wires the alert framework end-to-end: - alert-rules.ts: 10 rule evaluators implemented as pure async fns over the existing schema. reservation.no_agreement, interest.stale, document.signer_overdue, berth.under_offer_stalled, expense.duplicate, expense.unscanned, interest.high_value_silent, eoi.unsigned_long, audit.suspicious_login fire against real conditions. document.expiring_soon stays inert until the documents schema gets an expires_at column. audit.suspicious_login also stays inert until the auth layer logs 'login.failed' rows (TODO noted in the rule body). - alert-engine.ts: runAlertEngine() walks every port × every rule and calls reconcileAlertsForPort. Errors per (port, rule) are collected in the summary, not thrown — one bad evaluator can't stop the sweep. - alerts.service.ts: reconcileAlertsForPort now emits 'alert:created' socket events on insert and 'alert:resolved' on auto-resolve; dismissAlert emits 'alert:dismissed'. All scoped to port:{portId} rooms. - socket/events.ts: adds the three Server→Client alert event types. - queue/scheduler.ts: registers 'alerts-evaluate' on the maintenance queue with cron */5 * * * * (every 5 min, per spec risk register). - queue/workers/maintenance.ts: dispatches 'alerts-evaluate' to runAlertEngine; logs sweep summary. Tests: - tests/integration/alerts-engine.test.ts (6 cases): seeds reservation → fires, runs twice → no dupe, adds agreement → auto-resolves; seeds stale interest → fires; hot lead silent → critical; engine summary shape on no-data port. Socket emit module is vi.mocked. Vitest 681/681 (was 675; +6). tsc clean. Lint clean. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-04-28 14:50:55 +02:00
{ queue: 'email', name: 'notification-digest', pattern: '0 * * * *' },
// Cleanup jobs
feat(alerts): rule engine, recurring evaluator, socket fanout PR2 of Phase B. Wires the alert framework end-to-end: - alert-rules.ts: 10 rule evaluators implemented as pure async fns over the existing schema. reservation.no_agreement, interest.stale, document.signer_overdue, berth.under_offer_stalled, expense.duplicate, expense.unscanned, interest.high_value_silent, eoi.unsigned_long, audit.suspicious_login fire against real conditions. document.expiring_soon stays inert until the documents schema gets an expires_at column. audit.suspicious_login also stays inert until the auth layer logs 'login.failed' rows (TODO noted in the rule body). - alert-engine.ts: runAlertEngine() walks every port × every rule and calls reconcileAlertsForPort. Errors per (port, rule) are collected in the summary, not thrown — one bad evaluator can't stop the sweep. - alerts.service.ts: reconcileAlertsForPort now emits 'alert:created' socket events on insert and 'alert:resolved' on auto-resolve; dismissAlert emits 'alert:dismissed'. All scoped to port:{portId} rooms. - socket/events.ts: adds the three Server→Client alert event types. - queue/scheduler.ts: registers 'alerts-evaluate' on the maintenance queue with cron */5 * * * * (every 5 min, per spec risk register). - queue/workers/maintenance.ts: dispatches 'alerts-evaluate' to runAlertEngine; logs sweep summary. Tests: - tests/integration/alerts-engine.test.ts (6 cases): seeds reservation → fires, runs twice → no dupe, adds agreement → auto-resolves; seeds stale interest → fires; hot lead silent → critical; engine summary shape on no-data port. Socket emit module is vi.mocked. Vitest 681/681 (was 675; +6). tsc clean. Lint clean. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-04-28 14:50:55 +02:00
{ queue: 'maintenance', name: 'temp-file-cleanup', pattern: '0 5 * * *' },
{ queue: 'maintenance', name: 'form-expiry-check', pattern: '0 * * * *' },
// Phase B: alert rule engine sweep
{ queue: 'maintenance', name: 'alerts-evaluate', pattern: '*/5 * * * *' },
feat(analytics): real computations + 15-min snapshot refresh job PR3 of Phase B. Replaces the no-op stubs in analytics.service.ts with working drizzle queries and adds the recurring BullMQ job that warms the cache. Computations: - computePipelineFunnel: groups interests by pipeline_stage filtered by port + range + not archived; emits 8-row stages array with conversion pct relative to 'open' as the funnel top. - computeOccupancyTimeline: per day in range, counts berths covered by an active reservation (start_date ≤ day, end_date IS NULL OR ≥ day); emits {date, occupied, total, occupancyPct}. - computeRevenueBreakdown: sums invoices.total grouped by status + currency; filters out archived rows. - computeLeadSourceAttribution: counts interests by source descending; null source bucketed as 'unspecified'. Public API (getPipelineFunnel, getOccupancyTimeline, etc.) reads analytics_snapshots first; falls back to compute + writeSnapshot. TTL 15 minutes (matches the cron interval). Cron: - queue/scheduler.ts registers 'analytics-refresh' on maintenance with pattern '*/15 * * * *'. - queue/workers/maintenance.ts dispatches to refreshSnapshotsForPort for every port; per-port try/catch so one bad port doesn't kill the sweep. Tests: tests/integration/analytics-service.test.ts (9 cases). Pipeline funnel math (incl. zero state), occupancy timeline shape/percentages with seeded reservations, revenue grouped by status + currency, lead source attribution incl. null bucketing, cache hit (mutate snapshot directly → next read returns mutated value), refreshSnapshotsForPort warms every metric×range combo. Vitest 690/690 (+9). tsc + lint clean. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-04-28 14:54:46 +02:00
// Phase B: analytics snapshot warm
{ queue: 'maintenance', name: 'analytics-refresh', pattern: '*/15 * * * *' },
// Phase 3d: GDPR Article 17 — actually delete expired export bundles
{ queue: 'maintenance', name: 'gdpr-export-cleanup', pattern: '0 4 * * *' },
// Phase 3b: AI usage ledger retention (90-day rolling window)
{ queue: 'maintenance', name: 'ai-usage-retention', pattern: '0 5 * * *' },
];
for (const job of recurring) {
const queue = getQueue(job.queue);
await queue.upsertJobScheduler(
job.name,
{ pattern: job.pattern },
{ data: {}, name: job.name },
);
feat(alerts): rule engine, recurring evaluator, socket fanout PR2 of Phase B. Wires the alert framework end-to-end: - alert-rules.ts: 10 rule evaluators implemented as pure async fns over the existing schema. reservation.no_agreement, interest.stale, document.signer_overdue, berth.under_offer_stalled, expense.duplicate, expense.unscanned, interest.high_value_silent, eoi.unsigned_long, audit.suspicious_login fire against real conditions. document.expiring_soon stays inert until the documents schema gets an expires_at column. audit.suspicious_login also stays inert until the auth layer logs 'login.failed' rows (TODO noted in the rule body). - alert-engine.ts: runAlertEngine() walks every port × every rule and calls reconcileAlertsForPort. Errors per (port, rule) are collected in the summary, not thrown — one bad evaluator can't stop the sweep. - alerts.service.ts: reconcileAlertsForPort now emits 'alert:created' socket events on insert and 'alert:resolved' on auto-resolve; dismissAlert emits 'alert:dismissed'. All scoped to port:{portId} rooms. - socket/events.ts: adds the three Server→Client alert event types. - queue/scheduler.ts: registers 'alerts-evaluate' on the maintenance queue with cron */5 * * * * (every 5 min, per spec risk register). - queue/workers/maintenance.ts: dispatches 'alerts-evaluate' to runAlertEngine; logs sweep summary. Tests: - tests/integration/alerts-engine.test.ts (6 cases): seeds reservation → fires, runs twice → no dupe, adds agreement → auto-resolves; seeds stale interest → fires; hot lead silent → critical; engine summary shape on no-data port. Socket emit module is vi.mocked. Vitest 681/681 (was 675; +6). tsc clean. Lint clean. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-04-28 14:50:55 +02:00
logger.info(
{ queue: job.queue, job: job.name, pattern: job.pattern },
'Registered recurring job',
);
}
logger.info({ count: recurring.length }, 'All recurring jobs registered');
}