1import { and, eq, isNull, lte, or, sql } from 'drizzle-orm';2import { connectors as connectorsTable } from '@rareindex/database';3import { logger } from '@rareindex/shared';4import { db } from '../lib/db.ts';5import { JOBS, type Queue } from '../lib/queue.ts';67const PRIORITY: Record<string, number> = { high: 10, medium: 5, low: 1 };89/**10 * Dynamic crawl scheduling (§141): every tick, enqueue a crawl for active connectors whose11 * next_run_at is due (or never ran). A singleton key per connector prevents overlapping runs.12 */13export async function scheduleDueCrawls(queue: Queue): Promise<number> {14 const now = new Date();15 const due = await db()16 .select({ id: connectorsTable.id, priority: connectorsTable.priority })17 .from(connectorsTable)18 .where(and(eq(connectorsTable.status, 'active'), or(isNull(connectorsTable.nextRunAt), lte(connectorsTable.nextRunAt, now))));19 let n = 0;20 for (const c of due) {21 const id = await queue.send(JOBS.crawlRun, { connectorId: c.id, mode: 'incremental', trigger: 'schedule' }, { singletonKey: `crawl:${c.id}`, priority: PRIORITY[c.priority] ?? 5, expireInSeconds: 6 * 3600, retryLimit: 1 });22 if (id) n++;23 }24 // push next_run_at forward slightly so the same tick does not re-enqueue before the run starts25 if (due.length) {26 await db()27 .update(connectorsTable)28 .set({ nextRunAt: sql`now() + interval '10 minutes'` })29 .where(and(eq(connectorsTable.status, 'active'), or(isNull(connectorsTable.nextRunAt), lte(connectorsTable.nextRunAt, now))));30 }31 if (n) logger.info({ enqueued: n }, 'scheduled due crawls');32 return n;33}34