import { and, eq, isNull, lte, or, sql } from 'drizzle-orm';
import { connectors as connectorsTable } from '@rareindex/database';
import { logger } from '@rareindex/shared';
import { db } from '../lib/db.ts';
import { JOBS, type Queue } from '../lib/queue.ts';

const PRIORITY: Record<string, number> = { high: 10, medium: 5, low: 1 };

/**
 * Dynamic crawl scheduling (§141): every tick, enqueue a crawl for active connectors whose
 * next_run_at is due (or never ran). A singleton key per connector prevents overlapping runs.
 */
export async function scheduleDueCrawls(queue: Queue): Promise<number> {
  const now = new Date();
  const due = await db()
    .select({ id: connectorsTable.id, priority: connectorsTable.priority })
    .from(connectorsTable)
    .where(and(eq(connectorsTable.status, 'active'), or(isNull(connectorsTable.nextRunAt), lte(connectorsTable.nextRunAt, now))));
  let n = 0;
  // pg-boss singleton keys do not dedupe across states reliably; skip connectors that already have a queued/active crawl.
  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 }>;
  const busy = new Set(queued.map((q) => q.id));
  for (const c of due) {
    if (busy.has(c.id)) continue;
    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 });
    if (id) n++;
  }
  // push next_run_at forward slightly so the same tick does not re-enqueue before the run starts
  if (due.length) {
    await db()
      .update(connectorsTable)
      .set({ nextRunAt: sql`now() + interval '10 minutes'` })
      .where(and(eq(connectorsTable.status, 'active'), or(isNull(connectorsTable.nextRunAt), lte(connectorsTable.nextRunAt, now))));
  }
  if (n) logger.info({ enqueued: n }, 'scheduled due crawls');
  return n;
}