435 lines
19 KiB
TypeScript
435 lines
19 KiB
TypeScript
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('rejects active commenter candidates admitted by the legacy role policy', () => {
|
|
const legacyComment = {
|
|
...post('comment-legacy', 'alice', 300),
|
|
targetComment: {
|
|
body: 'I wish that was me', permalink: '/r/test/comment-legacy/c1',
|
|
sourcePostId: 'source', sourcePostAuthor: 'source-author',
|
|
expressedGender: 'unknown' as const, confidence: 0.93,
|
|
},
|
|
};
|
|
const currentComment = {
|
|
...post('comment-current', 'bob', 200),
|
|
targetComment: {
|
|
body: 'I want that too', permalink: '/r/test/comment-current/c2',
|
|
sourcePostId: 'source', sourcePostAuthor: 'source-author',
|
|
expressedGender: 'unknown' as const, confidence: 0.91, classifierVersion: 'recipient-role-v2' as const,
|
|
},
|
|
};
|
|
const editedComment = {
|
|
...post('comment-edited', 'dana', 150),
|
|
targetComment: {
|
|
body: 'I would love that', permalink: '/r/test/comment-edited/c3',
|
|
sourcePostId: 'source', sourcePostAuthor: 'source-author',
|
|
expressedGender: 'unknown' as const, confidence: 0.9,
|
|
},
|
|
};
|
|
queue.enqueue([legacyComment, currentComment, editedComment, post('post-author', 'carol', 100)], 'image', 1, 1_000);
|
|
for (let index = 0; index < 4; index += 1) {
|
|
const worker = `worker-${index}`;
|
|
const candidate = queue.claimForScreening(worker, 100, 1_010)!;
|
|
const contract = candidate.postId === 'comment-current'
|
|
? 'comment-author-v2'
|
|
: candidate.postId === 'post-author' ? 'post-author-v1' : undefined;
|
|
queue.markReady(candidate.postId, worker, {
|
|
...prepared(candidate.post), openerContract: contract,
|
|
operatorEdited: candidate.postId === 'comment-edited',
|
|
}, 1_011);
|
|
}
|
|
|
|
expect(queue.rejectLegacyCommentCandidates(1_020)).toBe(2);
|
|
expect(queue.listActive()).toEqual(expect.arrayContaining([
|
|
expect.objectContaining({ postId: 'comment-current', status: 'ready' }),
|
|
expect.objectContaining({ postId: 'post-author', status: 'ready' }),
|
|
]));
|
|
const activeIds = queue.listActive().map((candidate) => candidate.postId);
|
|
expect(activeIds).not.toContain('comment-legacy');
|
|
expect(activeIds).not.toContain('comment-edited');
|
|
});
|
|
|
|
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('lists the latest 50 rejected candidates by their terminal update time', () => {
|
|
queue.enqueue([post('older', 'alice'), post('newer', 'bob')], 'image', 1, 1_000);
|
|
const older = queue.claimForScreening('older-worker', 1_000, 1_001)!;
|
|
queue.reject(older.postId, 'older-worker', 'No gender tag in title/body', 1_002);
|
|
const newer = queue.claimForScreening('newer-worker', 1_000, 1_003)!;
|
|
queue.reject(newer.postId, 'newer-worker', 'Seller signals detected', 1_004);
|
|
queue.enqueue([post('manual', 'carol')], 'image', 1, 1_005);
|
|
queue.cancelUnsent('manual', 1_006);
|
|
|
|
expect(queue.listRecentRejected()).toEqual([
|
|
expect.objectContaining({ postId: 'newer', lastError: 'Seller signals detected', updatedAt: 1_004 }),
|
|
expect.objectContaining({ postId: 'older', lastError: 'No gender tag in title/body', updatedAt: 1_002 }),
|
|
]);
|
|
});
|
|
|
|
it('edits only prepared unsent messages and clears generated attribution', () => {
|
|
queue.enqueue([post('editable', 'alice', 200), post('waiting', 'bob', 100)], 'image', 1, 1_000);
|
|
const editable = queue.claimForScreening('screen', 1_000, 1_001)!;
|
|
queue.markReady(editable.postId, 'screen', prepared(editable.post), 1_002);
|
|
db.prepare(`
|
|
INSERT INTO processed_posts (account_id, id, subreddit, author, title, generated_message, processed_at)
|
|
VALUES ('account-a', ?, 'test', 'alice', 'title', 'prepared message', 1)
|
|
`).run(editable.postId);
|
|
|
|
expect(queue.updatePreparedMessage(editable.postId, 'Operator-written opener', 1_003)).toBe('updated');
|
|
expect(queue.listActive().find((candidate) => candidate.postId === editable.postId)?.prepared).toMatchObject({
|
|
generatedMessage: 'Operator-written opener',
|
|
strategyId: null,
|
|
experimentId: null,
|
|
promptVersion: null,
|
|
selectionReason: null,
|
|
modelId: 'operator-edited',
|
|
operatorEdited: true,
|
|
});
|
|
expect((db.prepare(`SELECT generated_message FROM processed_posts WHERE id = ?`).get(editable.postId) as {
|
|
generated_message: string;
|
|
}).generated_message).toBe('Operator-written opener');
|
|
expect(queue.updatePreparedMessage('waiting', 'Nope', 1_004)).toBe('not-editable');
|
|
expect(queue.updatePreparedMessage('missing', 'Nope', 1_004)).toBe('not-found');
|
|
|
|
queue.claimReady('account-a', 'send', 1_000, 1_005);
|
|
expect(queue.updatePreparedMessage(editable.postId, 'Too late', 1_006)).toBe('not-editable');
|
|
});
|
|
|
|
it('removes one unsent candidate but refuses to cancel an in-flight send', () => {
|
|
queue.enqueue([post('remove', 'alice')], 'image', 1, 1_000);
|
|
expect(queue.cancelUnsent('remove', 1_001)).toBe('cancelled');
|
|
expect(queue.getStats()).toMatchObject({ byStatus: { rejected: 1 } });
|
|
expect(queue.cancelUnsent('remove', 1_002)).toBe('not-cancellable');
|
|
expect(queue.cancelUnsent('missing', 1_002)).toBe('not-found');
|
|
|
|
queue.enqueue([post('sending', 'bob')], 'image', 1, 1_003);
|
|
const candidate = queue.claimForScreening('screen', 1_000, 1_004)!;
|
|
queue.markReady(candidate.postId, 'screen', prepared(candidate.post), 1_005);
|
|
queue.claimReady('account-a', 'send', 1_000, 1_006);
|
|
expect(queue.cancelUnsent('sending', 1_007)).toBe('not-cancellable');
|
|
expect(queue.listActive()).toEqual([expect.objectContaining({ postId: 'sending', status: 'sending' })]);
|
|
});
|
|
|
|
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();
|
|
});
|
|
});
|