import type Database from 'better-sqlite3'; import type { PhysiqueTargeting } from '../config/index.js'; import type { RedditPost } from '../reddit/api.js'; import { allowsSending, parseBotMode, type BotMode } from '../services/bot-mode.js'; import { evaluatePhysiqueTargeting } from '../services/physique-targeting.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; deferredAtCapacity: number; } export interface PhysiqueTargetingReconciliation { rejected: number; retained: number; } export interface CandidateQueueClearResult { discovered: number; screening: number; ready: number; retryable: number; total: number; retainedSending: number; } export type CandidateEditResult = 'updated' | 'not-found' | 'not-editable'; export type CandidateCancelResult = 'cancelled' | 'not-found' | 'not-cancellable'; 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(), maxNewCandidates: number = Number.POSITIVE_INFINITY, ): CandidateEnqueueResult { if ((!Number.isInteger(maxNewCandidates) && maxNewCandidates !== Number.POSITIVE_INFINITY) || maxNewCandidates < 0) { throw new Error('Candidate intake capacity must be a non-negative integer'); } const result: CandidateEnqueueResult = { discovered: 0, duplicatePost: 0, duplicateAuthor: 0, previouslyProcessed: 0, previouslyContacted: 0, deferredAtCapacity: 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', ?, ?, ?) `); const newestPostsFirst = [...posts].sort((left, right) => right.created_utc - left.created_utc); this.db.transaction(() => { for (const [index, post] of newestPostsFirst.entries()) { if (result.discovered >= maxNewCandidates) { result.deferredAtCapacity += newestPostsFirst.length - index; break; } 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(maxReadyCandidates: number, now: number = Date.now()): number { if (!Number.isInteger(maxReadyCandidates) || maxReadyCandidates < 1) { throw new Error('Maximum ready candidates must be a positive integer'); } const capacity = maxReadyCandidates - this.countReady(); if (capacity <= 0) return 0; return this.db.prepare(` UPDATE candidate_queue SET status = 'ready', updated_at = ?, assigned_account_id = NULL WHERE post_id IN ( SELECT post_id FROM candidate_queue WHERE status = 'retryable' AND available_at <= ? ORDER BY post_created_at DESC, priority DESC, discovered_at ASC LIMIT ? ) `).run(now, now, capacity).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; } /** Rejects queued comments admitted by a broader historical role policy. */ rejectLegacyCommentCandidates(now: number = Date.now()): number { const rows = this.db.prepare(` SELECT post_id, payload_json FROM candidate_queue WHERE status IN ('discovered', 'screening', 'ready', 'retryable') `).all() as Array<{ post_id: string; payload_json: string }>; const update = this.db.prepare(` UPDATE candidate_queue SET status = 'rejected', updated_at = ?, lease_owner = NULL, lease_expires_at = NULL, assigned_account_id = NULL, last_error = 'Rejected after commenter recipient-role policy upgrade' WHERE post_id = ? AND status IN ('discovered', 'screening', 'ready', 'retryable') `); return this.db.transaction(() => { let rejected = 0; for (const row of rows) { const post = JSON.parse(row.payload_json) as RedditPost; if (!post.targetComment || post.targetComment.classifierVersion === 'recipient-role-v2') continue; rejected += update.run(now, row.post_id).changes; } return rejected; })(); } /** * Cancels every candidate that cannot already be crossing the remote-send * boundary. Sending leases are deliberately retained because their Matrix * outcome may be unknown while this transaction runs. */ clearUnsent(now: number = Date.now()): CandidateQueueClearResult { return this.db.transaction(() => { const statuses = ['discovered', 'screening', 'ready', 'retryable'] as const; const counts = Object.fromEntries(statuses.map((status) => [status, 0])) as Record<(typeof statuses)[number], number>; const rows = this.db.prepare(` SELECT status, COUNT(*) AS count FROM candidate_queue WHERE status IN (${statuses.map(() => '?').join(', ')}) GROUP BY status `).all(...statuses) as Array<{ status: (typeof statuses)[number]; count: number }>; for (const row of rows) counts[row.status] = row.count; const retainedSending = (this.db.prepare( `SELECT COUNT(*) AS count FROM candidate_queue WHERE status = 'sending'`, ).get() as { count: number }).count; const changed = this.db.prepare(` UPDATE candidate_queue SET status = 'rejected', last_error = 'Cleared manually by operator', updated_at = ?, lease_owner = NULL, lease_expires_at = NULL, assigned_account_id = NULL WHERE status IN (${statuses.map(() => '?').join(', ')}) `).run(now, ...statuses).changes; return { ...counts, total: changed, retainedSending }; })(); } updatePreparedMessage(postId: string, message: string, now: number = Date.now()): CandidateEditResult { return this.db.transaction(() => { const row = this.db.prepare(`SELECT status, prepared_json FROM candidate_queue WHERE post_id = ?`) .get(postId) as { status: CandidateStatus; prepared_json: string | null } | undefined; if (!row) return 'not-found'; if (!['ready', 'retryable'].includes(row.status) || !row.prepared_json) return 'not-editable'; const prepared = JSON.parse(row.prepared_json) as ProcessedPost; const edited: ProcessedPost = { ...prepared, generatedMessage: message, strategyId: null, experimentId: null, promptVersion: null, selectionReason: null, modelId: 'operator-edited', operatorEdited: true, }; const changed = this.db.prepare(` UPDATE candidate_queue SET prepared_json = ?, updated_at = ? WHERE post_id = ? AND status IN ('ready', 'retryable') `).run(JSON.stringify(edited), now, postId).changes; if (changed === 1) { this.db.prepare(`UPDATE processed_posts SET generated_message = ? WHERE id = ?`).run(message, postId); } return changed === 1 ? 'updated' : 'not-editable'; })(); } cancelUnsent(postId: string, now: number = Date.now()): CandidateCancelResult { return this.db.transaction(() => { const row = this.db.prepare(`SELECT status FROM candidate_queue WHERE post_id = ?`) .get(postId) as { status: CandidateStatus } | undefined; if (!row) return 'not-found'; if (!['discovered', 'screening', 'ready', 'retryable'].includes(row.status)) return 'not-cancellable'; const changed = this.db.prepare(` UPDATE candidate_queue SET status = 'rejected', last_error = 'Removed manually by operator', updated_at = ?, lease_owner = NULL, lease_expires_at = NULL, assigned_account_id = NULL WHERE post_id = ? AND status IN ('discovered', 'screening', 'ready', 'retryable') `).run(now, postId).changes; return changed === 1 ? 'cancelled' : 'not-cancellable'; })(); } /** Re-evaluates prepared image candidates that have not yet been leased for sending. */ reconcilePhysiqueTargeting( targeting: PhysiqueTargeting, now: number = Date.now(), ): PhysiqueTargetingReconciliation { const rows = this.db.prepare(` SELECT post_id, prepared_json FROM candidate_queue WHERE source_type = 'image' AND status IN ('ready', 'retryable') AND prepared_json IS NOT NULL `).all() as Array<{ post_id: string; prepared_json: string }>; const preparedRows = rows.map((row) => ({ postId: row.post_id, prepared: JSON.parse(row.prepared_json) as ProcessedPost, })); const reject = 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 ('ready', 'retryable') `); const retain = this.db.prepare(` UPDATE candidate_queue SET prepared_json = ?, updated_at = ? WHERE post_id = ? AND status IN ('ready', 'retryable') `); return this.db.transaction(() => { const result: PhysiqueTargetingReconciliation = { rejected: 0, retained: 0 }; for (const row of preparedRows) { const score = typeof row.prepared.physiqueScore === 'number' ? row.prepared.physiqueScore : null; const decision = evaluatePhysiqueTargeting(score, targeting); if (!decision.accepted) { result.rejected += reject.run(decision.reason, now, row.postId).changes; continue; } const refreshed: ProcessedPost = { ...row.prepared, physiqueTargeting: { ...targeting }, }; result.retained += 1; const previous = row.prepared.physiqueTargeting; const snapshotMatches = previous?.enabled === targeting.enabled && previous.minScore === targeting.minScore && previous.maxScore === targeting.maxScore; if (!snapshotMatches) retain.run(JSON.stringify(refreshed), now, row.postId); } return result; })(); } 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; } getOldestActiveAt(): number | null { const row = this.db.prepare(` SELECT MIN(discovered_at) AS at FROM candidate_queue WHERE status IN (${ACTIVE_STATUSES.map(() => '?').join(', ')}) `).get(...ACTIVE_STATUSES) as { at: number | null }; return row.at; } 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, }; } listActive(limit = 500): CandidateQueueRecord[] { if (!Number.isInteger(limit) || limit < 1 || limit > 500) { throw new Error('Candidate queue limit must be an integer between 1 and 500'); } const rows = this.db.prepare(` SELECT * FROM candidate_queue WHERE status IN (${ACTIVE_STATUSES.map(() => '?').join(', ')}) ORDER BY post_created_at DESC, priority DESC, discovered_at ASC LIMIT ? `).all(...ACTIVE_STATUSES, limit) as Array>; return rows.map((row) => this.mapRow(row)); } /** Most recently completed screening rejections for the operator audit stream. */ listRecentRejected(limit = 50): CandidateQueueRecord[] { if (!Number.isInteger(limit) || limit < 1 || limit > 50) { throw new Error('Rejected candidate limit must be an integer between 1 and 50'); } const rows = this.db.prepare(` SELECT * FROM candidate_queue WHERE status = 'rejected' AND COALESCE(last_error, '') NOT IN ('Cleared manually by operator', 'Removed manually by operator') ORDER BY updated_at DESC, post_created_at DESC LIMIT ? `).all(limit) as Array>; return rows.map((row) => this.mapRow(row)); } 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 post_created_at DESC, priority 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 this.mapRow(row); } private mapRow(row: Record): CandidateQueueRecord { 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 run/scan/stop state across container restarts. */ export class RuntimeStateRepository { constructor(private readonly db: Database.Database) {} getBotMode(): BotMode { const row = this.db.prepare('SELECT bot_running, bot_mode FROM runtime_state WHERE id = 1') .get() as { bot_running: number; bot_mode: string | null } | undefined; if (!row) return 'stopped'; // `bot_mode` is null for rows written by an older build; fall back to the boolean. return parseBotMode(row.bot_mode, row.bot_running === 1 ? 'running' : 'stopped'); } setBotMode(mode: BotMode, now: number = Date.now()): void { this.db.prepare(` INSERT INTO runtime_state (id, bot_running, bot_mode, updated_at) VALUES (1, ?, ?, ?) ON CONFLICT(id) DO UPDATE SET bot_running = excluded.bot_running, bot_mode = excluded.bot_mode, updated_at = excluded.updated_at `).run(allowsSending(mode) ? 1 : 0, mode, now); } isBotRunning(): boolean { return allowsSending(this.getBotMode()); } setBotRunning(running: boolean, now: number = Date.now()): void { this.setBotMode(running ? 'running' : 'stopped', now); } }