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 } from './sync-service';
import { postgresClient } from './postgres-client';

export interface ScheduleConfig {
  id: string;
  name: string;
  description: string;
  cron_expression: string;
  sync_type: 'incremental' | 'full';
  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;

  /**
   * 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(20) NOT NULL CHECK (sync_type IN ('incremental', 'full')),
        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, // Disabled by default - user must enable
      },
      {
        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, // Disabled by default - user must enable
      },
    ];

    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)`,
        [
          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 === 'incremental') {
        await syncService.startIncrementalSync('scheduled');
      } else {
        await syncService.startFullSync('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,
    };
  }

  /**
   * 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);
  });
}