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, physiqueScore: number | null = 5): ProcessedPost { return { post: candidate, imageAnalysis: 'analysis', detectedGender: 'female', attractivenessScore: 64, physiqueScore, physiqueTargeting: { enabled: true, minScore: 3, maxScore: 7 }, sourceImageUrl: 'https://i.redd.it/example.jpg', 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, deferredAtCapacity: 0, }); expect(queue.getStats().byStatus.discovered).toBe(2); }); it('fills only the requested intake deficit and reports untouched posts', () => { const result = queue.enqueue([ post('p1', 'alice'), post('p2', 'bob'), post('p3', 'carol'), ], 'image', 1, 1_000, 2); expect(result.discovered).toBe(2); expect(result.deferredAtCapacity).toBe(1); expect(queue.countActive()).toBe(2); }); it('admits, screens, and sends the newest posts before older higher-priority posts', () => { queue.enqueue([ post('old-high-priority', 'alice', 100), post('new-low-priority', 'bob', 300), ], 'image', 1, 1_000, 1); expect(queue.listActive().map((candidate) => candidate.postId)).toEqual(['new-low-priority']); queue.enqueue([post('old-high-priority', 'alice', 100)], 'image', 10, 1_001); const newestScreening = queue.claimForScreening('screen-new', 100, 1_010)!; expect(newestScreening.postId).toBe('new-low-priority'); queue.markReady(newestScreening.postId, 'screen-new', prepared(newestScreening.post), 1_011); const olderScreening = queue.claimForScreening('screen-old', 100, 1_012)!; queue.markReady(olderScreening.postId, 'screen-old', prepared(olderScreening.post), 1_013); expect(queue.claimReady('account-a', 'send', 100, 1_014)?.postId).toBe('new-low-priority'); }); it('promotes the newest due delivery retry when ready capacity is limited', () => { queue.enqueue([post('new-low-priority', 'bob', 300)], 'image', 1, 1_000); queue.enqueue([post('old-high-priority', 'alice', 100)], 'image', 10, 1_001); for (const worker of ['screen-new', 'screen-old']) { const candidate = queue.claimForScreening(worker, 100, 1_010)!; queue.markReady(candidate.postId, worker, prepared(candidate.post), 1_011); } for (const [accountId, worker] of [['account-a', 'send-new'], ['account-b', 'send-old']] as const) { const candidate = queue.claimReady(accountId, worker, 100, 1_012)!; queue.deferDelivery(candidate.postId, worker, 'Matrix rate limited', 1_100, 1_013); } expect(queue.promoteDueRetries(1, 1_100)).toBe(1); expect(queue.claimReady('account-a', 'retry', 100, 1_101)?.postId).toBe('new-low-priority'); }); 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('manually clears every unsent stage while retaining an in-flight send', () => { queue.enqueue([post('sending', 'alice')], 'image', 1, 1_000); const sendingScreen = queue.claimForScreening('screen-sending', 1_000, 1_001)!; queue.markReady(sendingScreen.postId, 'screen-sending', prepared(sendingScreen.post), 1_002); queue.claimReady('account-a', 'send-worker', 1_000, 1_003); queue.enqueue([post('retryable', 'bob')], 'image', 1, 1_004); const retryScreen = queue.claimForScreening('screen-retry', 1_000, 1_005)!; queue.markReady(retryScreen.postId, 'screen-retry', prepared(retryScreen.post), 1_006); queue.claimReady('account-b', 'retry-worker', 1_000, 1_007); queue.deferDelivery('retryable', 'retry-worker', 'rate limited', 2_000, 1_008); queue.enqueue([post('ready', 'carol')], 'image', 1, 1_009); const readyScreen = queue.claimForScreening('screen-ready', 1_000, 1_010)!; queue.markReady(readyScreen.postId, 'screen-ready', prepared(readyScreen.post), 1_011); queue.enqueue([post('screening', 'dana')], 'image', 1, 1_012); queue.claimForScreening('screen-active', 1_000, 1_013); queue.enqueue([post('discovered', 'erin')], 'image', 1, 1_014); expect(queue.clearUnsent(1_020)).toEqual({ discovered: 1, screening: 1, ready: 1, retryable: 1, total: 4, retainedSending: 1, }); expect(queue.listActive()).toEqual([ expect.objectContaining({ postId: 'sending', status: 'sending', leaseOwner: 'send-worker' }), ]); expect(queue.markReady('screening', 'screen-active', prepared(post('screening', 'dana')), 1_021)).toBe(false); expect(queue.markSent('sending', 'send-worker', 1_022)).toBe(true); expect(queue.getStats()).toMatchObject({ byStatus: { rejected: 4, sent: 1 } }); const reasons = db.prepare(`SELECT DISTINCT last_error FROM candidate_queue WHERE status = 'rejected'`) .all() as Array<{ last_error: string }>; expect(reasons).toEqual([{ last_error: 'Cleared manually by operator' }]); }); 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(25, 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' }), }); }); it('reconciles ready and retryable images when the physique range changes', () => { queue.enqueue([ post('inside', 'alice', 300), post('outside', 'bob', 200), post('retry', 'carol', 100), ], 'image', 1, 1_000); for (const worker of ['screen-1', 'screen-2', 'screen-3']) { const candidate = queue.claimForScreening(worker, 100, 1_010)!; const score = candidate.postId === 'outside' ? 7 : 4; queue.markReady(candidate.postId, worker, prepared(candidate.post, score), 1_011); } const retry = queue.claimReady('account-a', 'send', 100, 1_012)!; expect(retry.postId).toBe('inside'); queue.deferDelivery(retry.postId, 'send', 'rate limited', 2_000, 1_013); expect(queue.reconcilePhysiqueTargeting( { enabled: true, minScore: 1, maxScore: 6 }, 1_020, )).toEqual({ retained: 2, rejected: 1 }); const active = queue.listActive(); expect(active.map((candidate) => candidate.postId)).toEqual(['inside', 'retry']); expect(active.every((candidate) => ( candidate.prepared?.physiqueTargeting?.enabled === true && candidate.prepared.physiqueTargeting.maxScore === 6 ))).toBe(true); expect(queue.getStats().byStatus.rejected).toBe(1); }); it('fails closed for an unscored prepared image but does not touch a leased send', () => { queue.enqueue([ post('unscored', 'alice', 200), post('sending', 'bob', 100), ], 'image', 1, 1_000); for (const worker of ['screen-1', 'screen-2']) { const candidate = queue.claimForScreening(worker, 100, 1_010)!; queue.markReady(candidate.postId, worker, prepared(candidate.post, null), 1_011); } const sending = queue.claimReady('account-a', 'send', 100, 1_012)!; expect(queue.reconcilePhysiqueTargeting( { enabled: true, minScore: 1, maxScore: 6 }, 1_020, )).toEqual({ retained: 0, rejected: 1 }); expect(queue.listActive()).toEqual([ expect.objectContaining({ postId: sending.postId, status: 'sending' }), ]); }); it('lists only active candidates in newest-post order across stages', () => { queue.enqueue([ post('waiting', 'alice', 100), post('screening', 'bob', 500), post('ready', 'carol', 400), post('sending', 'dana', 300), post('rejected', 'erin', 200), ], 'image', 1, 1_000); queue.claimForScreening('worker-1', 1_000, 1_010); const ready = queue.claimForScreening('worker-2', 1_000, 1_020)!; queue.markReady(ready.postId, 'worker-2', prepared(ready.post), 1_021); const sending = queue.claimReady('account-a', 'worker-3', 1_000, 1_030)!; const rejected = queue.claimForScreening('worker-4', 1_000, 1_040)!; queue.reject(rejected.postId, 'worker-4', 'Seller signals detected', 1_041); const active = queue.listActive(); expect(active.map((candidate) => candidate.postId)).toEqual([ 'screening', 'ready', 'rejected', 'waiting', ]); expect(queue.listActive().map((candidate) => candidate.postId)).not.toContain(rejected.postId); expect(queue.getOldestActiveAt()).toBe(1_000); }); }); 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(); }); });