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);