wulf-pulse/lib/services/sync-scheduler.ts

/**
 * 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';
  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,
      },
    ];

    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 === '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);
  });
}