669 lines
26 KiB
TypeScript
669 lines
26 KiB
TypeScript
import type { AppContainer } from './bootstrap.js';
|
|
import { withRetry } from './utils/retry.js';
|
|
import { isSkip } from './services/processor.js';
|
|
import { reportDeliveryResult, reportInvitationCooldown } from './web/event-reporting.js';
|
|
import { SessionBoundRuntime } from './services/session-bound-runtime.js';
|
|
import { RedditRateLimitError } from './reddit/api.js';
|
|
|
|
export interface StoppableSchedule {
|
|
stop(): void;
|
|
}
|
|
|
|
/** The limiter increments after a successful send, so its count is already current. */
|
|
export function currentHourlySendCount(
|
|
limiter: { getStats(): { messagesThisHour: number } },
|
|
): number {
|
|
return limiter.getStats().messagesThisHour;
|
|
}
|
|
|
|
/**
|
|
* Runs a task again only after its previous run has settled. This deliberately
|
|
* replaces setInterval: a slow browser response check must never overlap the
|
|
* next check and contend for the same browser/session resources.
|
|
*/
|
|
export function startNonOverlappingSchedule(
|
|
task: () => Promise<void>,
|
|
intervalMs: number,
|
|
): StoppableSchedule {
|
|
let stopped = false;
|
|
let timer: NodeJS.Timeout | null = null;
|
|
|
|
const run = async (): Promise<void> => {
|
|
if (stopped) return;
|
|
try {
|
|
await task();
|
|
} finally {
|
|
if (!stopped) timer = setTimeout(() => void run(), intervalMs);
|
|
}
|
|
};
|
|
|
|
void run();
|
|
return {
|
|
stop: () => {
|
|
stopped = true;
|
|
if (timer) clearTimeout(timer);
|
|
},
|
|
};
|
|
}
|
|
|
|
function sleep(ms: number, signal?: AbortSignal): Promise<void> {
|
|
if (signal?.aborted) return Promise.reject(new DOMException('Operation aborted', 'AbortError'));
|
|
return new Promise((resolve, reject) => {
|
|
let timer: NodeJS.Timeout;
|
|
const abort = (): void => {
|
|
clearTimeout(timer);
|
|
cleanup();
|
|
reject(new DOMException('Operation aborted', 'AbortError'));
|
|
};
|
|
const cleanup = (): void => signal?.removeEventListener('abort', abort);
|
|
const done = (): void => {
|
|
cleanup();
|
|
resolve();
|
|
};
|
|
timer = setTimeout(done, ms);
|
|
signal?.addEventListener('abort', abort, { once: true });
|
|
});
|
|
}
|
|
|
|
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<void> {
|
|
if (app.accountRuntimeManager) {
|
|
await runConcurrentAccountRuntimes(app);
|
|
return;
|
|
}
|
|
const {
|
|
config,
|
|
logger,
|
|
browserClient,
|
|
accountAuth,
|
|
postsRepo,
|
|
messagesRepo,
|
|
recipientDeduplicator,
|
|
redditApi,
|
|
monitorService,
|
|
processorService,
|
|
responseChecker,
|
|
chatService,
|
|
rateLimiter,
|
|
mediaSyncService,
|
|
db,
|
|
webServer,
|
|
} = app;
|
|
|
|
let isShuttingDown = false;
|
|
const shutdownController = new AbortController();
|
|
let responseCheckSchedule: StoppableSchedule | null = null;
|
|
let sessionRuntime: SessionBoundRuntime | null = null;
|
|
|
|
const shutdown = async (): Promise<void> => {
|
|
if (isShuttingDown) return;
|
|
isShuttingDown = true;
|
|
shutdownController.abort();
|
|
|
|
sessionRuntime?.shutdown();
|
|
|
|
logger.info('Shutting down gracefully...');
|
|
webServer?.stop();
|
|
await browserClient.close();
|
|
db.close();
|
|
logger.info('Goodbye!');
|
|
};
|
|
|
|
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<void> => {
|
|
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);
|
|
}
|
|
};
|
|
|
|
let startupSyncStarted = false;
|
|
sessionRuntime = new SessionBoundRuntime(
|
|
() => {
|
|
responseCheckSchedule = startNonOverlappingSchedule(checkResponses, RESPONSE_CHECK_INTERVAL_MS);
|
|
if (config.media.syncEnabled) mediaSyncService.start();
|
|
if (!startupSyncStarted) {
|
|
startupSyncStarted = true;
|
|
chatService
|
|
.runStartupSync()
|
|
.then(async ({ kind, chats }) => {
|
|
logger.info(`[chat] startup sync (${kind}) complete: ${chats} chats`);
|
|
await chatService.enrichUsernames();
|
|
})
|
|
.catch((err) => logger.error('[chat] startup sync failed:', err));
|
|
}
|
|
},
|
|
() => {
|
|
responseCheckSchedule?.stop();
|
|
responseCheckSchedule = null;
|
|
mediaSyncService.stop();
|
|
},
|
|
);
|
|
|
|
let lastPausedLog = 0;
|
|
let lastCooldownLog = 0;
|
|
let reportedCooldownUntil: number | null = null;
|
|
const COOLDOWN_LOG_INTERVAL_MS = 5 * 60_000;
|
|
|
|
while (!isShuttingDown) {
|
|
try {
|
|
const authenticated = accountAuth.isFullyAuthenticated()
|
|
|| await accountAuth.ensureFullyAuthenticated();
|
|
if (!sessionRuntime.reconcile(authenticated)) {
|
|
if (Date.now() - lastPausedLog > PAUSED_LOG_INTERVAL_MS) {
|
|
logger.info('Reddit is logged out. Waiting for a session dump from the extension.');
|
|
lastPausedLog = Date.now();
|
|
}
|
|
await sleep(PAUSE_POLL_MS, shutdownController.signal);
|
|
continue;
|
|
}
|
|
|
|
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, shutdownController.signal);
|
|
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) {
|
|
if (webServer && reportedCooldownUntil !== cooldown.until) {
|
|
reportInvitationCooldown(webServer.dashboardEvents, cooldown);
|
|
reportedCooldownUntil = cooldown.until;
|
|
}
|
|
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, shutdownController.signal);
|
|
continue;
|
|
}
|
|
if (webServer && reportedCooldownUntil !== null) {
|
|
reportInvitationCooldown(webServer.dashboardEvents, null);
|
|
reportedCooldownUntil = null;
|
|
}
|
|
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);
|
|
const postUrl = redditApi.getPostUrl(post);
|
|
postsRepo.insert({
|
|
id: post.id,
|
|
subreddit: post.subreddit,
|
|
author: post.author,
|
|
title: post.title,
|
|
postUrl,
|
|
imageUrl: postImageUrl,
|
|
postCreatedAt: post.created_utc * 1000,
|
|
processedAt: Date.now(),
|
|
});
|
|
|
|
// 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 () => {
|
|
if (config.browser.dryRun) {
|
|
const reason = 'Dry run: message prepared but not sent';
|
|
logger.info(`${reason} for u/${post.author}`);
|
|
postsRepo.setError(post.id, reason);
|
|
webServer?.logActivity('skipped', `${reason} to u/${post.author}`, post.author);
|
|
webServer?.pushLog('info', `🧪 ${reason} to u/${post.author}`);
|
|
return false;
|
|
}
|
|
if (!recipientDeduplicator.tryReserve(post.author)) {
|
|
const reason = 'User already contacted across accounts';
|
|
logger.info(`Skipping u/${post.author} - ${reason.toLowerCase()}`);
|
|
postsRepo.setError(post.id, reason);
|
|
webServer?.logActivity(
|
|
'skipped',
|
|
`Skipped u/${post.author} (already contacted across accounts)`,
|
|
post.author,
|
|
);
|
|
return false;
|
|
}
|
|
|
|
logger.info(`Sending message to u/${post.author}...`);
|
|
webServer?.pushLog('info', `Sending message to u/${post.author}...`);
|
|
|
|
let chat;
|
|
try {
|
|
chat = await chatService.startConversation(
|
|
post.author,
|
|
processed.generatedMessage,
|
|
post.author_fullname,
|
|
);
|
|
} catch (error) {
|
|
recipientDeduplicator.release(post.author);
|
|
throw error;
|
|
}
|
|
const success = chat !== null;
|
|
|
|
if (chat) {
|
|
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,
|
|
chatRoomId: chat.roomId,
|
|
experimentId: processed.experimentId,
|
|
promptVersion: processed.promptVersion,
|
|
modelId: processed.modelId,
|
|
selectionReason: processed.selectionReason,
|
|
sourceType: processed.sourceType,
|
|
});
|
|
recipientDeduplicator.commit(post.author);
|
|
|
|
const currentCount = currentHourlySendCount(rateLimiter);
|
|
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,
|
|
{ postTitle: post.title, postUrl },
|
|
);
|
|
webServer?.pushLog('info', `✅ Message sent to u/${post.author} (${currentCount}/${config.rateLimit.maxMessagesPerHour} this hour)`);
|
|
webServer?.updateStats(currentCount);
|
|
if (webServer) {
|
|
reportDeliveryResult(webServer.dashboardEvents, {
|
|
success: true,
|
|
messageId: post.id,
|
|
username: post.author,
|
|
conversationId: chat.roomId,
|
|
});
|
|
}
|
|
} else {
|
|
recipientDeduplicator.release(post.author);
|
|
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}`);
|
|
if (webServer) {
|
|
reportDeliveryResult(webServer.dashboardEvents, {
|
|
success: false,
|
|
messageId: post.id,
|
|
username: post.author,
|
|
error: 'Chat service did not create a conversation',
|
|
});
|
|
}
|
|
}
|
|
|
|
return success;
|
|
},
|
|
(success) => success === true,
|
|
shutdownController.signal,
|
|
);
|
|
} catch (error) {
|
|
if (isShuttingDown && error instanceof DOMException && error.name === 'AbortError') break;
|
|
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, shutdownController.signal);
|
|
}
|
|
} catch (error) {
|
|
if (isShuttingDown && error instanceof DOMException && error.name === 'AbortError') break;
|
|
if (error instanceof RedditRateLimitError) {
|
|
await sleep(Math.max(ERROR_BACKOFF_MS, error.retryAfterMs + 1_000), shutdownController.signal);
|
|
continue;
|
|
}
|
|
|
|
const errorMessage = error instanceof Error ? error.message : String(error);
|
|
logger.error(`Error in main loop: ${errorMessage}`);
|
|
await sleep(ERROR_BACKOFF_MS, shutdownController.signal);
|
|
}
|
|
}
|
|
}
|
|
|
|
async function runConcurrentAccountRuntimes(app: AppContainer): Promise<void> {
|
|
const { config, logger, accountRuntimeManager, browserClient, db, webServer, candidateQueueRepo } = app;
|
|
const workerInstance = `candidate-funnel-${process.pid}-${Date.now()}`;
|
|
const leaseMs = 5 * 60_000;
|
|
const maxActiveCandidates = 500;
|
|
const recovered = candidateQueueRepo.recoverAllLeases();
|
|
app.recipientDeduplicator.recoverStaleReservations(Date.now() - 10 * 60_000);
|
|
logger.info(
|
|
`Running ${accountRuntimeManager.list().length} account runtime(s) with durable candidate funnel`
|
|
+ (recovered > 0 ? `; recovered ${recovered} interrupted candidate(s)` : ''),
|
|
);
|
|
await new Promise<void>((resolve) => {
|
|
let shuttingDown = false;
|
|
const controller = new AbortController();
|
|
let nextScannerIndex = 0;
|
|
let nextSubredditIndex = 0;
|
|
|
|
const runningRuntimes = () => accountRuntimeManager.list()
|
|
.filter((snapshot) => snapshot.state === 'running')
|
|
.map((snapshot) => accountRuntimeManager.get(snapshot.accountId))
|
|
.filter((runtime): runtime is NonNullable<typeof runtime> => runtime !== null);
|
|
|
|
const nextScanner = async () => {
|
|
const runtimes = runningRuntimes();
|
|
for (let offset = 0; offset < runtimes.length; offset += 1) {
|
|
const index = (nextScannerIndex + offset) % runtimes.length;
|
|
const runtime = runtimes[index];
|
|
if (await runtime.canScan()) {
|
|
nextScannerIndex = index + 1;
|
|
return runtime;
|
|
}
|
|
}
|
|
return null;
|
|
};
|
|
|
|
const availableRecipients = async () => {
|
|
const runtimes = runningRuntimes();
|
|
return (await Promise.all(runtimes.map(async (runtime) => ({
|
|
runtime,
|
|
available: await runtime.canAcceptPost(),
|
|
})))).filter((candidate) => candidate.available)
|
|
.map((candidate) => candidate.runtime)
|
|
.sort((left, right) => (
|
|
left.resources.rateLimiter.getStats().messagesThisHour
|
|
- right.resources.rateLimiter.getStats().messagesThisHour
|
|
));
|
|
};
|
|
|
|
const discoveryLoop = (async (): Promise<void> => {
|
|
while (!controller.signal.aborted) {
|
|
try {
|
|
if (!(webServer?.isRunning() ?? true)) {
|
|
await sleep(1_000, controller.signal).catch(() => undefined);
|
|
continue;
|
|
}
|
|
if (candidateQueueRepo.countActive() >= maxActiveCandidates) {
|
|
await sleep(5_000, controller.signal).catch(() => undefined);
|
|
continue;
|
|
}
|
|
const enabled = config.reddit.subreddits.filter((subreddit) => subreddit.enabled !== false);
|
|
if (enabled.length === 0) {
|
|
await sleep(5_000, controller.signal).catch(() => undefined);
|
|
continue;
|
|
}
|
|
const scanner = await nextScanner();
|
|
if (!scanner) {
|
|
await sleep(5_000, controller.signal).catch(() => undefined);
|
|
continue;
|
|
}
|
|
const subreddit = enabled[nextSubredditIndex % enabled.length];
|
|
nextSubredditIndex += 1;
|
|
const posts = await scanner.resources.monitorService.fetchCandidates(subreddit);
|
|
const intake = candidateQueueRepo.enqueue(posts, subreddit.type, subreddit.weight ?? 1);
|
|
logger.info(
|
|
`Candidate intake r/${subreddit.name}: ${posts.length} fetched; ${intake.discovered} queued; `
|
|
+ `${intake.duplicatePost} duplicate posts; ${intake.duplicateAuthor} duplicate authors; `
|
|
+ `${intake.previouslyProcessed} processed; ${intake.previouslyContacted} contacted; `
|
|
+ `${candidateQueueRepo.countActive()} active in funnel`,
|
|
);
|
|
} catch (error) {
|
|
if (controller.signal.aborted) return;
|
|
if (error instanceof RedditRateLimitError) {
|
|
await sleep(error.retryAfterMs + 1_000, controller.signal).catch(() => undefined);
|
|
continue;
|
|
}
|
|
logger.error('Candidate intake failed:', error);
|
|
}
|
|
// RedditRequestCoordinator supplies the real 6.5s pacing. This short
|
|
// yield keeps the loop responsive without adding another poll interval.
|
|
await sleep(250, controller.signal).catch(() => undefined);
|
|
}
|
|
})();
|
|
|
|
const screeningLoop = (async (): Promise<void> => {
|
|
while (!controller.signal.aborted) {
|
|
if (!(webServer?.isRunning() ?? true)) {
|
|
await sleep(1_000, controller.signal).catch(() => undefined);
|
|
continue;
|
|
}
|
|
candidateQueueRepo.expireStale(
|
|
Date.now() - config.reddit.recentWindowMinutes * 60_000,
|
|
);
|
|
if (candidateQueueRepo.countReady() >= config.reddit.maxReadyCandidates) {
|
|
await sleep(2_000, controller.signal).catch(() => undefined);
|
|
continue;
|
|
}
|
|
const workerId = `${workerInstance}:screen`;
|
|
const candidate = candidateQueueRepo.claimForScreening(workerId, leaseMs);
|
|
if (!candidate) {
|
|
await sleep(1_000, controller.signal).catch(() => undefined);
|
|
continue;
|
|
}
|
|
const scanner = await nextScanner();
|
|
if (!scanner) {
|
|
candidateQueueRepo.deferScreening(
|
|
candidate.postId,
|
|
workerId,
|
|
'No authenticated scanner is available',
|
|
Date.now() + 60_000,
|
|
);
|
|
continue;
|
|
}
|
|
try {
|
|
const screen = await scanner.resources.monitorService.screenCandidate(
|
|
candidate.post,
|
|
candidate.sourceType,
|
|
);
|
|
if (!screen.eligible) {
|
|
scanner.resources.monitorService.recordRejection(
|
|
candidate.post,
|
|
candidate.sourceType,
|
|
screen.reason,
|
|
);
|
|
candidateQueueRepo.reject(candidate.postId, workerId, screen.reason);
|
|
continue;
|
|
}
|
|
const prepared = await scanner.preparePost(candidate.post, controller.signal);
|
|
if (isSkip(prepared)) {
|
|
candidateQueueRepo.reject(candidate.postId, workerId, prepared.skipReason);
|
|
continue;
|
|
}
|
|
candidateQueueRepo.markReady(candidate.postId, workerId, prepared);
|
|
logger.info(
|
|
`Candidate ready ${candidate.postId} (r/${candidate.subreddit}); `
|
|
+ `${candidateQueueRepo.countReady()} ready to send`,
|
|
);
|
|
} catch (error) {
|
|
if (controller.signal.aborted) return;
|
|
const reason = error instanceof Error ? error.message : String(error);
|
|
const retryAt = error instanceof RedditRateLimitError
|
|
? Date.now() + error.retryAfterMs + 1_000
|
|
: Date.now() + 60_000;
|
|
candidateQueueRepo.deferScreening(candidate.postId, workerId, reason, retryAt);
|
|
logger.warn(`Candidate screening deferred for ${candidate.postId}: ${reason}`);
|
|
}
|
|
await sleep(250, controller.signal).catch(() => undefined);
|
|
}
|
|
})();
|
|
|
|
const dispatchLoop = (async (): Promise<void> => {
|
|
while (!controller.signal.aborted) {
|
|
if (!(webServer?.isRunning() ?? true)) {
|
|
await sleep(1_000, controller.signal).catch(() => undefined);
|
|
continue;
|
|
}
|
|
candidateQueueRepo.promoteDueRetries(config.reddit.maxReadyCandidates);
|
|
candidateQueueRepo.recoverExpiredLeases();
|
|
app.recipientDeduplicator.recoverStaleReservations(Date.now() - 10 * 60_000);
|
|
const recipients = await availableRecipients();
|
|
const recipient = recipients[0];
|
|
if (!recipient) {
|
|
await sleep(1_000, controller.signal).catch(() => undefined);
|
|
continue;
|
|
}
|
|
const workerId = `${workerInstance}:send:${recipient.session.accountId}`;
|
|
const candidate = candidateQueueRepo.claimReady(
|
|
recipient.session.accountId,
|
|
workerId,
|
|
leaseMs,
|
|
);
|
|
if (!candidate?.prepared) {
|
|
await sleep(500, controller.signal).catch(() => undefined);
|
|
continue;
|
|
}
|
|
try {
|
|
const result = await recipient.dispatchPreparedPost(candidate.prepared, controller.signal);
|
|
if (result.status === 'sent') {
|
|
candidateQueueRepo.markSent(candidate.postId, workerId);
|
|
logger.info(
|
|
`Candidate sent ${candidate.postId} (r/${candidate.subreddit}) via u/${recipient.session.redditUsername}; `
|
|
+ `${candidateQueueRepo.countReady()} remain ready`,
|
|
);
|
|
} else if (result.status === 'terminal') {
|
|
candidateQueueRepo.reject(candidate.postId, workerId, result.reason);
|
|
logger.info(`Candidate rejected at delivery ${candidate.postId}: ${result.reason}`);
|
|
} else {
|
|
// The cooldown belongs to the account, not the candidate. Make the
|
|
// item visible quickly so a different healthy account can claim it;
|
|
// canAcceptPost prevents the throttled account from doing so.
|
|
candidateQueueRepo.deferDelivery(
|
|
candidate.postId,
|
|
workerId,
|
|
result.reason,
|
|
Math.min(result.retryAt, Date.now() + 5_000),
|
|
);
|
|
logger.warn(`Candidate delivery deferred ${candidate.postId}: ${result.reason}`);
|
|
}
|
|
} catch (error) {
|
|
if (controller.signal.aborted) return;
|
|
const reason = error instanceof Error ? error.message : String(error);
|
|
candidateQueueRepo.deferDelivery(
|
|
candidate.postId,
|
|
workerId,
|
|
reason,
|
|
Date.now() + 60_000,
|
|
);
|
|
logger.error(`Candidate delivery failed ${candidate.postId}:`, error);
|
|
}
|
|
await sleep(250, controller.signal).catch(() => undefined);
|
|
}
|
|
})();
|
|
|
|
const shutdown = async (): Promise<void> => {
|
|
if (shuttingDown) return;
|
|
shuttingDown = true;
|
|
controller.abort();
|
|
logger.info('Stopping all Reddit account runtimes...');
|
|
await Promise.all([discoveryLoop, screeningLoop, dispatchLoop]);
|
|
await accountRuntimeManager.stopAll();
|
|
webServer?.stop();
|
|
await browserClient.close();
|
|
db.close();
|
|
process.off('SIGINT', onSignal);
|
|
process.off('SIGTERM', onSignal);
|
|
logger.info('Goodbye!');
|
|
resolve();
|
|
};
|
|
const onSignal = (): void => { void shutdown(); };
|
|
process.on('SIGINT', onSignal);
|
|
process.on('SIGTERM', onSignal);
|
|
});
|
|
}
|