lorentz
b199d9991c
- extend sync_type union with 'phishing-sweep' - add defaultSchedules entry (disabled by default, daily 5am cron) - dispatch branch dynamically imports and calls sweepPhishingTickets - migrations/098_phishing_sweep_schedule.sql seeds the row for existing installs
817 lines
28 KiB
TypeScript
817 lines
28 KiB
TypeScript
/**
|
||
* Sync Scheduler Service
|
||
* Manages scheduled automatic syncs using node-cron
|
||
*/
|
||
|
||
import cron, { ScheduledTask } from 'node-cron';
|
||
import { SyncService, createSyncService } from './sync-service';
|
||
import { postgresClient } from './postgres-client';
|
||
import { AutotaskClient } from './autotask-client';
|
||
import { EntityType, SyncType } from '../types/sync';
|
||
import { VeeamSyncService } from './veeam-sync-service';
|
||
import { VeeamRpoService } from './veeam-rpo-service';
|
||
import { EngagementSyncService } from './engagement-sync-service';
|
||
import { isMsgraphConfigured } from './msgraph-factory';
|
||
import { ZoomSyncService } from './zoom-sync-service';
|
||
import { isZoomConfigured } from './zoom-factory';
|
||
import { MorningSummaryService } from './morning-summary-service';
|
||
import { TicketDigestService } from './ticket-digest-service';
|
||
|
||
export interface ScheduleConfig {
|
||
id: string;
|
||
name: string;
|
||
description: string;
|
||
cron_expression: string;
|
||
sync_type: 'incremental' | 'full' | 'veeam-incremental' | 'veeam-full' | 'veeam-rpo-check' | 'contract-services' | 'engagement-daily' | 'zoom-daily' | 'morning-summary' | 'ticket-digest-daily' | 'ticket-digest-weekly' | 'ticket-digest-monthly' | 'device-link-reconcile' | 'integration-health' | 'qbo' | 'appgate-sessions' | 'appgate-daily' | 'tickets-reconcile' | 'pax8-daily' | 'phishing-sweep';
|
||
years_back?: number;
|
||
is_enabled: boolean;
|
||
last_run?: Date;
|
||
next_run?: Date;
|
||
last_status?: 'success' | 'failed';
|
||
last_error?: string;
|
||
created_at: Date;
|
||
updated_at: Date;
|
||
}
|
||
|
||
export interface ScheduleStatus {
|
||
config: ScheduleConfig;
|
||
isRunning: boolean;
|
||
isValid: boolean;
|
||
nextRun?: Date;
|
||
}
|
||
|
||
class SyncScheduler {
|
||
private tasks: Map<string, ScheduledTask> = new Map();
|
||
private runningJobs: Set<string> = new Set();
|
||
private initialized = false;
|
||
private syncService: SyncService;
|
||
private _veeamSyncService: VeeamSyncService | null = null;
|
||
private _veeamRpoService: VeeamRpoService | null = null;
|
||
private _engagementSyncService: EngagementSyncService | null = null;
|
||
private _zoomSyncService: ZoomSyncService | null = null;
|
||
|
||
private getVeeamSyncService(): VeeamSyncService {
|
||
if (!this._veeamSyncService) {
|
||
this._veeamSyncService = new VeeamSyncService();
|
||
}
|
||
return this._veeamSyncService;
|
||
}
|
||
|
||
private getVeeamRpoService(): VeeamRpoService {
|
||
if (!this._veeamRpoService) {
|
||
this._veeamRpoService = new VeeamRpoService();
|
||
}
|
||
return this._veeamRpoService;
|
||
}
|
||
|
||
private getEngagementSyncService(): EngagementSyncService {
|
||
if (!this._engagementSyncService) {
|
||
this._engagementSyncService = new EngagementSyncService();
|
||
}
|
||
return this._engagementSyncService;
|
||
}
|
||
|
||
private _morningSummaryService?: MorningSummaryService;
|
||
private getMorningSummaryService(): MorningSummaryService {
|
||
if (!this._morningSummaryService) {
|
||
this._morningSummaryService = new MorningSummaryService();
|
||
}
|
||
return this._morningSummaryService;
|
||
}
|
||
|
||
private _ticketDigestService?: TicketDigestService;
|
||
private getTicketDigestService(): TicketDigestService {
|
||
if (!this._ticketDigestService) {
|
||
this._ticketDigestService = new TicketDigestService();
|
||
}
|
||
return this._ticketDigestService;
|
||
}
|
||
|
||
private getZoomSyncService(): ZoomSyncService {
|
||
if (!this._zoomSyncService) {
|
||
this._zoomSyncService = new ZoomSyncService();
|
||
}
|
||
return this._zoomSyncService;
|
||
}
|
||
|
||
constructor() {
|
||
// Create sync service instance
|
||
const autotaskClient = new AutotaskClient({
|
||
apiUrl: process.env.AUTOTASK_API_URL || '',
|
||
username: process.env.AUTOTASK_USERNAME || '',
|
||
password: process.env.AUTOTASK_SECRET || '',
|
||
apiIntegrationCode: process.env.AUTOTASK_API_INTEGRATION_CODE || '',
|
||
});
|
||
this.syncService = createSyncService(autotaskClient);
|
||
}
|
||
|
||
/**
|
||
* Initialize the scheduler and load schedules from database
|
||
*/
|
||
async initialize(): Promise<void> {
|
||
if (this.initialized) {
|
||
console.log('[SCHEDULER] Already initialized');
|
||
return;
|
||
}
|
||
|
||
console.log('[SCHEDULER] Initializing sync scheduler...');
|
||
|
||
try {
|
||
// Create schedules table if it doesn't exist
|
||
await this.createSchedulesTable();
|
||
|
||
// Create default schedules if none exist
|
||
await this.createDefaultSchedules();
|
||
|
||
// Load and start all enabled schedules
|
||
await this.loadSchedules();
|
||
|
||
this.initialized = true;
|
||
console.log('[SCHEDULER] Sync scheduler initialized successfully');
|
||
} catch (error) {
|
||
console.error('[SCHEDULER] Failed to initialize:', error);
|
||
throw error;
|
||
}
|
||
}
|
||
|
||
/**
|
||
* Create schedules table
|
||
*/
|
||
private async createSchedulesTable(): Promise<void> {
|
||
const query = `
|
||
CREATE TABLE IF NOT EXISTS sync_schedules (
|
||
id VARCHAR(50) PRIMARY KEY,
|
||
name VARCHAR(100) NOT NULL,
|
||
description TEXT,
|
||
cron_expression VARCHAR(50) NOT NULL,
|
||
sync_type VARCHAR(30) NOT NULL,
|
||
years_back INTEGER DEFAULT 2,
|
||
is_enabled BOOLEAN NOT NULL DEFAULT true,
|
||
last_run TIMESTAMP,
|
||
next_run TIMESTAMP,
|
||
last_status VARCHAR(20),
|
||
last_error TEXT,
|
||
created_at TIMESTAMP NOT NULL DEFAULT NOW(),
|
||
updated_at TIMESTAMP NOT NULL DEFAULT NOW()
|
||
);
|
||
|
||
CREATE INDEX IF NOT EXISTS idx_sync_schedules_enabled ON sync_schedules(is_enabled);
|
||
CREATE INDEX IF NOT EXISTS idx_sync_schedules_next_run ON sync_schedules(next_run);
|
||
`;
|
||
|
||
await postgresClient.query(query);
|
||
}
|
||
|
||
/**
|
||
* Create default schedules if none exist
|
||
*/
|
||
private async createDefaultSchedules(): Promise<void> {
|
||
const countResult = await postgresClient.query(
|
||
'SELECT COUNT(*) as count FROM sync_schedules'
|
||
);
|
||
|
||
if (parseInt(countResult.rows[0].count) > 0) {
|
||
console.log('[SCHEDULER] Schedules already exist, skipping defaults');
|
||
return;
|
||
}
|
||
|
||
console.log('[SCHEDULER] Creating default schedules...');
|
||
|
||
const defaultSchedules = [
|
||
{
|
||
id: 'daily-incremental',
|
||
name: 'Daily Incremental Sync',
|
||
description: 'Syncs changes from the last 24 hours every day at 2 AM',
|
||
cron_expression: '0 2 * * *',
|
||
sync_type: 'incremental',
|
||
is_enabled: false,
|
||
},
|
||
{
|
||
id: 'weekly-full',
|
||
name: 'Weekly Full Sync',
|
||
description: 'Full sync of all data every Sunday at 3 AM',
|
||
cron_expression: '0 3 * * 0',
|
||
sync_type: 'full',
|
||
years_back: 2,
|
||
is_enabled: false,
|
||
},
|
||
{
|
||
id: 'veeam-incremental',
|
||
name: 'Veeam Incremental Sync',
|
||
description: 'Syncs Veeam backup data every 30 minutes',
|
||
cron_expression: '*/30 * * * *',
|
||
sync_type: 'veeam-incremental',
|
||
is_enabled: false,
|
||
},
|
||
{
|
||
id: 'veeam-full',
|
||
name: 'Veeam Full Sync',
|
||
description: 'Full Veeam backup data sync daily at 2:00 AM',
|
||
cron_expression: '0 2 * * *',
|
||
sync_type: 'veeam-full',
|
||
is_enabled: false,
|
||
},
|
||
{
|
||
id: 'veeam-rpo-check',
|
||
name: 'Veeam RPO Check',
|
||
description: 'RPO-based workstation backup alerting — creates/resolves Autotask tickets every 30 minutes',
|
||
cron_expression: '*/30 * * * *',
|
||
sync_type: 'veeam-rpo-check',
|
||
is_enabled: false,
|
||
},
|
||
{
|
||
id: 'contract-services',
|
||
name: 'Contract Services Sync',
|
||
description: 'Syncs Autotask contract service lines (service catalog items per contract) daily at 4 AM',
|
||
cron_expression: '0 4 * * *',
|
||
sync_type: 'contract-services',
|
||
is_enabled: true,
|
||
},
|
||
{
|
||
id: 'engagement-daily',
|
||
name: 'Engagement Daily Sync',
|
||
description: 'Syncs Microsoft Graph Teams and email activity for employee engagement dashboard daily at 6 AM',
|
||
cron_expression: '0 6 * * *',
|
||
sync_type: 'engagement-daily',
|
||
is_enabled: false,
|
||
},
|
||
{
|
||
id: 'zoom-daily',
|
||
name: 'Zoom Daily Sync',
|
||
description: 'Syncs Zoom Phone call logs and meeting data daily at 6 AM',
|
||
cron_expression: '0 6 * * *',
|
||
sync_type: 'zoom-daily',
|
||
is_enabled: false,
|
||
},
|
||
{
|
||
id: 'morning-summary',
|
||
name: 'Morning NOC Summary',
|
||
description: 'Posts a Zabbix overnight summary Adaptive Card to configured Teams channel webhooks at 6:30 AM Mon–Fri',
|
||
cron_expression: '30 6 * * 1-5',
|
||
sync_type: 'morning-summary',
|
||
is_enabled: false,
|
||
},
|
||
{
|
||
id: 'ticket-digest-daily',
|
||
name: 'Daily Ticket Digest',
|
||
description: 'LLM-analyzed ticket digest for the previous day, delivered to Teams at 7 AM Mon–Fri',
|
||
cron_expression: '0 7 * * 1-5',
|
||
sync_type: 'ticket-digest-daily',
|
||
is_enabled: false,
|
||
},
|
||
{
|
||
id: 'ticket-digest-weekly',
|
||
name: 'Weekly Ticket Digest',
|
||
description: 'LLM-analyzed ticket digest for the previous week, delivered to Teams at 7 AM Monday',
|
||
cron_expression: '0 7 * * 1',
|
||
sync_type: 'ticket-digest-weekly',
|
||
is_enabled: false,
|
||
},
|
||
{
|
||
id: 'ticket-digest-monthly',
|
||
name: 'Monthly Ticket Digest',
|
||
description: 'LLM-analyzed ticket digest for the previous month, delivered to Teams at 7 AM on the 1st',
|
||
cron_expression: '0 7 1 * *',
|
||
sync_type: 'ticket-digest-monthly',
|
||
is_enabled: false,
|
||
},
|
||
{
|
||
id: 'device-link-reconcile',
|
||
name: 'Device-Link Reconciliation',
|
||
description: 'Walks unlinked device_external_ids rows and links them to Autotask configuration_items via cascading match (serial → MAC → hostname-in-company). Conflicts are queued under /admin/device-link-conflicts. Hourly.',
|
||
cron_expression: '15 * * * *',
|
||
sync_type: 'device-link-reconcile',
|
||
is_enabled: false,
|
||
},
|
||
{
|
||
id: 'integration-health',
|
||
name: 'Integration Health Check',
|
||
description: 'Daily live auth check + token-expiry decode for each configured integration (S1, Datto RMM, IT Glue, Autotask, …). Posts an Adaptive Card to morning-summary webhooks when an integration is failing or a token expires within 14 days. Quiet on green days.',
|
||
cron_expression: '30 7 * * *',
|
||
sync_type: 'integration-health',
|
||
is_enabled: false,
|
||
},
|
||
{
|
||
id: 'tickets-reconcile',
|
||
name: 'Tickets Reconciliation',
|
||
description: 'STOPGAP backstop: re-fetches stale-open tickets (>7d since last sync) from Autotask and reconciles status / soft-deletes missing rows. Runs daily at 4:30 AM. Capped at 500 tickets per run.',
|
||
cron_expression: '30 4 * * *',
|
||
sync_type: 'tickets-reconcile',
|
||
is_enabled: false,
|
||
},
|
||
{
|
||
id: 'phishing-sweep',
|
||
name: 'Phishing Detection Sweep',
|
||
description: 'Reconciles recently-modified tickets through the phishing detector to catch reports missed by the webhook path (bounded to 500 tickets, idempotent on content hash).',
|
||
cron_expression: '0 5 * * *',
|
||
sync_type: 'phishing-sweep',
|
||
is_enabled: false,
|
||
},
|
||
];
|
||
|
||
for (const schedule of defaultSchedules) {
|
||
await postgresClient.query(
|
||
`INSERT INTO sync_schedules (id, name, description, cron_expression, sync_type, years_back, is_enabled)
|
||
VALUES ($1, $2, $3, $4, $5, $6, $7)
|
||
ON CONFLICT (id) DO NOTHING`,
|
||
[
|
||
schedule.id,
|
||
schedule.name,
|
||
schedule.description,
|
||
schedule.cron_expression,
|
||
schedule.sync_type,
|
||
schedule.years_back || null,
|
||
schedule.is_enabled,
|
||
]
|
||
);
|
||
}
|
||
|
||
console.log('[SCHEDULER] Default schedules created');
|
||
}
|
||
|
||
/**
|
||
* Load all schedules from database and start enabled ones
|
||
*/
|
||
async loadSchedules(): Promise<void> {
|
||
console.log('[SCHEDULER] Loading schedules from database...');
|
||
|
||
const result = await postgresClient.query(
|
||
'SELECT * FROM sync_schedules ORDER BY id'
|
||
);
|
||
|
||
const schedules = result.rows as ScheduleConfig[];
|
||
|
||
for (const schedule of schedules) {
|
||
if (schedule.is_enabled) {
|
||
this.startSchedule(schedule);
|
||
}
|
||
}
|
||
|
||
console.log(`[SCHEDULER] Loaded ${schedules.length} schedules (${this.tasks.size} active)`);
|
||
}
|
||
|
||
/**
|
||
* Start a schedule
|
||
*/
|
||
private startSchedule(config: ScheduleConfig): void {
|
||
// Stop existing task if any
|
||
this.stopSchedule(config.id);
|
||
|
||
// Validate cron expression
|
||
if (!cron.validate(config.cron_expression)) {
|
||
console.error(`[SCHEDULER] Invalid cron expression for ${config.id}: ${config.cron_expression}`);
|
||
return;
|
||
}
|
||
|
||
console.log(`[SCHEDULER] Starting schedule: ${config.name} (${config.cron_expression})`);
|
||
|
||
const task = cron.schedule(config.cron_expression, async () => {
|
||
await this.executeScheduledSync(config);
|
||
});
|
||
|
||
this.tasks.set(config.id, task);
|
||
|
||
// Calculate and update next run time
|
||
this.updateNextRunTime(config.id, config.cron_expression);
|
||
}
|
||
|
||
/**
|
||
* Stop a schedule
|
||
*/
|
||
private stopSchedule(scheduleId: string): void {
|
||
const task = this.tasks.get(scheduleId);
|
||
if (task) {
|
||
task.stop();
|
||
this.tasks.delete(scheduleId);
|
||
console.log(`[SCHEDULER] Stopped schedule: ${scheduleId}`);
|
||
}
|
||
}
|
||
|
||
/**
|
||
* Execute a scheduled sync
|
||
*/
|
||
private async executeScheduledSync(config: ScheduleConfig): Promise<void> {
|
||
// Prevent concurrent runs of the same schedule
|
||
if (this.runningJobs.has(config.id)) {
|
||
console.log(`[SCHEDULER] Schedule ${config.id} is already running, skipping`);
|
||
return;
|
||
}
|
||
|
||
this.runningJobs.add(config.id);
|
||
const startTime = new Date();
|
||
|
||
console.log(`[SCHEDULER] Executing scheduled sync: ${config.name}`);
|
||
|
||
try {
|
||
// Update last_run timestamp
|
||
await postgresClient.query(
|
||
'UPDATE sync_schedules SET last_run = NOW() WHERE id = $1',
|
||
[config.id]
|
||
);
|
||
|
||
// Execute the sync
|
||
if (config.sync_type === 'veeam-incremental') {
|
||
await this.getVeeamSyncService().incrementalSync('scheduled');
|
||
} else if (config.sync_type === 'veeam-full') {
|
||
await this.getVeeamSyncService().fullSync('scheduled');
|
||
} else if (config.sync_type === 'veeam-rpo-check') {
|
||
await this.getVeeamRpoService().runCheck();
|
||
} else if (config.sync_type === 'contract-services') {
|
||
await this.syncService.syncEntities([EntityType.AUTOTASK_SERVICES, EntityType.CONTRACT_SERVICES], SyncType.ENTITY_SPECIFIC, 'scheduled');
|
||
} else if (config.sync_type === 'engagement-daily') {
|
||
if (isMsgraphConfigured()) {
|
||
await this.getEngagementSyncService().sync();
|
||
} else {
|
||
console.log('[SCHEDULER] Skipping engagement sync — Microsoft Graph not configured');
|
||
}
|
||
} else if (config.sync_type === 'zoom-daily') {
|
||
if (isZoomConfigured()) {
|
||
await this.getZoomSyncService().sync();
|
||
} else {
|
||
console.log('[SCHEDULER] Skipping Zoom sync — Zoom credentials not configured');
|
||
}
|
||
} else if (config.sync_type === 'morning-summary') {
|
||
await this.getMorningSummaryService().run();
|
||
} else if (config.sync_type === 'ticket-digest-daily') {
|
||
await this.getTicketDigestService().run('daily');
|
||
} else if (config.sync_type === 'ticket-digest-weekly') {
|
||
await this.getTicketDigestService().run('weekly');
|
||
} else if (config.sync_type === 'ticket-digest-monthly') {
|
||
await this.getTicketDigestService().run('monthly');
|
||
} else if (config.sync_type === 'device-link-reconcile') {
|
||
const { reconcileUnlinkedDevices } = await import('@/lib/services/device-link-reconciler');
|
||
const result = await reconcileUnlinkedDevices({ limit: 5000 });
|
||
console.log(
|
||
`[SCHEDULER] device-link-reconcile: scanned=${result.scanned} linked=${result.linked} conflicts=${result.conflicts} unmatched=${result.unmatched}`
|
||
);
|
||
} else if (config.sync_type === 'integration-health') {
|
||
const { runIntegrationHealthAlertJob } = await import('@/lib/services/integration-health-alerts');
|
||
const result = await runIntegrationHealthAlertJob();
|
||
console.log(
|
||
`[SCHEDULER] integration-health: failed=${result.summary.failed} expired=${result.summary.expired} expiringSoon=${result.summary.expiringWithin14Days} alertSent=${result.alertSent}`
|
||
);
|
||
} else if (config.sync_type === 'appgate-sessions' || config.sync_type === 'appgate-daily') {
|
||
const { isAppgateConfigured } = await import('@/lib/services/appgate-factory');
|
||
if (!isAppgateConfigured()) {
|
||
console.log(`[SCHEDULER] Skipping ${config.sync_type} — AppGate not configured`);
|
||
} else {
|
||
const { getAppgateSyncService } = await import('@/lib/services/appgate-sync-service');
|
||
const svc = getAppgateSyncService();
|
||
if (config.sync_type === 'appgate-daily') {
|
||
await svc.dailySync('scheduled');
|
||
} else {
|
||
await svc.sessionsSync('scheduled');
|
||
}
|
||
}
|
||
} else if (config.sync_type === 'tickets-reconcile') {
|
||
const { reconcileStaleTickets } = await import('@/lib/services/ticket-reconciliation-service');
|
||
const result = await reconcileStaleTickets();
|
||
console.log(
|
||
`[SCHEDULER] tickets-reconcile: scanned=${result.scanned} updated=${result.updated} flippedComplete=${result.statusFlippedToComplete} softDeleted=${result.softDeleted} errors=${result.errors}`
|
||
);
|
||
} else if (config.sync_type === 'phishing-sweep') {
|
||
const { sweepPhishingTickets } = await import('@/lib/services/phishing-sweep-service');
|
||
const result = await sweepPhishingTickets();
|
||
console.log(
|
||
`[SCHEDULER] phishing-sweep: scanned=${result.scanned} flagged=${result.flagged} skippedUnchanged=${result.skippedUnchanged} errors=${result.errors}`
|
||
);
|
||
} else if (config.sync_type === 'pax8-daily') {
|
||
const { isPax8Configured } = await import('@/lib/services/pax8-factory');
|
||
if (!isPax8Configured()) {
|
||
console.log('[SCHEDULER] Skipping pax8-daily — PAX8 not configured');
|
||
} else {
|
||
const disabledRes = await postgresClient.query<{ disabled: boolean }>(
|
||
`SELECT disabled FROM integration_settings WHERE key = 'pax8'`
|
||
);
|
||
const isDisabled = disabledRes.rows[0]?.disabled === true;
|
||
if (isDisabled) {
|
||
console.log('[SCHEDULER] Skipping pax8-daily — PAX8 disabled via /admin/integrations');
|
||
} else {
|
||
const { getPax8SyncService } = await import('@/lib/services/pax8-sync-service');
|
||
await getPax8SyncService().fullSync('scheduled');
|
||
}
|
||
}
|
||
} else if (config.sync_type === 'incremental') {
|
||
await this.syncService.incrementalSync('scheduled');
|
||
} else {
|
||
await this.syncService.fullSync('scheduled', config.years_back || 2);
|
||
}
|
||
|
||
// Update success status
|
||
await postgresClient.query(
|
||
`UPDATE sync_schedules
|
||
SET last_status = 'success', last_error = NULL, updated_at = NOW()
|
||
WHERE id = $1`,
|
||
[config.id]
|
||
);
|
||
|
||
const duration = Date.now() - startTime.getTime();
|
||
console.log(`[SCHEDULER] Scheduled sync ${config.name} completed successfully in ${duration}ms`);
|
||
} catch (error) {
|
||
const errorMessage = error instanceof Error ? error.message : String(error);
|
||
console.error(`[SCHEDULER] Scheduled sync ${config.name} failed:`, errorMessage);
|
||
|
||
// Update failure status
|
||
await postgresClient.query(
|
||
`UPDATE sync_schedules
|
||
SET last_status = 'failed', last_error = $2, updated_at = NOW()
|
||
WHERE id = $1`,
|
||
[config.id, errorMessage]
|
||
);
|
||
} finally {
|
||
this.runningJobs.delete(config.id);
|
||
|
||
// Update next run time
|
||
this.updateNextRunTime(config.id, config.cron_expression);
|
||
}
|
||
}
|
||
|
||
/**
|
||
* Calculate and update next run time
|
||
*/
|
||
private async updateNextRunTime(scheduleId: string, cronExpression: string): Promise<void> {
|
||
try {
|
||
const nextRun = this.getNextRunTime(cronExpression);
|
||
if (nextRun) {
|
||
await postgresClient.query(
|
||
'UPDATE sync_schedules SET next_run = $2 WHERE id = $1',
|
||
[scheduleId, nextRun]
|
||
);
|
||
}
|
||
} catch (error) {
|
||
console.error(`[SCHEDULER] Failed to update next run time for ${scheduleId}:`, error);
|
||
}
|
||
}
|
||
|
||
/**
|
||
* Get next run time for a cron expression
|
||
*/
|
||
private getNextRunTime(cronExpression: string): Date | null {
|
||
try {
|
||
// Parse cron expression and calculate next run
|
||
// This is a simplified calculation - node-cron doesn't expose this directly
|
||
const parts = cronExpression.split(' ');
|
||
if (parts.length !== 5) return null;
|
||
|
||
const [minute, hour, dayOfMonth, month, dayOfWeek] = parts;
|
||
|
||
const now = new Date();
|
||
const next = new Date(now);
|
||
|
||
// Simple calculation for common patterns
|
||
if (minute !== '*') next.setMinutes(parseInt(minute));
|
||
if (hour !== '*') next.setHours(parseInt(hour));
|
||
|
||
// If time has passed today, move to next occurrence
|
||
if (next <= now) {
|
||
if (dayOfWeek !== '*') {
|
||
// Weekly schedule
|
||
const targetDay = parseInt(dayOfWeek);
|
||
const currentDay = next.getDay();
|
||
const daysToAdd = targetDay >= currentDay ? targetDay - currentDay : 7 - currentDay + targetDay;
|
||
next.setDate(next.getDate() + daysToAdd);
|
||
} else {
|
||
// Daily schedule
|
||
next.setDate(next.getDate() + 1);
|
||
}
|
||
}
|
||
|
||
next.setSeconds(0);
|
||
next.setMilliseconds(0);
|
||
|
||
return next;
|
||
} catch (error) {
|
||
console.error('[SCHEDULER] Error calculating next run time:', error);
|
||
return null;
|
||
}
|
||
}
|
||
|
||
/**
|
||
* Get all schedules
|
||
*/
|
||
async getSchedules(): Promise<ScheduleStatus[]> {
|
||
const result = await postgresClient.query(
|
||
'SELECT * FROM sync_schedules ORDER BY id'
|
||
);
|
||
|
||
const schedules = result.rows as ScheduleConfig[];
|
||
|
||
return schedules.map(config => ({
|
||
config,
|
||
isRunning: this.runningJobs.has(config.id),
|
||
isValid: cron.validate(config.cron_expression),
|
||
nextRun: config.next_run || undefined,
|
||
}));
|
||
}
|
||
|
||
/**
|
||
* Get a specific schedule
|
||
*/
|
||
async getSchedule(scheduleId: string): Promise<ScheduleStatus | null> {
|
||
const result = await postgresClient.query(
|
||
'SELECT * FROM sync_schedules WHERE id = $1',
|
||
[scheduleId]
|
||
);
|
||
|
||
if (result.rows.length === 0) return null;
|
||
|
||
const config = result.rows[0] as ScheduleConfig;
|
||
|
||
return {
|
||
config,
|
||
isRunning: this.runningJobs.has(config.id),
|
||
isValid: cron.validate(config.cron_expression),
|
||
nextRun: config.next_run || undefined,
|
||
};
|
||
}
|
||
|
||
/**
|
||
* Stop every running cron task and re-load schedules from the DB. Use after
|
||
* seeding new schedule rows directly via SQL (the constructor only runs the
|
||
* default-seeding path on a virgin table).
|
||
*/
|
||
async reloadAllSchedules(): Promise<{ active: number; total: number }> {
|
||
console.log('[SCHEDULER] Reloading all schedules from DB...');
|
||
for (const id of Array.from(this.tasks.keys())) {
|
||
this.stopSchedule(id);
|
||
}
|
||
await this.loadSchedules();
|
||
const total = await postgresClient.query<{ count: string }>(
|
||
'SELECT COUNT(*)::text AS count FROM sync_schedules'
|
||
);
|
||
return {
|
||
active: this.tasks.size,
|
||
total: parseInt(total.rows[0]?.count ?? '0', 10),
|
||
};
|
||
}
|
||
|
||
/**
|
||
* Update a schedule
|
||
*/
|
||
async updateSchedule(
|
||
scheduleId: string,
|
||
updates: Partial<Pick<ScheduleConfig, 'name' | 'description' | 'cron_expression' | 'sync_type' | 'years_back' | 'is_enabled'>>
|
||
): Promise<ScheduleConfig> {
|
||
// Validate cron expression if provided
|
||
if (updates.cron_expression && !cron.validate(updates.cron_expression)) {
|
||
throw new Error(`Invalid cron expression: ${updates.cron_expression}`);
|
||
}
|
||
|
||
// Build update query
|
||
const fields: string[] = [];
|
||
const values: any[] = [];
|
||
let paramIndex = 1;
|
||
|
||
if (updates.name !== undefined) {
|
||
fields.push(`name = $${paramIndex++}`);
|
||
values.push(updates.name);
|
||
}
|
||
if (updates.description !== undefined) {
|
||
fields.push(`description = $${paramIndex++}`);
|
||
values.push(updates.description);
|
||
}
|
||
if (updates.cron_expression !== undefined) {
|
||
fields.push(`cron_expression = $${paramIndex++}`);
|
||
values.push(updates.cron_expression);
|
||
}
|
||
if (updates.sync_type !== undefined) {
|
||
fields.push(`sync_type = $${paramIndex++}`);
|
||
values.push(updates.sync_type);
|
||
}
|
||
if (updates.years_back !== undefined) {
|
||
fields.push(`years_back = $${paramIndex++}`);
|
||
values.push(updates.years_back);
|
||
}
|
||
if (updates.is_enabled !== undefined) {
|
||
fields.push(`is_enabled = $${paramIndex++}`);
|
||
values.push(updates.is_enabled);
|
||
}
|
||
|
||
fields.push(`updated_at = NOW()`);
|
||
values.push(scheduleId);
|
||
|
||
const query = `
|
||
UPDATE sync_schedules
|
||
SET ${fields.join(', ')}
|
||
WHERE id = $${paramIndex}
|
||
RETURNING *
|
||
`;
|
||
|
||
const result = await postgresClient.query(query, values);
|
||
const config = result.rows[0] as ScheduleConfig;
|
||
|
||
// Restart the schedule if it's enabled
|
||
if (config.is_enabled) {
|
||
this.startSchedule(config);
|
||
} else {
|
||
this.stopSchedule(scheduleId);
|
||
}
|
||
|
||
return config;
|
||
}
|
||
|
||
/**
|
||
* Create a new schedule
|
||
*/
|
||
async createSchedule(
|
||
schedule: Omit<ScheduleConfig, 'created_at' | 'updated_at' | 'last_run' | 'next_run' | 'last_status' | 'last_error'>
|
||
): Promise<ScheduleConfig> {
|
||
// Validate cron expression
|
||
if (!cron.validate(schedule.cron_expression)) {
|
||
throw new Error(`Invalid cron expression: ${schedule.cron_expression}`);
|
||
}
|
||
|
||
const query = `
|
||
INSERT INTO sync_schedules (id, name, description, cron_expression, sync_type, years_back, is_enabled)
|
||
VALUES ($1, $2, $3, $4, $5, $6, $7)
|
||
RETURNING *
|
||
`;
|
||
|
||
const result = await postgresClient.query(query, [
|
||
schedule.id,
|
||
schedule.name,
|
||
schedule.description,
|
||
schedule.cron_expression,
|
||
schedule.sync_type,
|
||
schedule.years_back || null,
|
||
schedule.is_enabled,
|
||
]);
|
||
|
||
const config = result.rows[0] as ScheduleConfig;
|
||
|
||
// Start the schedule if enabled
|
||
if (config.is_enabled) {
|
||
this.startSchedule(config);
|
||
}
|
||
|
||
return config;
|
||
}
|
||
|
||
/**
|
||
* Delete a schedule
|
||
*/
|
||
async deleteSchedule(scheduleId: string): Promise<void> {
|
||
// Stop the schedule first
|
||
this.stopSchedule(scheduleId);
|
||
|
||
// Delete from database
|
||
await postgresClient.query(
|
||
'DELETE FROM sync_schedules WHERE id = $1',
|
||
[scheduleId]
|
||
);
|
||
|
||
console.log(`[SCHEDULER] Deleted schedule: ${scheduleId}`);
|
||
}
|
||
|
||
/**
|
||
* Manually trigger a schedule
|
||
*/
|
||
async triggerSchedule(scheduleId: string): Promise<void> {
|
||
const result = await postgresClient.query(
|
||
'SELECT * FROM sync_schedules WHERE id = $1',
|
||
[scheduleId]
|
||
);
|
||
|
||
if (result.rows.length === 0) {
|
||
throw new Error(`Schedule not found: ${scheduleId}`);
|
||
}
|
||
|
||
const config = result.rows[0] as ScheduleConfig;
|
||
await this.executeScheduledSync(config);
|
||
}
|
||
|
||
/**
|
||
* Validate a cron expression
|
||
*/
|
||
validateCronExpression(expression: string): boolean {
|
||
return cron.validate(expression);
|
||
}
|
||
|
||
/**
|
||
* Shutdown the scheduler
|
||
*/
|
||
shutdown(): void {
|
||
console.log('[SCHEDULER] Shutting down sync scheduler...');
|
||
|
||
for (const [id, task] of this.tasks.entries()) {
|
||
task.stop();
|
||
console.log(`[SCHEDULER] Stopped schedule: ${id}`);
|
||
}
|
||
|
||
this.tasks.clear();
|
||
this.runningJobs.clear();
|
||
this.initialized = false;
|
||
|
||
console.log('[SCHEDULER] Sync scheduler shut down');
|
||
}
|
||
}
|
||
|
||
// Export singleton instance
|
||
export const syncScheduler = new SyncScheduler();
|
||
|
||
// Initialize on server startup (only in Node.js environment)
|
||
if (typeof window === 'undefined') {
|
||
syncScheduler.initialize().catch(error => {
|
||
console.error('[SCHEDULER] Failed to initialize on startup:', error);
|
||
});
|
||
}
|