diff --git a/docs/api.md b/docs/api.md index dcad456..5056192 100644 --- a/docs/api.md +++ b/docs/api.md @@ -60,6 +60,7 @@ an administrator credential. | `GET` | `/api/v1/jobs/{id}` | Get asynchronous job state | read | no | | `POST` | `/api/v1/jobs` | Start chat or media synchronisation | control | yes | | `GET` | `/api/stats` | Get bot statistics | read | no | +| `GET` | `/api/analytics/candidate-funnel` | Get durable candidate funnel counts | read | no | | `GET` | `/api/activity` | Get recent activity | read | no | | `GET` | `/api/posts/by-author/{username}` | Get processed posts by author | read | no | | `GET` | `/api/messages` | Get recent sent messages | read | no | diff --git a/src/bootstrap.ts b/src/bootstrap.ts index c3ccc4d..b7ba865 100644 --- a/src/bootstrap.ts +++ b/src/bootstrap.ts @@ -51,6 +51,7 @@ import { AiUsageRepository } from './database/ai-usage-repository.js'; import { importLegacyAccountContent } from './database/legacy-content-import.js'; import { RecipientDeduplicator } from './services/recipient-deduplicator.js'; import { createConfigBackedSubredditPollingControl } from './services/subreddit-polling-control.js'; +import { CandidateQueueRepository, RuntimeStateRepository } from './database/candidate-queue-repository.js'; export interface AppContainer { config: Config; @@ -69,6 +70,8 @@ export interface AppContainer { conversationStateRepo: ConversationStateRepository; conversationOutcomesRepo: ConversationOutcomesRepository; recipientDeduplicator: RecipientDeduplicator; + candidateQueueRepo: CandidateQueueRepository; + runtimeStateRepo: RuntimeStateRepository; redditApi: RedditApiClient; aiClient: OpenRouterClient; @@ -126,6 +129,8 @@ export async function bootstrap( const conversationStateRepo = new ConversationStateRepository(db, bootstrapAccountId); const conversationOutcomesRepo = new ConversationOutcomesRepository(db, bootstrapAccountId); const recipientDeduplicator = new RecipientDeduplicator(new RecipientHistoryRepository(db)); + const candidateQueueRepo = new CandidateQueueRepository(db); + const runtimeStateRepo = new RuntimeStateRepository(db); const redditAccountsRepo = bootstrapAccountsRepo; const accountContentRepo = new AccountContentRepository(db); const aiUsageRepo = new AiUsageRepository(db); @@ -360,6 +365,8 @@ export async function bootstrap( broadcaster, process.cwd(), aiUsageRepo, + runtimeStateRepo.isBotRunning(), + (running) => runtimeStateRepo.setBotRunning(running), ); webServer.setChatService(chatService); @@ -429,6 +436,7 @@ export async function bootstrap( webServer.setRedditAccountsService(redditAccounts); webServer.setAccountRuntimeManager(accountRuntimeManager); webServer.setConversationStateRepository(conversationStateRepo); + webServer.setCandidateQueueRepository(candidateQueueRepo); webServer.setConfigStore(new ConfigStore(db)); webServer.setOnConfigChange(() => logger.info('Runtime config updated via dashboard')); } @@ -463,6 +471,8 @@ export async function bootstrap( conversationStateRepo, conversationOutcomesRepo, recipientDeduplicator, + candidateQueueRepo, + runtimeStateRepo, redditApi, aiClient, monitorService, diff --git a/src/database/candidate-queue-repository.ts b/src/database/candidate-queue-repository.ts new file mode 100644 index 0000000..b2363d8 --- /dev/null +++ b/src/database/candidate-queue-repository.ts @@ -0,0 +1,338 @@ +import type Database from 'better-sqlite3'; +import type { RedditPost } from '../reddit/api.js'; +import type { ProcessedPost } from '../services/processor.js'; + +export type CandidateStatus = + | 'discovered' + | 'screening' + | 'ready' + | 'sending' + | 'retryable' + | 'sent' + | 'rejected'; + +export interface CandidateQueueRecord { + postId: string; + subreddit: string; + author: string; + sourceType: 'image' | 'chat'; + post: RedditPost; + prepared: ProcessedPost | null; + priority: number; + status: CandidateStatus; + discoveredAt: number; + updatedAt: number; + availableAt: number; + leaseOwner: string | null; + leaseExpiresAt: number | null; + assignedAccountId: string | null; + attemptCount: number; + lastError: string | null; +} + +export interface CandidateQueueStats { + total: number; + byStatus: Record; + ready: number; + oldestReadyAt: number | null; + rejectionReasons: Array<{ reason: string; count: number }>; +} + +export interface CandidateEnqueueResult { + discovered: number; + duplicatePost: number; + duplicateAuthor: number; + previouslyProcessed: number; + previouslyContacted: number; +} + +const ACTIVE_STATUSES: CandidateStatus[] = ['discovered', 'screening', 'ready', 'sending', 'retryable']; + +/** Global, durable funnel from Reddit listing discovery through outbound delivery. */ +export class CandidateQueueRepository { + constructor(private readonly db: Database.Database) {} + + enqueue( + posts: RedditPost[], + sourceType: 'image' | 'chat', + priority: number, + now: number = Date.now(), + ): CandidateEnqueueResult { + const result: CandidateEnqueueResult = { + discovered: 0, + duplicatePost: 0, + duplicateAuthor: 0, + previouslyProcessed: 0, + previouslyContacted: 0, + }; + const existingPost = this.db.prepare('SELECT 1 FROM candidate_queue WHERE post_id = ?'); + const processedPost = this.db.prepare('SELECT 1 FROM processed_posts WHERE id = ? LIMIT 1'); + const contacted = this.db.prepare(` + SELECT 1 FROM ( + SELECT recipient FROM recipient_suppressions WHERE recipient = ? COLLATE NOCASE + UNION ALL + SELECT recipient FROM sent_messages WHERE recipient = ? COLLATE NOCASE + UNION ALL + SELECT username AS recipient FROM reddit_conversations WHERE username = ? COLLATE NOCASE + ) LIMIT 1 + `); + const activeAuthor = this.db.prepare(` + SELECT 1 FROM candidate_queue + WHERE author = ? COLLATE NOCASE + AND status IN (${ACTIVE_STATUSES.map(() => '?').join(', ')}) + LIMIT 1 + `); + const insert = this.db.prepare(` + INSERT INTO candidate_queue ( + post_id, subreddit, author, source_type, payload_json, post_created_at, priority, + status, discovered_at, updated_at, available_at + ) VALUES (?, ?, ?, ?, ?, ?, ?, 'discovered', ?, ?, ?) + `); + + this.db.transaction(() => { + for (const post of posts) { + if (existingPost.get(post.id)) { + result.duplicatePost += 1; + continue; + } + if (processedPost.get(post.id)) { + result.previouslyProcessed += 1; + continue; + } + if (contacted.get(post.author, post.author, post.author)) { + result.previouslyContacted += 1; + continue; + } + if (activeAuthor.get(post.author, ...ACTIVE_STATUSES)) { + result.duplicateAuthor += 1; + continue; + } + insert.run( + post.id, + post.subreddit, + post.author, + sourceType, + JSON.stringify(post), + post.created_utc * 1_000, + priority, + now, + now, + now, + ); + result.discovered += 1; + } + })(); + return result; + } + + claimForScreening(workerId: string, leaseMs: number, now: number = Date.now()): CandidateQueueRecord | null { + return this.claim('discovered', 'screening', workerId, leaseMs, now, false); + } + + claimReady(accountId: string, workerId: string, leaseMs: number, now: number = Date.now()): CandidateQueueRecord | null { + const claimed = this.claim('ready', 'sending', workerId, leaseMs, now, true, accountId); + return claimed; + } + + markReady(postId: string, workerId: string, prepared: ProcessedPost, now: number = Date.now()): boolean { + return this.db.prepare(` + UPDATE candidate_queue + SET status = 'ready', prepared_json = ?, available_at = ?, updated_at = ?, + lease_owner = NULL, lease_expires_at = NULL, last_error = NULL + WHERE post_id = ? AND status = 'screening' AND lease_owner = ? + `).run(JSON.stringify(prepared), now, now, postId, workerId).changes === 1; + } + + reject(postId: string, workerId: string, reason: string, now: number = Date.now()): boolean { + return this.db.prepare(` + UPDATE candidate_queue + SET status = 'rejected', last_error = ?, updated_at = ?, + lease_owner = NULL, lease_expires_at = NULL, assigned_account_id = NULL + WHERE post_id = ? AND status IN ('screening', 'sending') AND lease_owner = ? + `).run(reason, now, postId, workerId).changes === 1; + } + + deferScreening( + postId: string, + workerId: string, + reason: string, + availableAt: number, + now: number = Date.now(), + ): boolean { + return this.db.prepare(` + UPDATE candidate_queue + SET status = 'discovered', last_error = ?, available_at = ?, updated_at = ?, + lease_owner = NULL, lease_expires_at = NULL + WHERE post_id = ? AND status = 'screening' AND lease_owner = ? + `).run(reason, availableAt, now, postId, workerId).changes === 1; + } + + deferDelivery( + postId: string, + workerId: string, + reason: string, + availableAt: number, + now: number = Date.now(), + ): boolean { + return this.db.prepare(` + UPDATE candidate_queue + SET status = 'retryable', last_error = ?, available_at = ?, updated_at = ?, + lease_owner = NULL, lease_expires_at = NULL, assigned_account_id = NULL + WHERE post_id = ? AND status = 'sending' AND lease_owner = ? + `).run(reason, availableAt, now, postId, workerId).changes === 1; + } + + markSent(postId: string, workerId: string, now: number = Date.now()): boolean { + return this.db.prepare(` + UPDATE candidate_queue + SET status = 'sent', last_error = NULL, updated_at = ?, + lease_owner = NULL, lease_expires_at = NULL + WHERE post_id = ? AND status = 'sending' AND lease_owner = ? + `).run(now, postId, workerId).changes === 1; + } + + promoteDueRetries(now: number = Date.now()): number { + return this.db.prepare(` + UPDATE candidate_queue + SET status = 'ready', updated_at = ?, assigned_account_id = NULL + WHERE status = 'retryable' AND available_at <= ? + `).run(now, now).changes; + } + + recoverExpiredLeases(now: number = Date.now(), sendingDelayMs = 10 * 60_000): number { + return this.db.prepare(` + UPDATE candidate_queue + SET status = CASE status WHEN 'screening' THEN 'discovered' ELSE 'retryable' END, + lease_owner = NULL, lease_expires_at = NULL, assigned_account_id = NULL, + available_at = CASE status WHEN 'screening' THEN ? ELSE ? END, + updated_at = ?, last_error = 'Recovered after interrupted worker' + WHERE status IN ('screening', 'sending') AND lease_expires_at <= ? + `).run(now, now + sendingDelayMs, now, now).changes; + } + + recoverAllLeases(now: number = Date.now(), sendingDelayMs = 10 * 60_000): number { + return this.db.prepare(` + UPDATE candidate_queue + SET status = CASE status WHEN 'screening' THEN 'discovered' ELSE 'retryable' END, + lease_owner = NULL, lease_expires_at = NULL, assigned_account_id = NULL, + available_at = CASE status WHEN 'screening' THEN ? ELSE ? END, + updated_at = ?, last_error = 'Recovered after process restart' + WHERE status IN ('screening', 'sending') + `).run(now, now + sendingDelayMs, now).changes; + } + + expireStale(cutoff: number, now: number = Date.now()): number { + return this.db.prepare(` + UPDATE candidate_queue + SET status = 'rejected', last_error = 'Candidate expired before delivery', + lease_owner = NULL, lease_expires_at = NULL, assigned_account_id = NULL, updated_at = ? + WHERE status IN ('discovered', 'ready', 'retryable') AND post_created_at < ? + `).run(now, cutoff).changes; + } + + countActive(): number { + const row = this.db.prepare(` + SELECT COUNT(*) AS count FROM candidate_queue + WHERE status IN (${ACTIVE_STATUSES.map(() => '?').join(', ')}) + `).get(...ACTIVE_STATUSES) as { count: number }; + return row.count; + } + + countReady(): number { + return (this.db.prepare(`SELECT COUNT(*) AS count FROM candidate_queue WHERE status = 'ready'`) + .get() as { count: number }).count; + } + + getStats(): CandidateQueueStats { + const statuses: CandidateStatus[] = ['discovered', 'screening', 'ready', 'sending', 'retryable', 'sent', 'rejected']; + const byStatus = Object.fromEntries(statuses.map((status) => [status, 0])) as Record; + const rows = this.db.prepare('SELECT status, COUNT(*) AS count FROM candidate_queue GROUP BY status') + .all() as Array<{ status: CandidateStatus; count: number }>; + for (const row of rows) byStatus[row.status] = row.count; + const oldest = this.db.prepare(`SELECT MIN(discovered_at) AS at FROM candidate_queue WHERE status = 'ready'`) + .get() as { at: number | null }; + const rejectionReasons = this.db.prepare(` + SELECT COALESCE(last_error, 'Unknown') AS reason, COUNT(*) AS count + FROM candidate_queue WHERE status = 'rejected' + GROUP BY reason ORDER BY count DESC LIMIT 20 + `).all() as Array<{ reason: string; count: number }>; + return { + total: Object.values(byStatus).reduce((sum, count) => sum + count, 0), + byStatus, + ready: byStatus.ready, + oldestReadyAt: oldest.at, + rejectionReasons, + }; + } + + private claim( + from: 'discovered' | 'ready', + to: 'screening' | 'sending', + workerId: string, + leaseMs: number, + now: number, + requirePrepared: boolean, + accountId?: string, + ): CandidateQueueRecord | null { + if (!workerId.trim()) throw new Error('Candidate worker ID is required'); + if (!Number.isFinite(leaseMs) || leaseMs <= 0) throw new Error('Candidate lease must be positive'); + return this.db.transaction(() => { + const row = this.db.prepare(` + SELECT post_id FROM candidate_queue + WHERE status = ? AND available_at <= ? ${requirePrepared ? 'AND prepared_json IS NOT NULL' : ''} + ORDER BY priority DESC, post_created_at DESC, discovered_at ASC + LIMIT 1 + `).get(from, now) as { post_id: string } | undefined; + if (!row) return null; + const changed = this.db.prepare(` + UPDATE candidate_queue + SET status = ?, lease_owner = ?, lease_expires_at = ?, updated_at = ?, + assigned_account_id = ?, attempt_count = attempt_count + 1 + WHERE post_id = ? AND status = ? + `).run(to, workerId, now + leaseMs, now, accountId ?? null, row.post_id, from).changes; + return changed === 1 ? this.get(row.post_id) : null; + })(); + } + + private get(postId: string): CandidateQueueRecord | null { + const row = this.db.prepare('SELECT * FROM candidate_queue WHERE post_id = ?') + .get(postId) as Record | undefined; + if (!row) return null; + return { + postId: row.post_id as string, + subreddit: row.subreddit as string, + author: row.author as string, + sourceType: row.source_type as 'image' | 'chat', + post: JSON.parse(row.payload_json as string) as RedditPost, + prepared: row.prepared_json ? JSON.parse(row.prepared_json as string) as ProcessedPost : null, + priority: row.priority as number, + status: row.status as CandidateStatus, + discoveredAt: row.discovered_at as number, + updatedAt: row.updated_at as number, + availableAt: row.available_at as number, + leaseOwner: (row.lease_owner as string | null) ?? null, + leaseExpiresAt: (row.lease_expires_at as number | null) ?? null, + assignedAccountId: (row.assigned_account_id as string | null) ?? null, + attemptCount: row.attempt_count as number, + lastError: (row.last_error as string | null) ?? null, + }; + } +} + +/** Persists the operator's intended pause/run state across container restarts. */ +export class RuntimeStateRepository { + constructor(private readonly db: Database.Database) {} + + isBotRunning(): boolean { + const row = this.db.prepare('SELECT bot_running FROM runtime_state WHERE id = 1') + .get() as { bot_running: number } | undefined; + return row?.bot_running === 1; + } + + setBotRunning(running: boolean, now: number = Date.now()): void { + this.db.prepare(` + INSERT INTO runtime_state (id, bot_running, updated_at) VALUES (1, ?, ?) + ON CONFLICT(id) DO UPDATE SET bot_running = excluded.bot_running, updated_at = excluded.updated_at + `).run(running ? 1 : 0, now); + } +} diff --git a/src/database/client.ts b/src/database/client.ts index 756d221..8d518b5 100644 --- a/src/database/client.ts +++ b/src/database/client.ts @@ -276,6 +276,42 @@ export function runMigrations(db: Database.Database): void { () => { addColumnIfNotExists(db, 'processed_posts', 'post_created_at', 'INTEGER'); }, + () => db.exec(` + CREATE TABLE IF NOT EXISTS candidate_queue ( + post_id TEXT PRIMARY KEY, + subreddit TEXT NOT NULL, + author TEXT NOT NULL COLLATE NOCASE, + source_type TEXT NOT NULL CHECK(source_type IN ('image', 'chat')), + payload_json TEXT NOT NULL, + prepared_json TEXT, + post_created_at INTEGER NOT NULL, + priority REAL NOT NULL DEFAULT 1, + status TEXT NOT NULL CHECK(status IN ( + 'discovered', 'screening', 'ready', 'sending', 'retryable', 'sent', 'rejected' + )), + discovered_at INTEGER NOT NULL, + updated_at INTEGER NOT NULL, + available_at INTEGER NOT NULL, + lease_owner TEXT, + lease_expires_at INTEGER, + assigned_account_id TEXT, + attempt_count INTEGER NOT NULL DEFAULT 0, + last_error TEXT, + FOREIGN KEY (assigned_account_id) REFERENCES reddit_accounts(account_id) ON DELETE SET NULL + ); + CREATE INDEX IF NOT EXISTS idx_candidate_queue_claim + ON candidate_queue(status, available_at, priority DESC, post_created_at DESC); + CREATE INDEX IF NOT EXISTS idx_candidate_queue_author + ON candidate_queue(author COLLATE NOCASE, status); + CREATE INDEX IF NOT EXISTS idx_candidate_queue_updated + ON candidate_queue(updated_at DESC); + + CREATE TABLE IF NOT EXISTS runtime_state ( + id INTEGER PRIMARY KEY CHECK(id = 1), + bot_running INTEGER NOT NULL CHECK(bot_running IN (0, 1)), + updated_at INTEGER NOT NULL + ); + `), ]; let currentVersion = db.pragma('user_version', { simple: true }) as number; diff --git a/src/database/repositories.ts b/src/database/repositories.ts index ebb40a2..368cdbb 100644 --- a/src/database/repositories.ts +++ b/src/database/repositories.ts @@ -129,7 +129,7 @@ export class RecipientHistoryRepository { private readonly commitClaimStmt: Database.Statement; private readonly releaseClaimStmt: Database.Statement; - constructor(db: Database.Database) { + constructor(private readonly db: Database.Database) { this.hasSentToStmt = db.prepare(` SELECT 1 FROM ( SELECT recipient FROM sent_messages WHERE recipient = ? COLLATE NOCASE @@ -179,6 +179,21 @@ export class RecipientHistoryRepository { releaseClaim(recipient: string): void { this.releaseClaimStmt.run(recipient); } + + recoverStaleClaims(cutoff: number): number { + return this.db.prepare(` + DELETE FROM recipient_suppressions + WHERE reason = 'pending-send' AND created_at <= ? + AND NOT EXISTS ( + SELECT 1 FROM sent_messages + WHERE recipient = recipient_suppressions.recipient COLLATE NOCASE + ) + AND NOT EXISTS ( + SELECT 1 FROM reddit_conversations + WHERE username = recipient_suppressions.recipient COLLATE NOCASE + ) + `).run(cutoff).changes; + } } /** @@ -408,6 +423,7 @@ export class ActivityLogRepository { export class PostsRepository { private insertStmt: Database.Statement; + private insertIfAbsentStmt: Database.Statement; private existsStmt: Database.Statement; private markSentStmt: Database.Statement; private updateAnalysisStmt: Database.Statement; @@ -424,6 +440,11 @@ export class PostsRepository { account_id, id, subreddit, author, title, post_url, image_url, post_created_at, processed_at ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) `); + this.insertIfAbsentStmt = db.prepare(` + INSERT OR IGNORE INTO processed_posts ( + account_id, id, subreddit, author, title, post_url, image_url, post_created_at, processed_at + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) + `); this.existsStmt = db.prepare(`SELECT 1 FROM processed_posts WHERE account_id = ? AND id = ?`); @@ -431,7 +452,7 @@ export class PostsRepository { this.updateAnalysisStmt = db.prepare(` UPDATE processed_posts - SET image_analysis = ?, generated_message = ? + SET image_analysis = ?, generated_message = ?, error = NULL WHERE account_id = ? AND id = ? `); @@ -481,6 +502,30 @@ export class PostsRepository { ); } + /** Idempotent insert used when a leased candidate resumes after a restart. */ + insertIfAbsent(post: { + id: string; + subreddit: string; + author: string; + title: string; + postUrl?: string | null; + imageUrl: string | null; + postCreatedAt?: number | null; + processedAt: number; + }): boolean { + return this.insertIfAbsentStmt.run( + this.accountId, + post.id, + post.subreddit, + post.author, + post.title, + post.postUrl ?? null, + post.imageUrl, + post.postCreatedAt ?? null, + post.processedAt, + ).changes === 1; + } + updateAnalysis(postId: string, analysis: string, message: string): void { this.updateAnalysisStmt.run(analysis, message, this.accountId, postId); } diff --git a/src/main-loop.ts b/src/main-loop.ts index b8ee13f..4a69115 100644 --- a/src/main-loop.ts +++ b/src/main-loop.ts @@ -425,32 +425,44 @@ export async function runMainLoop(app: AppContainer): Promise { } async function runConcurrentAccountRuntimes(app: AppContainer): Promise { - 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((resolve) => { let shuttingDown = false; const controller = new AbortController(); let nextScannerIndex = 0; + let nextSubredditIndex = 0; - const discoverAndDispatch = async (): Promise => { - const runtimes = accountRuntimeManager.list() + const runningRuntimes = () => accountRuntimeManager.list() .filter((snapshot) => snapshot.state === 'running') .map((snapshot) => accountRuntimeManager.get(snapshot.accountId)) .filter((runtime): runtime is NonNullable => 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 { 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 => { 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 => { + 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 => { + 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 { 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(); diff --git a/src/services/chat/index.ts b/src/services/chat/index.ts index 79edcc9..b06ab35 100644 --- a/src/services/chat/index.ts +++ b/src/services/chat/index.ts @@ -7,6 +7,7 @@ export type { SyncPhase, SyncCompletion, SyncProgressListener, + ConversationStartResult, } from './service.js'; export type { ActiveChat, diff --git a/src/services/chat/service.ts b/src/services/chat/service.ts index 34d70bb..2c0a3a8 100644 --- a/src/services/chat/service.ts +++ b/src/services/chat/service.ts @@ -53,6 +53,11 @@ export interface UsernameResolver { resolveUsernameToT2?(username: string): Promise; } +export type ConversationStartResult = + | { status: 'sent'; chat: ActiveChat } + | { status: 'terminal'; reason: string } + | { status: 'retryable'; reason: string; retryAt: number }; + interface BuildChatOpts { knownUsernames: Map; includeTimelineMessages?: boolean; @@ -620,21 +625,37 @@ export class ChatService { message: string, knownT2Id?: string, ): Promise { - if (!this.accountAuth?.isFullyAuthenticated()) return null; + const result = await this.startConversationDetailed(username, message, knownT2Id); + return result.status === 'sent' ? result.chat : null; + } + + /** Start a DM while preserving terminal versus retryable failure semantics. */ + async startConversationDetailed( + username: string, + message: string, + knownT2Id?: string, + ): Promise { + if (!this.accountAuth?.isFullyAuthenticated()) { + return { + status: 'retryable', + reason: 'Account authentication is temporarily unavailable', + retryAt: Date.now() + 60_000, + }; + } const active = this.getInvitationCooldown(); if (active) { const secondsLeft = Math.ceil((active.until - Date.now()) / 1000); this.logger.warn( `[chat] skipping startConversation(${username}) — invitation cooldown active for ${secondsLeft}s (${active.redditCode ?? 'rate-limited'})`, ); - return null; + return { status: 'retryable', reason: active.reason, retryAt: active.until }; } return this.enqueue(async () => { try { const t2Id = await this.resolveUsernameToT2(username, knownT2Id); if (!t2Id) { this.logger.error(`[chat] cannot start conversation: unknown user "${username}"`); - return null; + return { status: 'terminal', reason: 'Reddit user no longer exists' }; } const mxid = MatrixClient.t2ToMxid(t2Id); const { roomId, eventId } = await this.withAuth(() => @@ -666,14 +687,30 @@ export class ChatService { const merged = [chat, ...existing.filter((c) => c.roomId !== roomId)]; this.cache.set(merged); await this.cache.save(merged); - return chat; + return { status: 'sent', chat }; } catch (error) { if (error instanceof MatrixRateLimitError) { this.setInvitationCooldown(error); - return null; + const cooldown = this.getInvitationCooldown(); + return { + status: 'retryable', + reason: cooldown?.reason ?? 'Reddit temporarily rate-limited chat invitations', + retryAt: cooldown?.until ?? Date.now() + error.retryAfterMs + 5_000, + }; } this.logger.error('[chat] startConversation failed:', error); - return null; + const message = error instanceof Error ? error.message : String(error); + if (/do not accept chat requests/i.test(message)) { + return { status: 'terminal', reason: 'Recipient does not accept chat requests' }; + } + if (/unknown user|not found|M_NOT_FOUND/i.test(message)) { + return { status: 'terminal', reason: 'Reddit user is unavailable' }; + } + return { + status: 'retryable', + reason: 'Reddit chat invitation failed temporarily', + retryAt: Date.now() + 60_000, + }; } }); } diff --git a/src/services/monitor.ts b/src/services/monitor.ts index 2d95ced..7bc90b9 100644 --- a/src/services/monitor.ts +++ b/src/services/monitor.ts @@ -16,11 +16,17 @@ interface ListingScreeningSummary { sellerSignals: number; missingGenderTag: number; genderNotTargeted: number; + missingAgeTag: number; + underage: number; eligible: number; outsideRecentWindow: number; deferred: number; } +export type CandidateScreeningResult = + | { eligible: true } + | { eligible: false; reason: string }; + /** Coordinates permanent polling disablement across account runtimes. */ export interface SubredditPollingControl { disableUnavailableSubreddit(subreddit: string): void; @@ -67,6 +73,8 @@ export class MonitorService { sellerSignals: 0, missingGenderTag: 0, genderNotTargeted: 0, + missingAgeTag: 0, + underage: 0, eligible: 0, outsideRecentWindow: 0, deferred: 0, @@ -93,6 +101,39 @@ export class MonitorService { continue; } + // Chat subs almost always have a self-declared gender tag in the + // title/body (e.g. [25F], M4F). Run this free check before seller + // profile requests so obvious non-target posts consume no Reddit + // request budget. + if (subreddit.type === 'chat') { + const tag = parseGenderTag(`${post.title}\n${post.selftext ?? ''}`); + if (targetGenders.length > 0 && !tag) { + summary.missingGenderTag += 1; + this.logger.debug(`Skipping chat post ${post.id} - no gender tag in title/body`); + this.recordRejection(post, subreddit.type, 'No gender tag in title/body'); + continue; + } + if (tag && targetGenders.length > 0 && !targetGenders.includes(tag)) { + summary.genderNotTargeted += 1; + this.logger.debug( + `Skipping chat post ${post.id} - gender tag "${tag}" not in target list [${targetGenders.join(', ')}]`, + ); + this.recordRejection(post, subreddit.type, `Gender tag "${tag}" not targeted`); + continue; + } + const age = parseSelfDeclaredAge(`${post.title}\n${post.selftext ?? ''}`); + if (age === null) { + summary.missingAgeTag += 1; + this.recordRejection(post, subreddit.type, 'No adult age tag in title/body'); + continue; + } + if (age < 18) { + summary.underage += 1; + this.recordRejection(post, subreddit.type, `Self-declared age ${age} is under 18`); + continue; + } + } + if ( this.config.reddit.filterSellers && (this.config.reddit.sellerKeywords.length > 0 || this.config.reddit.sellerDomains.length > 0) @@ -111,27 +152,6 @@ export class MonitorService { } } - // Chat subs almost always have a self-declared gender tag in the - // title/body (e.g. [25F], M4F). Drop posts that don't match the - // configured targetGenders here so they never enter the processor. - if (subreddit.type === 'chat' && targetGenders.length > 0) { - const tag = parseGenderTag(`${post.title}\n${post.selftext ?? ''}`); - if (!tag) { - summary.missingGenderTag += 1; - this.logger.debug(`Skipping chat post ${post.id} - no gender tag in title/body`); - this.recordRejection(post, subreddit.type, 'No gender tag in title/body'); - continue; - } - if (!targetGenders.includes(tag)) { - summary.genderNotTargeted += 1; - this.logger.debug( - `Skipping chat post ${post.id} - gender tag "${tag}" not in target list [${targetGenders.join(', ')}]`, - ); - this.recordRejection(post, subreddit.type, `Gender tag "${tag}" not targeted`); - continue; - } - } - // At this point the candidate has passed every check. Select it now // rather than spending more Reddit requests screening users that // cannot be acted on during this poll cycle. @@ -186,6 +206,61 @@ export class MonitorService { return []; } + /** Fetch one listing without performing per-author screening. */ + async fetchCandidates(subreddit: Config['reddit']['subreddits'][number]): Promise { + if (!this.isSubredditEnabled(subreddit.name)) return []; + try { + this.logger.debug(`Collecting candidate posts from r/${subreddit.name} (${subreddit.type})...`); + return await this.redditApi.getNewPosts(subreddit.name, 25, subreddit.type); + } catch (error) { + if (error instanceof RedditRateLimitError) throw error; + if (error instanceof RedditSubredditUnavailableError) { + this.pollingControl?.disableUnavailableSubreddit(subreddit.name); + this.logger.warn(`Disabled r/${subreddit.name}: Reddit returned HTTP 404`); + return []; + } + const message = error instanceof Error ? error.message : String(error); + this.logger.error(`Failed to collect candidates from r/${subreddit.name}: ${message}`); + return []; + } + } + + /** Apply cheap eligibility checks before any profile or AI work. */ + async screenCandidate(post: RedditPost, sourceType: 'image' | 'chat'): Promise { + if (this.recipientHistory.hasSentTo(post.author)) { + return { eligible: false, reason: 'User already contacted across accounts' }; + } + + if (sourceType === 'chat') { + const text = `${post.title}\n${post.selftext ?? ''}`; + const tag = parseGenderTag(text); + if (this.config.reddit.targetGenders.length > 0 && !tag) { + return { eligible: false, reason: 'No gender tag in title/body' }; + } + if (tag && this.config.reddit.targetGenders.length > 0 && !this.config.reddit.targetGenders.includes(tag)) { + return { eligible: false, reason: `Gender tag "${tag}" not targeted` }; + } + const age = parseSelfDeclaredAge(text); + if (age === null) return { eligible: false, reason: 'No adult age tag in title/body' }; + if (age < 18) return { eligible: false, reason: `Self-declared age ${age} is under 18` }; + } + + if ( + this.config.reddit.filterSellers + && (this.config.reddit.sellerKeywords.length > 0 || this.config.reddit.sellerDomains.length > 0) + && await this.redditApi.isSellerProfile( + post.author, + this.config.reddit.sellerKeywords, + this.config.reddit.sellerDomains, + post, + ) + ) { + return { eligible: false, reason: 'Seller signals detected' }; + } + + return { eligible: true }; + } + /** * Build an interleaved check order that balances image and chat subs. * @@ -252,6 +327,8 @@ export class MonitorService { `${summary.sellerSignals} seller signals`, `${summary.missingGenderTag} missing gender tag`, `${summary.genderNotTargeted} gender not targeted`, + `${summary.missingAgeTag} missing age tag`, + `${summary.underage} underage`, `${summary.eligible} eligible${selected ? ' (selected)' : ''}`, `${summary.outsideRecentWindow} outside recent window`, ]; @@ -262,12 +339,12 @@ export class MonitorService { } /** Persist a completed screening decision so later polls do not repeat it. */ - private recordRejection( + recordRejection( post: RedditPost, subredditType: 'image' | 'chat', reason: string, ): void { - this.postsRepo.insert({ + this.postsRepo.insertIfAbsent({ id: post.id, subreddit: post.subreddit, author: post.author, @@ -280,3 +357,16 @@ export class MonitorService { this.postsRepo.setError(post.id, reason); } } + +/** Parse common adult-subreddit age tags without treating arbitrary numbers as ages. */ +export function parseSelfDeclaredAge(text: string): number | null { + const patterns = [ + /(?:^|[\s[(])([1-9]\d?)\s*\[?(?:F|M|NB|TF|TM)(?:4[A-Z]+)?\b/i, + /(?:^|[\s[(])(?:F|M|NB|TF|TM)\s*([1-9]\d?)\b/i, + ]; + for (const pattern of patterns) { + const match = pattern.exec(text); + if (match) return Number.parseInt(match[1], 10); + } + return null; +} diff --git a/src/services/recipient-deduplicator.ts b/src/services/recipient-deduplicator.ts index 276bf0c..367ea30 100644 --- a/src/services/recipient-deduplicator.ts +++ b/src/services/recipient-deduplicator.ts @@ -3,6 +3,7 @@ export interface RecipientHistory { tryClaim?(recipient: string): boolean; commitClaim?(recipient: string): void; releaseClaim?(recipient: string): void; + recoverStaleClaims?(cutoff: number): number; } /** @@ -42,6 +43,10 @@ export class RecipientDeduplicator { this.reservations.delete(key); this.history.releaseClaim?.(recipient); } + + recoverStaleReservations(cutoff: number): number { + return this.history.recoverStaleClaims?.(cutoff) ?? 0; + } } function normalizeRecipient(recipient: string): string { diff --git a/src/services/reddit-account-runtime-builder.ts b/src/services/reddit-account-runtime-builder.ts index 9bdff21..216422b 100644 --- a/src/services/reddit-account-runtime-builder.ts +++ b/src/services/reddit-account-runtime-builder.ts @@ -17,7 +17,7 @@ import { RedditApiClient, RedditRateLimitError, type RedditPost } from '../reddi import { RedditRequestCoordinator } from '../reddit/request-coordinator.js'; import { OpenRouterClient } from '../ai/openrouter.js'; import { MonitorService } from './monitor.js'; -import { ProcessorService, isSkip } from './processor.js'; +import { ProcessorService, isSkip, type ProcessResult, type ProcessedPost } from './processor.js'; import { ResponseCheckerService } from './response-checker.js'; import { StrategySelector } from './strategy-selector.js'; import { @@ -37,7 +37,7 @@ import { RateLimiter } from '../utils/rate-limiter.js'; import { RedditAccountRuntime, RedditAccountRuntimeFactory, accountBrowserProfileDir, type RedditAccountRuntimeResources, - type VerifiedAccountSession, + type VerifiedAccountSession, type PreparedDeliveryResult, } from './reddit-account-runtime.js'; import { RecipientDeduplicator } from './recipient-deduplicator.js'; import { ConfigStore } from '../config/persistence.js'; @@ -291,80 +291,89 @@ class AccountRuntimeLoop { } async dispatchPost(post: RedditPost, signal?: AbortSignal): Promise { - const { postsRepo, messagesRepo, redditApi, processorService, rateLimiter, chatService, responseChecker } = this.resources; - if (signal?.aborted) return; - try { - const subreddit = this.config.reddit.subreddits.find((entry) => entry.name.toLowerCase() === post.subreddit.toLowerCase()); - const isChat = subreddit?.type === 'chat'; - const imageUrl = isChat ? null : redditApi.getImageUrl(post); - postsRepo.insert({ - id: post.id, subreddit: post.subreddit, author: post.author, title: post.title, - postUrl: redditApi.getPostUrl(post), imageUrl, - postCreatedAt: post.created_utc * 1000, processedAt: Date.now(), + const processed = await this.preparePost(post, signal); + if (isSkip(processed)) return; + await this.dispatchPreparedPost(processed, signal); + } + + async preparePost(post: RedditPost, signal?: AbortSignal): Promise { + const { postsRepo, redditApi, processorService, responseChecker } = this.resources; + if (signal?.aborted) throw new DOMException('Operation aborted', 'AbortError'); + const subreddit = this.config.reddit.subreddits.find((entry) => entry.name.toLowerCase() === post.subreddit.toLowerCase()); + const isChat = subreddit?.type === 'chat'; + postsRepo.insertIfAbsent({ + id: post.id, subreddit: post.subreddit, author: post.author, title: post.title, + postUrl: redditApi.getPostUrl(post), imageUrl: isChat ? null : redditApi.getImageUrl(post), + postCreatedAt: post.created_utc * 1_000, processedAt: Date.now(), + }); + processorService.setSuccessfulExamples(responseChecker.getSuccessfulExamples(3, post.subreddit)); + const processed = await processorService.processPost(post); + if (isSkip(processed)) { + postsRepo.setError(post.id, processed.skipReason); + this.resources.activityLogRepo.insert({ + timestamp: Date.now(), type: 'skipped', + message: `Skipped post by u/${post.author} (${processed.skipReason})`, recipient: post.author, }); - // Seller screening is performed once by the shared MonitorService before - // assignment. Do not repeat its profile request for every recipient. - processorService.setSuccessfulExamples(responseChecker.getSuccessfulExamples(3, post.subreddit)); - const processed = await processorService.processPost(post); - if (isSkip(processed)) { - postsRepo.setError(post.id, processed.skipReason); - this.resources.activityLogRepo.insert({ timestamp: Date.now(), type: 'skipped', message: `Skipped post by u/${post.author} (${processed.skipReason})`, recipient: post.author }); - return; - } - postsRepo.updateAnalysis(post.id, processed.imageAnalysis, processed.generatedMessage); - await rateLimiter.throttle(async () => { - if (this.config.browser.dryRun) { - const reason = 'Dry run: message prepared but not sent'; - postsRepo.setError(post.id, reason); - this.resources.activityLogRepo.insert({ - timestamp: Date.now(), type: 'skipped', - message: `${reason} to u/${post.author}`, recipient: post.author, - }); - return false; - } - if (!this.recipientDeduplicator.tryReserve(post.author)) { - postsRepo.setError(post.id, 'User already contacted across accounts'); - this.resources.activityLogRepo.insert({ - timestamp: Date.now(), type: 'skipped', - message: `Skipped u/${post.author} (already contacted across accounts)`, - recipient: post.author, - }); - return false; - } - let chat; - try { - chat = await chatService.startConversation(post.author, processed.generatedMessage, post.author_fullname); - } catch (error) { - this.recipientDeduplicator.release(post.author); - throw error; - } - if (!chat) { - this.recipientDeduplicator.release(post.author); - return false; - } - postsRepo.markMessageSent(post.id); - 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, - }); - this.recipientDeduplicator.commit(post.author); - this.resources.activityLogRepo.insert({ - timestamp: Date.now(), type: 'sent', message: `Sent message to u/${post.author} (r/${post.subreddit})`, - recipient: post.author, imageUrl: imageUrl ?? undefined, postTitle: post.title, postUrl: redditApi.getPostUrl(post), - }); - void chatService.refreshTopChats(5); - return true; - }, (success) => success, signal); - } catch (error) { - if (signal?.aborted) return; - postsRepo.setError(post.id, error instanceof Error ? error.message : String(error)); - this.resources.activityLogRepo.insert({ timestamp: Date.now(), type: 'error', message: `Failed processing post by u/${post.author}`, recipient: post.author }); - this.logger.error(`Account runtime failed processing post ${post.id}:`, error); + return processed; } + postsRepo.updateAnalysis(post.id, processed.imageAnalysis, processed.generatedMessage); + return processed; + } + + async dispatchPreparedPost(processed: ProcessedPost, signal?: AbortSignal): Promise { + const { postsRepo, messagesRepo, redditApi, rateLimiter, chatService } = this.resources; + const { post } = processed; + if (signal?.aborted) { + return { status: 'retryable', reason: 'Delivery interrupted by shutdown', retryAt: Date.now() + 5_000 }; + } + return rateLimiter.throttle(async (): Promise => { + if (this.config.browser.dryRun) { + const reason = 'Dry run: message prepared but not sent'; + postsRepo.setError(post.id, reason); + return { status: 'terminal', reason }; + } + if (!this.recipientDeduplicator.tryReserve(post.author)) { + const reason = 'User already contacted across accounts'; + postsRepo.setError(post.id, reason); + return { status: 'terminal', reason }; + } + + const start = await chatService.startConversationDetailed( + post.author, + processed.generatedMessage, + post.author_fullname, + ); + if (start.status !== 'sent') { + this.recipientDeduplicator.release(post.author); + if (start.status === 'terminal') { + postsRepo.setError(post.id, start.reason); + this.resources.activityLogRepo.insert({ + timestamp: Date.now(), type: 'skipped', + message: `Skipped u/${post.author} (${start.reason})`, recipient: post.author, + }); + } + return start; + } + + postsRepo.markMessageSent(post.id); + messagesRepo.insert({ + postId: post.id, recipient: post.author, subject: post.author, + message: processed.generatedMessage, sentAt: Date.now(), + strategyId: processed.strategyId, chatRoomId: start.chat.roomId, + experimentId: processed.experimentId, promptVersion: processed.promptVersion, + modelId: processed.modelId, selectionReason: processed.selectionReason, + sourceType: processed.sourceType, + }); + this.recipientDeduplicator.commit(post.author); + this.resources.activityLogRepo.insert({ + timestamp: Date.now(), type: 'sent', message: `Sent message to u/${post.author} (r/${post.subreddit})`, + recipient: post.author, + imageUrl: processed.sourceType === 'image' ? redditApi.getImageUrl(post) ?? undefined : undefined, + postTitle: post.title, postUrl: redditApi.getPostUrl(post), + }); + void chatService.refreshTopChats(5); + return { status: 'sent', conversationId: start.chat.roomId! }; + }, (result) => result.status === 'sent', signal); } } diff --git a/src/services/reddit-account-runtime.ts b/src/services/reddit-account-runtime.ts index 93893ca..dbf6678 100644 --- a/src/services/reddit-account-runtime.ts +++ b/src/services/reddit-account-runtime.ts @@ -31,6 +31,7 @@ import type { ProcessorService } from './processor.js'; import type { ResponseCheckerService } from './response-checker.js'; import type { RedditPost } from '../reddit/api.js'; import type { ConversationAutomationService } from './conversation-automation.js'; +import type { ProcessedPost, ProcessResult } from './processor.js'; /** Credentials passed to a factory. They must never be exposed by list/snapshot APIs. */ export interface VerifiedAccountSession { @@ -85,8 +86,15 @@ export interface RedditAccountRuntimeLifecycle { stop(): void | Promise; /** Process a post assigned by the shared discovery scheduler. */ dispatchPost?(post: RedditPost, signal?: AbortSignal): Promise; + preparePost?(post: RedditPost, signal?: AbortSignal): Promise; + dispatchPreparedPost?(post: ProcessedPost, signal?: AbortSignal): Promise; } +export type PreparedDeliveryResult = + | { status: 'sent'; conversationId: string } + | { status: 'terminal'; reason: string } + | { status: 'retryable'; reason: string; retryAt: number }; + export type RedditAccountRuntimeState = | 'created' | 'starting' @@ -185,6 +193,26 @@ export class RedditAccountRuntime { return this.resources.rateLimiter.getTimers().nextMessageAvailableAt === null; } + async canScan(): Promise { + if (this.state !== 'running') return false; + return this.resources.accountAuth.isFullyAuthenticated() + || await this.resources.accountAuth.ensureFullyAuthenticated(); + } + + preparePost(post: RedditPost, signal?: AbortSignal): Promise { + if (this.state !== 'running' || !this.lifecycle.preparePost) { + throw new Error(`Runtime ${this.session.redditUsername} cannot prepare candidates`); + } + return this.lifecycle.preparePost(post, signal); + } + + dispatchPreparedPost(post: ProcessedPost, signal?: AbortSignal): Promise { + if (this.state !== 'running' || !this.lifecycle.dispatchPreparedPost) { + throw new Error(`Runtime ${this.session.redditUsername} cannot dispatch prepared candidates`); + } + return this.lifecycle.dispatchPreparedPost(post, signal); + } + dispatchPost(post: RedditPost, signal?: AbortSignal): Promise { if (this.state !== 'running') { throw new Error(`Cannot assign a post to ${this.session.redditUsername}: runtime is not running`); diff --git a/src/web/api-catalog.ts b/src/web/api-catalog.ts index 69896bf..52912d2 100644 --- a/src/web/api-catalog.ts +++ b/src/web/api-catalog.ts @@ -68,6 +68,7 @@ export const API_OPERATIONS: readonly ApiOperation[] = [ operation('createJob', 'POST', '/api/v1/jobs', 'Start chat or media synchronisation', ['Jobs'], { access: 'control', requestSchema: AgentJobRequestSchema, responseSchema: AgentJobSchema, successStatus: 202 }), operation('getStats', 'GET', '/api/stats', 'Get bot statistics', ['Dashboard']), + operation('getCandidateFunnel', 'GET', '/api/analytics/candidate-funnel', 'Get durable candidate funnel counts', ['Analytics']), operation('getActivity', 'GET', '/api/activity', 'Get recent activity', ['Dashboard']), operation('getPostsByAuthor', 'GET', '/api/posts/by-author/{username}', 'Get processed posts by author', ['Dashboard']), operation('getMessages', 'GET', '/api/messages', 'Get recent sent messages', ['Dashboard']), diff --git a/src/web/context.ts b/src/web/context.ts index 935c762..1b3c604 100644 --- a/src/web/context.ts +++ b/src/web/context.ts @@ -23,6 +23,7 @@ import type { ConfigStore } from '../config/persistence.js'; import type { RedditAccountsService } from '../services/reddit-accounts.js'; import type { RedditAccountRuntimeManager } from '../services/reddit-account-runtime.js'; import type { AiUsageRepository } from '../database/ai-usage-repository.js'; +import type { CandidateQueueRepository } from '../database/candidate-queue-repository.js'; /** Bag of everything the route modules might need. */ export interface WebContext { @@ -54,6 +55,7 @@ export interface WebContext { configStore?: ConfigStore; redditAccounts?: RedditAccountsService; accountRuntimes?: RedditAccountRuntimeManager; + candidateQueue?: CandidateQueueRepository; }; /** Bot pause/resume toggle read by the main loop. */ diff --git a/src/web/routes/analytics.ts b/src/web/routes/analytics.ts index d432678..0fe81f4 100644 --- a/src/web/routes/analytics.ts +++ b/src/web/routes/analytics.ts @@ -46,5 +46,13 @@ export function createAnalyticsRouter(ctx: WebContext): Router { res.json(ctx.repos.aiUsage.getSummary(days)); }); + router.get('/analytics/candidate-funnel', (_req, res) => { + if (!ctx.services.candidateQueue) { + res.status(503).json({ error: 'Candidate funnel is not initialized' }); + return; + } + res.json(ctx.services.candidateQueue.getStats()); + }); + return router; } diff --git a/src/web/server.ts b/src/web/server.ts index 18b3b10..e44ee5c 100644 --- a/src/web/server.ts +++ b/src/web/server.ts @@ -43,6 +43,7 @@ import { resolveDashboardAssets } from './dashboard-assets.js'; import { createAgentRouter, createHealthRouter } from './routes/agent.js'; import { createAttentionRouter } from './routes/attention.js'; import type { AiUsageRepository } from '../database/ai-usage-repository.js'; +import type { CandidateQueueRepository } from '../database/candidate-queue-repository.js'; export type { BotStats, ActivityItem, LogEntry, RuntimeConfig } from './types.js'; @@ -77,7 +78,10 @@ export class WebServer { broadcaster?: LogBroadcaster, private workspaceDirectory = process.cwd(), private aiUsageRepo?: AiUsageRepository, + initialBotRunning = false, + private readonly persistBotRunning: (running: boolean) => void = () => undefined, ) { + this.botRunning = initialBotRunning; this.broadcaster = broadcaster ?? new LogBroadcaster(); this.activity = new ActivityLog(activityLogRepo); this.runtimeConfig = { @@ -113,6 +117,7 @@ export class WebServer { getBotRunning: () => this.botRunning, setBotRunning: (running) => { this.botRunning = running; + this.persistBotRunning(running); }, getMessagesThisHour: () => this.messagesThisHour, getOnConfigChange: () => this.onConfigChange, @@ -228,6 +233,10 @@ export class WebServer { this.services.conversationState = conversationState; } + setCandidateQueueRepository(candidateQueue: CandidateQueueRepository): void { + this.services.candidateQueue = candidateQueue; + } + setConfigStore(configStore: ConfigStore): void { this.services.configStore = configStore; } diff --git a/tests/candidate-queue.test.ts b/tests/candidate-queue.test.ts new file mode 100644 index 0000000..622fb9b --- /dev/null +++ b/tests/candidate-queue.test.ts @@ -0,0 +1,168 @@ +import { afterEach, beforeEach, describe, expect, it } from 'vitest'; +import Database from 'better-sqlite3'; +import { runMigrations } from '../src/database/client.js'; +import { + CandidateQueueRepository, + RuntimeStateRepository, +} from '../src/database/candidate-queue-repository.js'; +import { RecipientHistoryRepository } from '../src/database/repositories.js'; +import type { RedditPost } from '../src/reddit/api.js'; +import type { ProcessedPost } from '../src/services/processor.js'; + +function post(id: string, author: string, created = 100): RedditPost { + return { + id, + subreddit: 'test', + author, + author_fullname: `t2_${author}`, + title: `[25F] ${id}`, + url: `https://example.com/${id}.jpg`, + permalink: `/r/test/${id}`, + thumbnail: '', + is_self: false, + created_utc: created, + over_18: true, + spoiler: false, + }; +} + +function prepared(candidate: RedditPost): ProcessedPost { + return { + post: candidate, + imageAnalysis: 'analysis', + detectedGender: 'female', + generatedMessage: 'prepared message', + strategyId: null, + experimentId: null, + promptVersion: null, + selectionReason: null, + modelId: 'model', + sourceType: 'image', + toneId: 'default', + scripted: false, + }; +} + +describe('CandidateQueueRepository', () => { + let db: Database.Database; + let queue: CandidateQueueRepository; + + beforeEach(() => { + db = new Database(':memory:'); + db.pragma('foreign_keys = ON'); + runMigrations(db); + const insertAccount = db.prepare(` + INSERT INTO reddit_accounts ( + account_id, reddit_username, matrix_user_id, status, is_active, + cookies_json, matrix_credentials_json, created_at, updated_at + ) VALUES (?, ?, ?, 'verified', 1, '[]', '{}', 1, 1) + `); + insertAccount.run('account-a', 'account-a', '@account-a:reddit.com'); + insertAccount.run('account-b', 'account-b', '@account-b:reddit.com'); + queue = new CandidateQueueRepository(db); + }); + + afterEach(() => db.close()); + + it('builds a broad pool while keeping only one active candidate per author', () => { + const result = queue.enqueue([ + post('p1', 'alice'), + post('p2', 'alice'), + post('p3', 'bob'), + ], 'image', 2, 1_000); + + expect(result).toEqual({ + discovered: 2, + duplicatePost: 0, + duplicateAuthor: 1, + previouslyProcessed: 0, + previouslyContacted: 0, + }); + expect(queue.getStats().byStatus.discovered).toBe(2); + }); + + it('excludes posts and recipients already handled anywhere', () => { + db.prepare(` + INSERT INTO processed_posts (account_id, id, subreddit, author, title, processed_at) + VALUES ('__legacy_unassigned__', 'old-post', 'test', 'old', 'title', 1) + `).run(); + db.prepare(`INSERT INTO recipient_suppressions (recipient, reason, created_at) VALUES ('contacted', 'sent', 1)`).run(); + + const result = queue.enqueue([ + post('old-post', 'different'), + post('fresh', 'contacted'), + ], 'image', 1, 2_000); + + expect(result.previouslyProcessed).toBe(1); + expect(result.previouslyContacted).toBe(1); + expect(queue.getStats().total).toBe(0); + }); + + it('recovers screening and sending leases after an interrupted deployment', () => { + queue.enqueue([post('screen', 'alice'), post('send', 'bob')], 'image', 1, 1_000); + const screening = queue.claimForScreening('screen-worker', 100, 1_010)!; + expect(queue.markReady(screening.postId, 'screen-worker', prepared(screening.post), 1_020)).toBe(true); + const sending = queue.claimReady('account-a', 'send-worker', 100, 1_030)!; + expect(sending.status).toBe('sending'); + expect(queue.claimForScreening('other-worker', 100, 1_040)?.status).toBe('screening'); + + expect(queue.recoverExpiredLeases(1_200)).toBe(2); + expect(queue.getStats().byStatus.retryable).toBe(1); + expect(queue.getStats().byStatus.discovered).toBe(1); + }); + + it('retries a rate-limited prepared candidate and preserves its generated message', () => { + queue.enqueue([post('p1', 'alice')], 'image', 1, 1_000); + const screening = queue.claimForScreening('screen', 100, 1_001)!; + queue.markReady('p1', 'screen', prepared(screening.post), 1_002); + const sending = queue.claimReady('account-a', 'send-a', 100, 1_003)!; + queue.deferDelivery('p1', 'send-a', 'Matrix rate limited', 1_100, 1_004); + + expect(queue.claimReady('account-b', 'send-b', 100, 1_099)).toBeNull(); + expect(queue.promoteDueRetries(1_100)).toBe(1); + expect(queue.claimReady('account-b', 'send-b', 100, 1_101)).toMatchObject({ + postId: 'p1', + assignedAccountId: 'account-b', + prepared: expect.objectContaining({ generatedMessage: 'prepared message' }), + }); + }); +}); + +describe('RuntimeStateRepository', () => { + it('persists the operator run state across repository instances', () => { + const db = new Database(':memory:'); + runMigrations(db); + const first = new RuntimeStateRepository(db); + expect(first.isBotRunning()).toBe(false); + first.setBotRunning(true, 100); + expect(new RuntimeStateRepository(db).isBotRunning()).toBe(true); + first.setBotRunning(false, 200); + expect(new RuntimeStateRepository(db).isBotRunning()).toBe(false); + db.close(); + }); +}); + +describe('RecipientHistoryRepository recovery', () => { + it('releases only stale pending claims that have no delivery evidence', () => { + const db = new Database(':memory:'); + runMigrations(db); + const history = new RecipientHistoryRepository(db); + + expect(history.tryClaim('orphaned', 100)).toBe(true); + expect(history.tryClaim('delivered', 100)).toBe(true); + db.prepare(` + INSERT INTO processed_posts (account_id, id, subreddit, author, title, processed_at) + VALUES ('__legacy_unassigned__', 'post-1', 'test', 'delivered', 'title', 125) + `).run(); + db.prepare(` + INSERT INTO sent_messages ( + account_id, post_id, recipient, subject, message, sent_at + ) VALUES ('__legacy_unassigned__', 'post-1', 'delivered', 'subject', 'message', 150) + `).run(); + + expect(history.recoverStaleClaims(200)).toBe(1); + expect(history.hasSentTo('orphaned')).toBe(false); + expect(history.hasSentTo('delivered')).toBe(true); + db.close(); + }); +}); diff --git a/tests/chat-service-cooldown.test.ts b/tests/chat-service-cooldown.test.ts index b349446..354d045 100644 --- a/tests/chat-service-cooldown.test.ts +++ b/tests/chat-service-cooldown.test.ts @@ -31,7 +31,7 @@ function buildService() { path.join(os.tmpdir(), `chat-cache-${Date.now()}-${Math.random()}.json`), ); const svc = new ChatService(matrix, logger, undefined, undefined, undefined, undefined, cache, undefined, undefined, authenticatedAccountAuth); - return { svc, logger }; + return { svc, logger, matrix }; } describe('ChatService invitation cooldown', () => { @@ -82,6 +82,30 @@ describe('ChatService invitation cooldown', () => { expect(result).toBeNull(); }); + it('returns a retryable delivery result with the persisted cooldown deadline', async () => { + const { svc } = buildService(); + svc.setInvitationCooldown( + new MatrixRateLimitError('429', 60_000, 'M_LIMIT_EXCEEDED', 'rate.foo'), + ); + const cooldown = svc.getInvitationCooldown()!; + + await expect(svc.startConversationDetailed('someuser', 'hi')).resolves.toEqual({ + status: 'retryable', + reason: cooldown.reason, + retryAt: cooldown.until, + }); + }); + + it('classifies recipients that reject chat requests as terminal', async () => { + const { svc, matrix } = buildService(); + matrix.sendInitialDirectMessage = async () => { + throw new Error('some of the invited users do not accept chat requests'); + }; + + await expect(svc.startConversationDetailed('someuser', 'hi', 't2_someuser')) + .resolves.toEqual({ status: 'terminal', reason: 'Recipient does not accept chat requests' }); + }); + it('sendChatMessage short-circuits when a cooldown is active', async () => { const { svc } = buildService(); svc.setInvitationCooldown( diff --git a/tests/database.test.ts b/tests/database.test.ts index a257a2a..4ae9141 100644 --- a/tests/database.test.ts +++ b/tests/database.test.ts @@ -107,7 +107,11 @@ describe('Database Client', () => { .prepare("SELECT name FROM sqlite_master WHERE type='table' AND name NOT LIKE 'sqlite_%'") .all() as { name: string }[]; // Existing domain/config tables plus account, conversation, sync and media ownership stores. - expect(tables.length).toBe(16); + expect(tables.length).toBe(18); + expect(tables.map((table) => table.name)).toEqual(expect.arrayContaining([ + 'candidate_queue', + 'runtime_state', + ])); }); it('tracks schema version and migrates a legacy sent_messages table transactionally', () => { @@ -156,7 +160,7 @@ describe('Database Client', () => { DROP TABLE conversation_automation_claims; DROP TABLE ai_usage; `); - db.pragma(`user_version = ${latestVersion - 3}`); + db.pragma(`user_version = ${latestVersion - 4}`); runMigrations(db); expect(db.prepare('SELECT recipient, reason FROM recipient_suppressions ORDER BY recipient COLLATE NOCASE').all()) diff --git a/tests/monitor.test.ts b/tests/monitor.test.ts index 745ef2d..cb0eeda 100644 --- a/tests/monitor.test.ts +++ b/tests/monitor.test.ts @@ -1,5 +1,5 @@ import { describe, it, expect, vi, beforeEach } from 'vitest'; -import { MonitorService } from '../src/services/monitor.js'; +import { MonitorService, parseSelfDeclaredAge } from '../src/services/monitor.js'; import { RedditRateLimitError, RedditSubredditUnavailableError, @@ -24,6 +24,7 @@ function createMockPostsRepo(processedIds: Set = new Set()): PostsReposi isProcessed: vi.fn((id: string) => processedIds.has(id)), isProcessedAnywhere: vi.fn((id: string) => processedIds.has(id)), insert: vi.fn(), + insertIfAbsent: vi.fn(() => true), updateAnalysis: vi.fn(), markMessageSent: vi.fn(), setError: vi.fn(), @@ -168,7 +169,7 @@ describe('MonitorService', () => { expect(redditApi.isSellerProfile).toHaveBeenCalledWith( 'seller', config.reddit.sellerKeywords, config.reddit.sellerDomains, sellerPost, ); - expect(postsRepo.insert).toHaveBeenCalledWith(expect.objectContaining({ id: 't3_seller' })); + expect(postsRepo.insertIfAbsent).toHaveBeenCalledWith(expect.objectContaining({ id: 't3_seller' })); expect(postsRepo.setError).toHaveBeenCalledWith('t3_seller', 'Seller signals detected'); }); @@ -192,7 +193,7 @@ describe('MonitorService', () => { await expect(monitor.getUnprocessedPosts()).resolves.toEqual([selected]); expect(logger.logs).toContainEqual(expect.objectContaining({ level: 'info', - message: 'r/sub1 listing: 4 fetched; 3 screened; 0 already processed; 1 already messaged; 1 seller signals; 0 missing gender tag; 0 gender not targeted; 1 eligible (selected); 0 outside recent window; 1 not evaluated after selected recent candidate', + message: 'r/sub1 listing: 4 fetched; 3 screened; 0 already processed; 1 already messaged; 1 seller signals; 0 missing gender tag; 0 gender not targeted; 0 missing age tag; 0 underage; 1 eligible (selected); 0 outside recent window; 1 not evaluated after selected recent candidate', })); }); @@ -224,7 +225,7 @@ describe('MonitorService', () => { ); }); - it('finishes the seller and gender checks for one candidate before moving to the next', async () => { + it('runs free gender and age checks before seller profile requests', async () => { const newer: RedditPost = { id: 't3_newer', subreddit: 'chat-sub', author: 'newer', title: '[30M] hello', url: '', permalink: '/r/chat-sub/newer', thumbnail: '', is_self: true, @@ -249,11 +250,18 @@ describe('MonitorService', () => { await expect(monitor.getUnprocessedPosts()).resolves.toEqual([eligible]); expect(vi.mocked(redditApi.isSellerProfile).mock.calls.map(([username]) => username)) - .toEqual(['newer', 'eligible']); - expect(postsRepo.insert).toHaveBeenCalledWith(expect.objectContaining({ id: 't3_newer' })); + .toEqual(['eligible']); + expect(postsRepo.insertIfAbsent).toHaveBeenCalledWith(expect.objectContaining({ id: 't3_newer' })); expect(postsRepo.setError).toHaveBeenCalledWith('t3_newer', 'Gender tag "male" not targeted'); }); + it('parses adult subreddit age tags without treating unrelated numbers as ages', () => { + expect(parseSelfDeclaredAge('28 [F4M] Online')).toBe(28); + expect(parseSelfDeclaredAge('[25F] hello')).toBe(25); + expect(parseSelfDeclaredAge('F 31 looking to chat')).toBe(31); + expect(parseSelfDeclaredAge('F4M, 100 percent bored')).toBeNull(); + }); + it('tries next subreddit when first has no valid posts', async () => { const post1: RedditPost = { id: 't3_1', subreddit: 'sub2', author: 'user1', title: 'Post',