diff --git a/apps/dashboard/src/features/settings/settings-model.ts b/apps/dashboard/src/features/settings/settings-model.ts index c430dd1..78bdb84 100644 --- a/apps/dashboard/src/features/settings/settings-model.ts +++ b/apps/dashboard/src/features/settings/settings-model.ts @@ -26,6 +26,7 @@ export interface SettingsDraft { dryRun: boolean; prioritizeRecentPosts: boolean; recentWindowMinutes: string; + maxReadyCandidates: string; subreddits: SubredditDraft[]; advancedConfigJson: string; } @@ -53,6 +54,7 @@ const ValidatedSettingsSchema = z.strictObject({ dryRun: z.boolean(), prioritizeRecentPosts: z.boolean(), recentWindowMinutes: z.number().finite().positive('Recent window must be greater than 0 minutes.'), + maxReadyCandidates: z.number().int().min(1, 'Ready-post limit must be at least 1.').max(500, 'Ready-post limit cannot exceed 500.'), subreddits: z.array(EditableSubredditSchema).min(1, 'At least one subreddit is required.'), }).superRefine((value, context) => { const names = value.subreddits.map((subreddit) => subreddit.name.trim().toLowerCase()); @@ -69,6 +71,7 @@ export function createSettingsDraft(config: ConfigResponse): SettingsDraft { dryRun: config.browser.dryRun, prioritizeRecentPosts: config.reddit.prioritizeRecentPosts, recentWindowMinutes: String(config.reddit.recentWindowMinutes), + maxReadyCandidates: String(config.reddit.maxReadyCandidates), subreddits: config.subreddits.map((subreddit) => ({ name: subreddit.name, enabled: subreddit.enabled, @@ -90,6 +93,7 @@ export function validateSettingsDraft(draft: SettingsDraft): SettingsValidationR if (draft.maxMessagesPerHour.trim() === '') requiredErrors.maxMessagesPerHour = 'Maximum messages per hour is required.'; if (draft.jitterSeconds.trim() === '') requiredErrors.jitterSeconds = 'Jitter is required.'; if (draft.recentWindowMinutes.trim() === '') requiredErrors.recentWindowMinutes = 'Recent window is required.'; + if (draft.maxReadyCandidates.trim() === '') requiredErrors.maxReadyCandidates = 'Ready-post limit is required.'; draft.subreddits.forEach((subreddit, index) => { if (subreddit.name.trim() === '') requiredErrors[`subreddits.${index}.name`] = 'Subreddit name is required.'; if (subreddit.weight.trim() !== '' && Number(subreddit.weight) <= 0) { @@ -125,6 +129,7 @@ export function validateSettingsDraft(draft: SettingsDraft): SettingsValidationR dryRun: draft.dryRun, prioritizeRecentPosts: draft.prioritizeRecentPosts, recentWindowMinutes: Number(draft.recentWindowMinutes), + maxReadyCandidates: Number(draft.maxReadyCandidates), subreddits: draft.subreddits.map((subreddit) => ({ name: subreddit.name.trim(), enabled: subreddit.enabled, @@ -165,6 +170,7 @@ export function validateSettingsDraft(draft: SettingsDraft): SettingsValidationR reddit: { prioritizeRecentPosts: parsed.data.prioritizeRecentPosts, recentWindowMinutes: parsed.data.recentWindowMinutes, + maxReadyCandidates: parsed.data.maxReadyCandidates, }, subreddits: parsed.data.subreddits, ...(advanced ? { advanced } : {}), diff --git a/apps/dashboard/src/features/settings/settings-sections.tsx b/apps/dashboard/src/features/settings/settings-sections.tsx index 5fcf2af..ffd4bcb 100644 --- a/apps/dashboard/src/features/settings/settings-sections.tsx +++ b/apps/dashboard/src/features/settings/settings-sections.tsx @@ -58,6 +58,14 @@ export function MonitoringSection({ draft, errors, onChange }: SectionProps) { onChange={(recentWindowMinutes) => onChange({ ...draft, recentWindowMinutes })} value={draft.recentWindowMinutes} /> + onChange({ ...draft, maxReadyCandidates })} + value={draft.maxReadyCandidates} + /> ); } diff --git a/config.yaml b/config.yaml index d8a245e..2d3b5da 100644 --- a/config.yaml +++ b/config.yaml @@ -48,6 +48,7 @@ reddit: filterSellers: true # Skip profiles that appear to be OnlyFans/content sellers prioritizeRecentPosts: true # Prefer posts within recentWindowMinutes; older posts only as fallback recentWindowMinutes: 30 + maxReadyCandidates: 25 # Pause screening when this many posts are ready to send # sellerKeywords: [...] # Uncomment to override default keyword list openrouter: diff --git a/packages/api-contracts/src/config.ts b/packages/api-contracts/src/config.ts index 4a37cde..25968f1 100644 --- a/packages/api-contracts/src/config.ts +++ b/packages/api-contracts/src/config.ts @@ -21,6 +21,7 @@ export const RateLimitConfigSchema = z.strictObject({ export const RuntimeRedditConfigSchema = z.strictObject({ prioritizeRecentPosts: z.boolean(), recentWindowMinutes: z.number(), + maxReadyCandidates: z.number().int().min(1).max(500), }); export const RuntimeConfigSchema = z.strictObject({ rateLimit: RateLimitConfigSchema, diff --git a/packages/api-contracts/tests/contracts.test.ts b/packages/api-contracts/tests/contracts.test.ts index 71c6350..3afd21f 100644 --- a/packages/api-contracts/tests/contracts.test.ts +++ b/packages/api-contracts/tests/contracts.test.ts @@ -148,7 +148,7 @@ describe('dashboard endpoint contracts', () => { const payload = { rateLimit: { minDelayBetweenMessagesMs: 300_000, maxMessagesPerHour: 8, jitterMs: 60_000 }, browser: { dryRun: false }, - reddit: { prioritizeRecentPosts: true, recentWindowMinutes: 60 }, + reddit: { prioritizeRecentPosts: true, recentWindowMinutes: 60, maxReadyCandidates: 25 }, subreddits: [{ name: 'example', enabled: true, type: 'chat', weight: 2, tone: 'direct' }], tones: ['default', 'direct'], }; diff --git a/src/config/index.ts b/src/config/index.ts index 4eea306..05188c1 100644 --- a/src/config/index.ts +++ b/src/config/index.ts @@ -50,6 +50,7 @@ export const ConfigSchema = z.object({ filterSellers: z.boolean().default(true), prioritizeRecentPosts: z.boolean().default(false), recentWindowMinutes: z.number().positive().default(60), + maxReadyCandidates: z.number().int().min(1).max(500).default(25), sellerKeywords: z .array(z.string()) .default([ diff --git a/src/database/candidate-queue-repository.ts b/src/database/candidate-queue-repository.ts index 7b523aa..1f2a9f9 100644 --- a/src/database/candidate-queue-repository.ts +++ b/src/database/candidate-queue-repository.ts @@ -191,12 +191,22 @@ export class CandidateQueueRepository { `).run(now, postId, workerId).changes === 1; } - promoteDueRetries(now: number = Date.now()): number { + 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 status = 'retryable' AND available_at <= ? - `).run(now, now).changes; + WHERE post_id IN ( + SELECT post_id FROM candidate_queue + WHERE status = 'retryable' AND available_at <= ? + ORDER BY priority DESC, discovered_at ASC + LIMIT ? + ) + `).run(now, now, capacity).changes; } recoverExpiredLeases(now: number = Date.now(), sendingDelayMs = 10 * 60_000): number { diff --git a/src/main-loop.ts b/src/main-loop.ts index 4a69115..442efa6 100644 --- a/src/main-loop.ts +++ b/src/main-loop.ts @@ -429,7 +429,6 @@ async function runConcurrentAccountRuntimes(app: AppContainer): Promise { const workerInstance = `candidate-funnel-${process.pid}-${Date.now()}`; const leaseMs = 5 * 60_000; const maxActiveCandidates = 500; - const readyTarget = 100; const recovered = candidateQueueRepo.recoverAllLeases(); app.recipientDeduplicator.recoverStaleReservations(Date.now() - 10 * 60_000); logger.info( @@ -527,7 +526,7 @@ async function runConcurrentAccountRuntimes(app: AppContainer): Promise { candidateQueueRepo.expireStale( Date.now() - config.reddit.recentWindowMinutes * 60_000, ); - if (candidateQueueRepo.countReady() >= readyTarget) { + if (candidateQueueRepo.countReady() >= config.reddit.maxReadyCandidates) { await sleep(2_000, controller.signal).catch(() => undefined); continue; } @@ -590,7 +589,7 @@ async function runConcurrentAccountRuntimes(app: AppContainer): Promise { await sleep(1_000, controller.signal).catch(() => undefined); continue; } - candidateQueueRepo.promoteDueRetries(); + candidateQueueRepo.promoteDueRetries(config.reddit.maxReadyCandidates); candidateQueueRepo.recoverExpiredLeases(); app.recipientDeduplicator.recoverStaleReservations(Date.now() - 10 * 60_000); const recipients = await availableRecipients(); diff --git a/src/web/routes/config.ts b/src/web/routes/config.ts index 2416c5a..feb2124 100644 --- a/src/web/routes/config.ts +++ b/src/web/routes/config.ts @@ -10,6 +10,7 @@ const AdvancedConfigSchema = z.strictObject({ subreddits: true, prioritizeRecentPosts: true, recentWindowMinutes: true, + maxReadyCandidates: true, }), openrouter: ConfigSchema.shape.openrouter.omit({ apiKey: true }), browser: z.strictObject({ sessionDir: z.string() }), @@ -19,7 +20,13 @@ const AdvancedConfigSchema = z.strictObject({ }); function advancedConfig(ctx: WebContext): z.infer { - const { subreddits: _subreddits, prioritizeRecentPosts: _prioritize, recentWindowMinutes: _window, ...reddit } = ctx.config.reddit; + const { + subreddits: _subreddits, + prioritizeRecentPosts: _prioritize, + recentWindowMinutes: _window, + maxReadyCandidates: _maxReady, + ...reddit + } = ctx.config.reddit; const { apiKey: _apiKey, ...openrouter } = ctx.config.openrouter; const { dryRun: _dryRun, ...browser } = ctx.config.browser; return { @@ -98,6 +105,12 @@ export function createConfigRouter(ctx: WebContext): Router { if (typeof updates.reddit.recentWindowMinutes === 'number' && updates.reddit.recentWindowMinutes > 0) { ctx.runtimeConfig.reddit.recentWindowMinutes = updates.reddit.recentWindowMinutes; } + if (typeof updates.reddit.maxReadyCandidates === 'number' + && Number.isInteger(updates.reddit.maxReadyCandidates) + && updates.reddit.maxReadyCandidates >= 1 + && updates.reddit.maxReadyCandidates <= 500) { + ctx.runtimeConfig.reddit.maxReadyCandidates = updates.reddit.maxReadyCandidates; + } } // Mirror into the loaded config so getStats reflects new values. @@ -107,6 +120,7 @@ export function createConfigRouter(ctx: WebContext): Router { ctx.config.browser.dryRun = ctx.runtimeConfig.browser.dryRun; ctx.config.reddit.prioritizeRecentPosts = ctx.runtimeConfig.reddit.prioritizeRecentPosts; ctx.config.reddit.recentWindowMinutes = ctx.runtimeConfig.reddit.recentWindowMinutes; + ctx.config.reddit.maxReadyCandidates = ctx.runtimeConfig.reddit.maxReadyCandidates; if (parsedSubreddits?.success) { const next: SubredditConfig[] = parsedSubreddits.data; diff --git a/src/web/server.ts b/src/web/server.ts index bfb526e..00674d2 100644 --- a/src/web/server.ts +++ b/src/web/server.ts @@ -95,6 +95,7 @@ export class WebServer { reddit: { prioritizeRecentPosts: config.reddit.prioritizeRecentPosts, recentWindowMinutes: config.reddit.recentWindowMinutes, + maxReadyCandidates: config.reddit.maxReadyCandidates, }, }; this.app = express(); diff --git a/src/web/types.ts b/src/web/types.ts index d280e5c..13b8368 100644 --- a/src/web/types.ts +++ b/src/web/types.ts @@ -65,5 +65,6 @@ export interface RuntimeConfig { reddit: { prioritizeRecentPosts: boolean; recentWindowMinutes: number; + maxReadyCandidates: number; }; } diff --git a/tests/candidate-queue.test.ts b/tests/candidate-queue.test.ts index 726ce2a..0fa15da 100644 --- a/tests/candidate-queue.test.ts +++ b/tests/candidate-queue.test.ts @@ -119,7 +119,7 @@ describe('CandidateQueueRepository', () => { 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(1_100)).toBe(1); + expect(queue.promoteDueRetries(25, 1_100)).toBe(1); expect(queue.claimReady('account-b', 'send-b', 100, 1_101)).toMatchObject({ postId: 'p1', assignedAccountId: 'account-b', diff --git a/tests/config.test.ts b/tests/config.test.ts index 60f9d46..645190f 100644 --- a/tests/config.test.ts +++ b/tests/config.test.ts @@ -41,6 +41,18 @@ describe('ConfigSchema', () => { } }); + it('defaults the ready-post limit to 25 and requires a bounded integer', () => { + const result = ConfigSchema.safeParse(minimalConfig({ reddit: { subreddits: ['test'] } })); + expect(result.success).toBe(true); + if (result.success) expect(result.data.reddit.maxReadyCandidates).toBe(25); + expect(ConfigSchema.safeParse(minimalConfig({ + reddit: { subreddits: ['test'], maxReadyCandidates: 0 }, + })).success).toBe(false); + expect(ConfigSchema.safeParse(minimalConfig({ + reddit: { subreddits: ['test'], maxReadyCandidates: 25.5 }, + })).success).toBe(false); + }); + it('applies default tone to subreddits', () => { const result = ConfigSchema.safeParse(minimalConfig({ reddit: { subreddits: ['test'] } })); expect(result.success).toBe(true); diff --git a/tests/dashboard/settings.test.tsx b/tests/dashboard/settings.test.tsx index 935a566..0e9622c 100644 --- a/tests/dashboard/settings.test.tsx +++ b/tests/dashboard/settings.test.tsx @@ -13,7 +13,7 @@ import { SettingsRoute } from '../../apps/dashboard/src/features/settings/settin const config = { rateLimit: { minDelayBetweenMessagesMs: 300_000, maxMessagesPerHour: 8, jitterMs: 60_000 }, browser: { dryRun: true }, - reddit: { prioritizeRecentPosts: false, recentWindowMinutes: 60 }, + reddit: { prioritizeRecentPosts: false, recentWindowMinutes: 60, maxReadyCandidates: 25 }, subreddits: [{ name: 'casualconversation', type: 'chat', weight: 2, tone: 'default' }], tones: ['default', 'dominant', 'submissive', 'playful', 'direct', 'gentle', 'kink-friendly', 'teasing'], advanced: { openrouter: { chatModel: 'example/chat-model' } }, @@ -173,7 +173,7 @@ describe('settings and access', () => { expect(screen.getByRole('button', { name: 'Saving settings' })).toBeDisabled(); await waitFor(() => expect(resolveUpdate).toBeTypeOf('function')); - expect(requestBody).toMatchObject({ reddit: { prioritizeRecentPosts: true, recentWindowMinutes: 60 } }); + expect(requestBody).toMatchObject({ reddit: { prioritizeRecentPosts: true, recentWindowMinutes: 60, maxReadyCandidates: 25 } }); resolveUpdate?.(HttpResponse.json({ success: true, config: { diff --git a/tests/testUtils.ts b/tests/testUtils.ts index ea697bb..b5673d5 100644 --- a/tests/testUtils.ts +++ b/tests/testUtils.ts @@ -43,6 +43,7 @@ export function getTestConfig(overrides?: Partial): Config { sellerDomains: [], prioritizeRecentPosts: false, recentWindowMinutes: 60, + maxReadyCandidates: 25, ...overrides?.reddit, }, openrouter: {