/** * 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'; 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 = new Map(); private runningJobs: Set = 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 { 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 { 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 { 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, }, ]; 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 { 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 { // 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 === '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 { 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 { 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 { 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, }; } /** * Update a schedule */ async updateSchedule( scheduleId: string, updates: Partial> ): Promise { // 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 ): Promise { // 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 { // 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 { 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); }); }