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; } 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 CASE status WHEN 'sending' THEN 1 WHEN 'screening' THEN 2 WHEN 'ready' THEN 3 WHEN 'retryable' THEN 4 ELSE 5 END, priority DESC, post_created_at DESC, discovered_at ASC LIMIT ? `).all(...ACTIVE_STATUSES, 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 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 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 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); } }