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 // pg-boss singleton keys do not dedupe across states reliably; skip connectors that already have a queued/active crawl.21 const queued = (await db().execute(sql`select distinct data->>'connectorId' as id from pgboss.job where name = ${JOBS.crawlRun} and state in ('created','retry','active')`)) as unknown as Array<{ id: string }>;22 const busy = new Set(queued.map((q) => q.id));23 for (const c of due) {24 if (busy.has(c.id)) continue;25 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 });26 if (id) n++;27 }28 // push next_run_at forward slightly so the same tick does not re-enqueue before the run starts29 if (due.length) {30 await db()31 .update(connectorsTable)32 .set({ nextRunAt: sql`now() + interval '10 minutes'` })33 .where(and(eq(connectorsTable.status, 'active'), or(isNull(connectorsTable.nextRunAt), lte(connectorsTable.nextRunAt, now))));34 }35 if (n) logger.info({ enqueued: n }, 'scheduled due crawls');36 return n;37}38