Add durable candidate funnel

src/main-loop.ts

@@ -425,32 +425,44 @@ export async function runMainLoop(app: AppContainer): Promise<void> {
 }
 
 async function runConcurrentAccountRuntimes(app: AppContainer): Promise<void> {
-  const { config, logger, accountRuntimeManager, browserClient, db, webServer } = app;
-  logger.info(`Running ${accountRuntimeManager.list().length} account runtime(s) with shared post discovery`);
+  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 readyTarget = 100;
+  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 discoverAndDispatch = async (): Promise<void> => {
-      const runtimes = accountRuntimeManager.list()
+    const runningRuntimes = () => accountRuntimeManager.list()
         .filter((snapshot) => snapshot.state === 'running')
         .map((snapshot) => accountRuntimeManager.get(snapshot.accountId))
         .filter((runtime): runtime is NonNullable<typeof runtime> => runtime !== null);
-      if (runtimes.length === 0) return;
 
-      // Rotate the authenticated browser used to read Reddit, while keeping
-      // selection and delivery centralised. A subreddit is therefore fetched
-      // once per cycle, never once per account.
-      const scanner = runtimes[nextScannerIndex % runtimes.length];
-      nextScannerIndex += 1;
-      if (!(await scanner.canAcceptPost())) return;
+    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 posts = await scanner.resources.monitorService.getUnprocessedPosts();
-      const post = posts[0];
-      if (!post) return;
-
-      const recipients = (await Promise.all(runtimes.map(async (runtime) => ({
+    const availableRecipients = async () => {
+      const runtimes = runningRuntimes();
+      return (await Promise.all(runtimes.map(async (runtime) => ({
         runtime,
         available: await runtime.canAcceptPost(),
       })))).filter((candidate) => candidate.available)
@@ -459,29 +471,179 @@ async function runConcurrentAccountRuntimes(app: AppContainer): Promise<void> {
           left.resources.rateLimiter.getStats().messagesThisHour
           - right.resources.rateLimiter.getStats().messagesThisHour
         ));
-      const recipient = recipients[0];
-      if (!recipient) {
-        logger.debug(`Eligible post ${post.id} found, but no account is currently available`);
-        return;
-      }
-
-      logger.info(`Assigning eligible post ${post.id} (r/${post.subreddit}) to u/${recipient.session.redditUsername}`);
-      await recipient.dispatchPost(post, controller.signal);
     };
 
     const discoveryLoop = (async (): Promise<void> => {
       while (!controller.signal.aborted) {
         try {
-          if (webServer?.isRunning() ?? true) await discoverAndDispatch();
+          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('Shared post discovery failed:', error);
+          logger.error('Candidate intake failed:', error);
         }
-        await sleep(config.reddit.pollIntervalMs, controller.signal).catch(() => undefined);
+        // 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() >= readyTarget) {
+          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();
+        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);
       }
     })();
 
@@ -490,7 +652,7 @@ async function runConcurrentAccountRuntimes(app: AppContainer): Promise<void> {
       shuttingDown = true;
       controller.abort();
       logger.info('Stopping all Reddit account runtimes...');
-      await discoveryLoop;
+      await Promise.all([discoveryLoop, screeningLoop, dispatchLoop]);
       await accountRuntimeManager.stopAll();
       webServer?.stop();
       await browserClient.close();