workers/scrapeWorker.ts

import { scrapeQueueRepository } from '@core/storage/repositories/scrapeQueue.js';
import { channelRepository } from '@core/storage/repositories/channels.js';
import { userRepository } from '@core/storage/repositories/users.js';
import { logRepository } from '@core/storage/repositories/logs.js';
import { config } from '@config/index.js';
import { channelScraper } from '@modules/channels/scraper.js';
import { groqModerator } from '@modules/channels/ai.js';
import { sleep } from '@core/utils/helpers.js';

async function processScrapeQueue() {
  console.log('🔄 Starting scrape worker...');

  // Initialize repositories
  await Promise.all([
    scrapeQueueRepository.initialize(),
    channelRepository.initialize(),
    userRepository.initialize(),
    logRepository.initialize(),
  ]);

  let running = true;
  let consecutiveEmpty = 0;

  // Handle graceful shutdown
  process.on('SIGINT', () => { running = false; });
  process.on('SIGTERM', () => { running = false; });

  while (running) {
    try {
      const item = await scrapeQueueRepository.getNextPending();

      if (!item) {
        consecutiveEmpty++;
        if (consecutiveEmpty > 10) {
          // Wait longer when queue is empty
          await sleep(config.SCRAPER_INTERVAL_MS * 5);
        } else {
          await sleep(config.SCRAPER_INTERVAL_MS);
        }
        continue;
      }

      consecutiveEmpty = 0;
      console.log(`📦 Processing queue item: ${item.id} (${item.channelUrl})`);

      await processQueueItem(item);

      // Rate limiting: wait between requests
      await sleep(config.SCRAPER_INTERVAL_MS);

    } catch (error) {
      console.error('Scrape worker error:', error);
      await logRepository.log({
        level: 'error',
        message: 'Scrape worker error',
        error: { name: 'Error', message: String(error), stack: String(error) },
      });
      await sleep(5000);
    }
  }

  console.log('🛑 Scrape worker stopped');
}

async function processQueueItem(item: any) {
  const { id, channelUrl, userId, channelId, type } = item;

  try {
    // Update status to processing
    await scrapeQueueRepository.updateStatus(id, 'processing');

    // Step 1: Scrape channel data
    await scrapeQueueRepository.updateStatus(id, 'scraping');
    const scrapedData = await channelScraper.scrape(channelUrl);

    if (!scrapedData) {
      await handleFailure(id, 'Failed to scrape channel data', userId, channelUrl);
      return;
    }

    // Update queue with scraped data
    await scrapeQueueRepository.setScrapedData(id, scrapedData);

    // If update type, update existing channel
    if (type === 'update' && channelId) {
      await channelRepository.updateFromScrapedData(channelId, scrapedData);
      await channelRepository.updateScrapedAt(channelId);
      await scrapeQueueRepository.updateStatus(id, 'completed');
      console.log(`✅ Updated channel: ${channelId}`);
      return;
    }

    // Step 2: Generate verification code (for new channels)
    await scrapeQueueRepository.updateStatus(id, 'validating');
    const verificationCode = item.validationCode || generateVerificationCode();
    
    // Note: In real implementation, the verification code should already be set
    // when the channel was added to queue. Here we just verify it exists.
    if (!item.validationCode) {
      await scrapeQueueRepository.setValidationCode(id, verificationCode);
    }

    // Step 3: AI Moderation
    await scrapeQueueRepository.updateStatus(id, 'ai_check');
    const aiResult = await groqModerator.moderate(scrapedData.description);

    if (!aiResult.safe) {
      await handleFailure(id, 'Content rejected by AI moderation', userId, channelUrl, aiResult.modelUsed);
      return;
    }

    // Step 4: For new channels, verification is manual (owner must add code to description)
    // The worker waits for manual verification - in practice this would be triggered
    // by a separate verification check or webhook
    if (type === 'new') {
      await scrapeQueueRepository.updateStatus(id, 'validating');
      console.log(`⏳ Channel ${id} awaiting verification code: ${verificationCode}`);
      // In production, you'd have a separate verification checker
      // For now, mark as completed to allow manual verification flow
      await scrapeQueueRepository.updateStatus(id, 'completed');
    } else {
      await scrapeQueueRepository.updateStatus(id, 'completed');
    }

    console.log(`✅ Scraping completed for: ${scrapedData.name}`);

  } catch (error) {
    console.error(`Error processing queue item ${id}:`, error);
    await handleFailure(id, String(error), userId, channelUrl);
  }
}

async function handleFailure(queueId: string, error: string, userId: string, channelUrl: string, modelUsed?: string) {
  const item = await scrapeQueueRepository.findById(queueId);
  if (!item) return;

  if (item.attempts + 1 >= item.maxAttempts) {
    await scrapeQueueRepository.updateStatus(queueId, 'failed', { error });
    
    // If it was a new channel, decrement user's channel count
    if (item.type === 'new') {
      await userRepository.decrementChannelCount(userId);
    }

    await logRepository.log({
      level: 'error',
      message: 'Scrape queue item failed permanently',
      context: { queueId, channelUrl, error, attempts: item.attempts + 1, modelUsed },
      userId,
    });
  } else {
    await scrapeQueueRepository.incrementAttempts(queueId, error);
    await logRepository.log({
      level: 'warn',
      message: 'Scrape queue item failed, will retry',
      context: { queueId, channelUrl, error, attempt: item.attempts + 1 },
      userId,
    });
  }
}

function generateVerificationCode(): string {
  const chars = 'ABCDEFGHJKLMNPQRSTUVWXYZ23456789';
  let code = '';
  for (let i = 0; i < 6; i++) {
    code += chars[Math.floor(Math.random() * chars.length)];
  }
  return code;
}

// Start worker
processScrapeQueue().catch(console.error);