RedditBot/src/database/candidate-queue-repository.ts

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<CandidateStatus, number>;
  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;
}

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;
  }

  /** 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<CandidateStatus, number>;
    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<Record<string, unknown>>;
    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<string, unknown> | undefined;
    if (!row) return null;
    return this.mapRow(row);
  }

  private mapRow(row: Record<string, unknown>): 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);
  }
}