streamflow/backend/jobs/channelHealth.js

const cron = require('node-cron');
const axios = require('axios');
const { db } = require('../database/db');
const logger = require('../utils/logger');

const CHECK_TIMEOUT = 10000; // 10 seconds timeout
const BATCH_SIZE = 10; // Check 10 channels at a time

/**
 * Check if a channel URL is accessible
 */
async function checkChannelHealth(channelId, url) {
  try {
    const response = await axios.get(url, {
      timeout: CHECK_TIMEOUT,
      maxRedirects: 5,
      responseType: 'stream',
      headers: {
        'User-Agent': 'StreamFlow/1.0'
      }
    });

    // Close the stream immediately
    if (response.data && typeof response.data.destroy === 'function') {
      response.data.destroy();
    }

    // Consider 2xx and 3xx as healthy
    const isHealthy = response.status >= 200 && response.status < 400;
    const status = isHealthy ? 'healthy' : 'degraded';

    // Update channel health status
    db.run(
      'UPDATE channels SET health_status = ?, last_checked = CURRENT_TIMESTAMP WHERE id = ?',
      [status, channelId],
      (err) => {
        if (err) {
          logger.error(`Failed to update health status for channel ${channelId}:`, err);
        }
      }
    );

    return { channelId, status, healthy: isHealthy };
  } catch (error) {
    // Mark as dead if request fails
    const status = 'dead';
    
    db.run(
      'UPDATE channels SET health_status = ?, last_checked = CURRENT_TIMESTAMP WHERE id = ?',
      [status, channelId],
      (err) => {
        if (err) {
          logger.error(`Failed to update health status for channel ${channelId}:`, err);
        }
      }
    );

    logger.debug(`Channel ${channelId} health check failed: ${error.message}`);
    return { channelId, status, healthy: false, error: error.message };
  }
}

/**
 * Check all channels in batches
 */
async function checkAllChannels() {
  return new Promise((resolve, reject) => {
    db.all(
      'SELECT id, url FROM channels WHERE is_active = 1',
      [],
      async (err, channels) => {
        if (err) {
          logger.error('Failed to fetch channels for health check:', err);
          reject(err);
          return;
        }

        logger.info(`Starting health check for ${channels.length} channels`);
        const results = {
          total: channels.length,
          healthy: 0,
          degraded: 0,
          dead: 0
        };

        // Process channels in batches
        for (let i = 0; i < channels.length; i += BATCH_SIZE) {
          const batch = channels.slice(i, i + BATCH_SIZE);
          const promises = batch.map(channel => checkChannelHealth(channel.id, channel.url));
          
          try {
            const batchResults = await Promise.all(promises);
            batchResults.forEach(result => {
              if (result.status === 'healthy') results.healthy++;
              else if (result.status === 'degraded') results.degraded++;
              else if (result.status === 'dead') results.dead++;
            });

            // Small delay between batches to avoid overwhelming the server
            await new Promise(resolve => setTimeout(resolve, 1000));
          } catch (error) {
            logger.error('Error in batch health check:', error);
          }
        }

        logger.info('Health check completed:', results);
        resolve(results);
      }
    );
  });
}

// Check channel health every 6 hours
cron.schedule('0 */6 * * *', async () => {
  logger.info('Running scheduled channel health check');
  try {
    await checkAllChannels();
  } catch (error) {
    logger.error('Channel health check failed:', error);
  }
});

// Export for manual triggering
module.exports = {
  checkAllChannels,
  checkChannelHealth
};