import type { AppContainer } from './bootstrap.js'; import { withRetry } from './utils/retry.js'; import { isSkip } from './services/processor.js'; function sleep(ms: number): Promise { return new Promise((resolve) => setTimeout(resolve, ms)); } const PAUSED_LOG_INTERVAL_MS = 60_000; const PAUSE_POLL_MS = 1_000; const ERROR_BACKOFF_MS = 60_000; /** * Runs the bot's main polling loop and the background response checker until * SIGINT/SIGTERM. Resolves once shutdown completes. */ export async function runMainLoop(app: AppContainer): Promise { const { config, logger, browserClient, postsRepo, messagesRepo, redditApi, monitorService, processorService, responseChecker, chatService, rateLimiter, mediaSyncService, db, webServer, } = app; let isShuttingDown = false; let responseCheckInterval: NodeJS.Timeout | null = null; const shutdown = async (): Promise => { if (isShuttingDown) return; isShuttingDown = true; if (responseCheckInterval) clearInterval(responseCheckInterval); mediaSyncService.stop(); logger.info('Shutting down gracefully...'); webServer?.stop(); await browserClient.close(); db.close(); logger.info('Goodbye!'); process.exit(0); }; process.on('SIGINT', shutdown); process.on('SIGTERM', shutdown); logger.info(''); logger.info('Bot is now running!'); logger.info(`Monitoring subreddits: ${config.reddit.subreddits.map((s) => `${s.name}[${s.type}]`).join(', ')}`); logger.info(`Poll interval: ${config.reddit.pollIntervalMs / 1000} seconds`); logger.info(`Min delay between messages: ${config.rateLimit.minDelayBetweenMessagesMs / 1000} seconds`); logger.info(`Max messages per hour: ${config.rateLimit.maxMessagesPerHour}`); logger.info(''); // Background response checker const RESPONSE_CHECK_INTERVAL_MS = Math.max(config.reddit.pollIntervalMs * 5, 60_000); const checkResponses = async (): Promise => { if (isShuttingDown) return; try { logger.debug('Checking for responses to sent messages (Background Thread)...'); const responses = await responseChecker.checkForResponses(); for (const r of responses) { if (!r.gotResponse) continue; const snippet = r.responseSnippet ? `: ${r.responseSnippet.slice(0, 200)}` : ''; webServer?.logActivity( 'response', `Reply from u/${r.recipient}${snippet}`, r.recipient, ); webServer?.pushLog('info', `πŸ’¬ Reply from u/${r.recipient}`); } if (responses.some((r) => r.gotResponse)) { const updatedExamples = responseChecker.getSuccessfulExamples(3); processorService.setSuccessfulExamples(updatedExamples); } } catch (error) { logger.error('Error in background response checker:', error); } }; checkResponses(); responseCheckInterval = setInterval(checkResponses, RESPONSE_CHECK_INTERVAL_MS); let lastPausedLog = 0; let lastCooldownLog = 0; const COOLDOWN_LOG_INTERVAL_MS = 5 * 60_000; while (!isShuttingDown) { try { if (webServer && !webServer.isRunning()) { if (Date.now() - lastPausedLog > PAUSED_LOG_INTERVAL_MS) { logger.info('Bot is paused. Use the web dashboard to start.'); lastPausedLog = Date.now(); } await sleep(PAUSE_POLL_MS); continue; } // If Reddit told us to take a break on creating new chats, don't even // pull new posts β€” they'd just queue up and skip. Wait the cooldown out. const cooldown = chatService.getInvitationCooldown(); if (cooldown) { const minutesLeft = Math.ceil((cooldown.until - Date.now()) / 60_000); if (Date.now() - lastCooldownLog > COOLDOWN_LOG_INTERVAL_MS) { logger.warn(`Chat invitation cooldown active β€” ~${minutesLeft}m left. ${cooldown.reason}`); webServer?.pushLog( 'warn', `⏸️ Chat throttle: ${cooldown.reason} (~${minutesLeft}m remaining)`, ); lastCooldownLog = Date.now(); } await sleep(config.reddit.pollIntervalMs); continue; } lastCooldownLog = 0; const posts = await monitorService.getUnprocessedPosts(); if (posts.length > 0) { logger.info(`Found ${posts.length} unprocessed posts`); webServer?.pushLog('info', `Found ${posts.length} unprocessed posts`); } else { logger.info('No new posts found. Monitoring...'); } for (const post of posts) { if (isShuttingDown) break; try { const subredditEntry = config.reddit.subreddits.find( (s) => s.name.toLowerCase() === post.subreddit.toLowerCase(), ); const isChatSub = subredditEntry?.type === 'chat'; const postImageUrl = isChatSub ? null : redditApi.getImageUrl(post); postsRepo.insert({ id: post.id, subreddit: post.subreddit, author: post.author, title: post.title, imageUrl: postImageUrl, processedAt: Date.now(), }); // Skip OF/seller accounts before spending on AI. if ( config.reddit.filterSellers && (config.reddit.sellerKeywords.length > 0 || config.reddit.sellerDomains.length > 0) ) { const isSeller = await redditApi.isSellerProfile( post.author, config.reddit.sellerKeywords, config.reddit.sellerDomains ); if (isSeller) { logger.info(`Skipping u/${post.author} - seller/OnlyFans profile detected`); webServer?.logActivity( 'skipped', `Skipped post by u/${post.author} (seller/OnlyFans profile)`, post.author, ); webServer?.pushLog('info', `🚫 Skipped u/${post.author} - seller/OnlyFans profile`); postsRepo.setError(post.id, 'Seller profile detected'); continue; } } // Refresh examples weighted toward this post's subreddit. const subredditExamples = responseChecker.getSuccessfulExamples(3, post.subreddit); processorService.setSuccessfulExamples(subredditExamples); const processed = await withRetry( () => processorService.processPost(post), { logger, retries: 2, operationName: 'Post processing' }, ); if (isSkip(processed)) { logger.debug(`Skipping post ${post.id} - ${processed.skipReason}`); webServer?.logActivity( 'skipped', `Skipped post by u/${post.author} (${processed.skipReason})`, post.author, ); webServer?.pushLog('info', `⏭️ Skipped u/${post.author} β€” ${processed.skipReason}`); continue; } postsRepo.updateAnalysis(post.id, processed.imageAnalysis, processed.generatedMessage); webServer?.pushLog( 'info', `${isChatSub ? 'πŸ’¬' : 'πŸ–ΌοΈ'} ${isChatSub ? 'Chat post' : 'Image analysis'} for u/${post.author}: ${processed.imageAnalysis}`, ); if (processed.strategyId) { webServer?.pushLog('info', `🎯 Strategy: ${processed.strategyId}`); } await rateLimiter.throttle( async () => { logger.info(`Sending message to u/${post.author}...`); webServer?.pushLog('info', `Sending message to u/${post.author}...`); const chat = await chatService.startConversation(post.author, processed.generatedMessage); const success = chat !== null; if (success) { postsRepo.markMessageSent(post.id); // Refresh top chats in background so dashboard is up-to-date. chatService .refreshTopChats(5) .catch((err) => logger.error('Failed to trigger chat refresh:', err)); messagesRepo.insert({ postId: post.id, recipient: post.author, subject: post.author, message: processed.generatedMessage, sentAt: Date.now(), strategyId: processed.strategyId, }); const stats = rateLimiter.getStats(); const currentCount = stats.messagesThisHour + 1; logger.info( `Message sent successfully! (${currentCount}/${config.rateLimit.maxMessagesPerHour} this hour)`, ); webServer?.logActivity( 'sent', `Sent message to u/${post.author} (r/${post.subreddit})`, post.author, postImageUrl ?? undefined, ); webServer?.pushLog('info', `βœ… Message sent to u/${post.author} (${currentCount}/${config.rateLimit.maxMessagesPerHour} this hour)`); webServer?.updateStats(currentCount); } else { postsRepo.setError(post.id, 'Failed to send message'); webServer?.logActivity( 'error', `Failed to send to u/${post.author} (r/${post.subreddit})`, post.author, postImageUrl ?? undefined, ); webServer?.pushLog('error', `❌ Failed to send message to u/${post.author}`); } return success; }, (success) => success === true, ); } catch (error) { const errorMessage = error instanceof Error ? error.message : String(error); logger.error(`Error processing post ${post.id}:`, error); postsRepo.setError(post.id, errorMessage); } } if (!isShuttingDown) { logger.debug(`Waiting ${config.reddit.pollIntervalMs / 1000} seconds before next poll...`); await sleep(config.reddit.pollIntervalMs); } } catch (error) { logger.error('Error in main loop:', error); await sleep(ERROR_BACKOFF_MS); } } }