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 try {92 for await (const rec of connector.crawl(ctx)) {93 const input: RawRecordInput = rec;94 const payloadText = typeof input.payload === 'string' ? input.payload : JSON.stringify(input.payload ?? null);95 const contentHash = sha256(`${input.kind}|${input.externalId ?? input.url}|${payloadText}`);96 let snapshotRef: string | null = null;97 if (input.snapshot && input.snapshot.length > SNAPSHOT_THRESHOLD) {98 const p = snapshotPath(connectorId, contentHash);99 writeFileSync(p, input.snapshot);100 snapshotRef = p;101 }102 batch.push({103 id: newId('raw'),104 connectorId,105 sourceId: meta.sourceId,106 runId,107 engine: input.engine,108 url: input.url,109 externalId: input.externalId ?? null,110 kind: input.kind,111 fetchedAt: input.fetchedAt ?? new Date(),112 contentHash,113 httpStatus: input.httpStatus ?? null,114 payload: input.snapshot && !snapshotRef && typeof input.payload === 'object' && input.payload ? { ...(input.payload as object), snapshot: input.snapshot } : (input.payload as object),115 snapshotRef,116 parserVersion: connector.parserVersion,117 connectorVersion: connector.version,118 });119 if (batch.length >= 200) await flush();120 if (ctx.signal?.aborted) break;121 }122 await flush();123 } catch (err) {124 error = err instanceof Error ? `${err.message}` : String(err);125 log.error({ err }, 'crawl failed');126 try {127 await flush();128 } catch (e2) {129 log.error({ err: e2 }, 'flush after failure failed');130 }131 }132133 const stats = ctx.engineStats;134 const attempted = Object.values(stats).reduce((a, s) => a + s.attempts, 0);135 const success = Object.values(stats).reduce((a, s) => a + s.success, 0);136 const credits = Object.values(stats).reduce((a, s) => a + s.credits, 0);137 const dupAnomaly = duplicateExplosion(raw + dupes, dupes);138 if (dupAnomaly && mode !== 'backfill') anomalies.push(dupAnomaly);139 if (attempted > 0 && success === 0) anomalies.push('all_pages_failed');140 const status: CrawlRunResult['status'] = error ? (raw > 0 ? 'partial' : 'failed') : 'success';141 const finishedAt = new Date();142 await db()143 .update(connectorRuns)144 .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 ?? {} })145 .where(eq(connectorRuns.id, runId));146 const refresh = state?.refreshFrequencyMinutes ?? meta.refreshFrequencyMinutes;147 const nextRunAt = new Date(finishedAt.getTime() + refresh * 60_000 * (status === 'failed' ? 2 : 1));148 await db()149 .update(connectorsTable)150 .set({151 lastRunAt: finishedAt,152 lastSuccessAt: status === 'failed' ? sql`${connectorsTable.lastSuccessAt}` : finishedAt,153 nextRunAt,154 config: cursor ? sql`${connectorsTable.config} || ${JSON.stringify({ cursor })}::jsonb` : sql`${connectorsTable.config}`,155 updatedAt: finishedAt,156 })157 .where(and(eq(connectorsTable.id, connectorId)));158 await flushCosts();159 await emit('connector_run_finished', { type: 'connector_run', id: runId }, { connectorId, status, raw, dupes, attempted, success, error });160 log.info({ runId, status, raw, dupes, attempted, success, credits, ms: Date.now() - started }, 'crawl finished');161 return { runId, connectorId, status, recordsRaw: raw, recordsDuplicate: dupes, pagesAttempted: attempted, pagesSuccess: success, error, durationMs: Date.now() - started };162}163