1import { mkdirSync, writeFileSync } from 'node:fs';2import path from 'node:path';3import { and, eq, sql } from 'drizzle-orm';4import { connectorRuns, connectors as connectorsTable, rawRecords } from '@rareindex/database';5import { createCrawlContext, loadConnector, type CrawlOptions, type RawRecordInput } from '@rareindex/connectors';6import { env, logger, newId, sha256 } from '@rareindex/shared';7import { db } from '../lib/db.ts';8import { createBudgetStore } from '../lib/budget.ts';9import { getRouter } from '../lib/router.ts';10import { flushCosts } from '../lib/costs.ts';11import { emit } from '../lib/events.ts';12import { duplicateExplosion } from '@rareindex/connectors';1314export interface CrawlRunResult {15 runId: string;16 connectorId: string;17 status: 'success' | 'partial' | 'failed';18 recordsRaw: number;19 recordsDuplicate: number;20 pagesAttempted: number;21 pagesSuccess: number;22 error: string | null;23 durationMs: number;24}2526const SNAPSHOT_THRESHOLD = 8 * 1024;2728function snapshotPath(connectorId: string, hash: string): string {29 const dir = path.resolve(env().RI_DATA_DIR, 'raw', connectorId);30 mkdirSync(dir, { recursive: true });31 return path.join(dir, `${hash}.html`);32}3334/**35 * Run one connector crawl (§108 raw layer): bookkeeping row in connector_runs, routed fetches with36 * budget + cost tracking, immutable raw_records with content-hash dedupe, cursor persistence.37 */38export async function runCrawl(connectorId: string, options: Partial<CrawlOptions> & { trigger?: string; onRaw?: (ids: string[]) => Promise<void> } = {}): Promise<CrawlRunResult> {39 const started = Date.now();40 const log = logger.child({ component: 'crawler', connector: connectorId });41 const connector = await loadConnector(connectorId);42 const meta = connector.meta;43 const [state] = await db().select().from(connectorsTable).where(eq(connectorsTable.id, connectorId)).limit(1);44 if (state && state.status !== 'active' && options.trigger !== 'manual') {45 log.info({ status: state.status }, 'connector not active; skipping');46 return { runId: '', connectorId, status: 'failed', recordsRaw: 0, recordsDuplicate: 0, pagesAttempted: 0, pagesSuccess: 0, error: `connector ${state.status}`, durationMs: 0 };47 }48 const runId = newId('job');49 const mode = options.mode ?? 'incremental';50 const savedCursor = (state?.config as { cursor?: Record<string, unknown> } | undefined)?.cursor;51 const crawlOptions: CrawlOptions = { mode, limit: options.limit, categories: options.categories, seeds: options.seeds, cursor: mode === 'backfill' ? undefined : savedCursor };52 await db().insert(connectorRuns).values({ id: runId, connectorId, trigger: options.trigger ?? 'schedule', startedAt: new Date(), status: 'running', cursor: savedCursor ?? {} });5354 const router = getRouter(connectorId);55 const anomalies: string[] = [];56 let cursor: Record<string, unknown> | null = null;57 const ctx = createCrawlContext({58 router,59 meta,60 options: crawlOptions,61 log,62 budget: createBudgetStore({ connectorId }),63 onCursor: async (c) => {64 cursor = c;65 await db().update(connectorRuns).set({ cursor: c }).where(eq(connectorRuns.id, runId));66 },67 onAnomaly: (kind, detail) => anomalies.push(detail ? `${kind}: ${detail}` : kind),68 });6970 let raw = 0;71 let dupes = 0;72 let error: string | null = null;73 let batch: Array<typeof rawRecords.$inferInsert> = [];74 const newIds: string[] = [];7576 async function flush() {77 if (batch.length === 0) return;78 const rows = batch;79 batch = [];80 const inserted = await db()81 .insert(rawRecords)82 .values(rows)83 .onConflictDoNothing({ target: [rawRecords.connectorId, rawRecords.contentHash] })84 .returning({ id: rawRecords.id });85 raw += inserted.length;86 dupes += rows.length - inserted.length;87 for (const r of inserted) newIds.push(r.id);88 if (options.onRaw && inserted.length) await options.onRaw(inserted.map((r) => r.id));89 }9091 // Time budget per run: very large sources (100k+ pages) would otherwise monopolise crawl slots for hours.92 // The connector's cursor is persisted at its checkpoints, so a time-boxed run simply resumes on the next tick.93 const maxRunMs = Number(process.env.RI_MAX_RUN_MINUTES ?? 40) * 60_000;94 let timeBoxed = false;95 try {96 for await (const rec of connector.crawl(ctx)) {97 if (mode !== 'probe' && Date.now() - started > maxRunMs) {98 timeBoxed = true;99 break;100 }101 const input: RawRecordInput = rec;102 const payloadText = typeof input.payload === 'string' ? input.payload : JSON.stringify(input.payload ?? null);103 const contentHash = sha256(`${input.kind}|${input.externalId ?? input.url}|${payloadText}`);104 let snapshotRef: string | null = null;105 if (input.snapshot && input.snapshot.length > SNAPSHOT_THRESHOLD) {106 const p = snapshotPath(connectorId, contentHash);107 writeFileSync(p, input.snapshot);108 snapshotRef = p;109 }110 batch.push({111 id: newId('raw'),112 connectorId,113 sourceId: meta.sourceId,114 runId,115 engine: input.engine,116 url: input.url,117 externalId: input.externalId ?? null,118 kind: input.kind,119 fetchedAt: input.fetchedAt ?? new Date(),120 contentHash,121 httpStatus: input.httpStatus ?? null,122 payload: input.snapshot && !snapshotRef && typeof input.payload === 'object' && input.payload ? { ...(input.payload as object), snapshot: input.snapshot } : (input.payload as object),123 snapshotRef,124 parserVersion: connector.parserVersion,125 connectorVersion: connector.version,126 });127 if (batch.length >= 200) await flush();128 if (ctx.signal?.aborted) break;129 }130 await flush();131 } catch (err) {132 error = err instanceof Error ? `${err.message}` : String(err);133 log.error({ err }, 'crawl failed');134 try {135 await flush();136 } catch (e2) {137 log.error({ err: e2 }, 'flush after failure failed');138 }139 }140141 const stats = ctx.engineStats;142 const attempted = Object.values(stats).reduce((a, s) => a + s.attempts, 0);143 const success = Object.values(stats).reduce((a, s) => a + s.success, 0);144 const credits = Object.values(stats).reduce((a, s) => a + s.credits, 0);145 const dupAnomaly = duplicateExplosion(raw + dupes, dupes);146 if (dupAnomaly && mode !== 'backfill') anomalies.push(dupAnomaly);147 if (attempted > 0 && success === 0) anomalies.push('all_pages_failed');148 if (timeBoxed) anomalies.push('time_budget_reached');149 const status: CrawlRunResult['status'] = error ? (raw > 0 ? 'partial' : 'failed') : timeBoxed ? 'partial' : 'success';150 const finishedAt = new Date();151 await db()152 .update(connectorRuns)153 .set({ finishedAt, status, pagesAttempted: attempted, pagesSuccess: success, recordsRaw: raw, recordsDuplicate: dupes, engineStats: stats, anomalies: [...anomalies, ...ctx.anomalies.filter((a) => !anomalies.includes(a))], error, costCredits: credits, cursor: cursor ?? savedCursor ?? {} })154 .where(eq(connectorRuns.id, runId));155 const refresh = state?.refreshFrequencyMinutes ?? meta.refreshFrequencyMinutes;156 // A time-boxed run resumes quickly (cursor kept); failures back off ×2.157 const nextRunAt = timeBoxed ? new Date(finishedAt.getTime() + 3 * 60_000) : new Date(finishedAt.getTime() + refresh * 60_000 * (status === 'failed' ? 2 : 1));158 await db()159 .update(connectorsTable)160 .set({161 lastRunAt: finishedAt,162 lastSuccessAt: status === 'failed' ? sql`${connectorsTable.lastSuccessAt}` : finishedAt,163 nextRunAt,164 config: cursor ? sql`${connectorsTable.config} || ${JSON.stringify({ cursor })}::jsonb` : sql`${connectorsTable.config}`,165 updatedAt: finishedAt,166 })167 .where(and(eq(connectorsTable.id, connectorId)));168 await flushCosts();169 await emit('connector_run_finished', { type: 'connector_run', id: runId }, { connectorId, status, raw, dupes, attempted, success, error });170 log.info({ runId, status, raw, dupes, attempted, success, credits, ms: Date.now() - started }, 'crawl finished');171 return { runId, connectorId, status, recordsRaw: raw, recordsDuplicate: dupes, pagesAttempted: attempted, pagesSuccess: success, error, durationMs: Date.now() - started };172}173