TypeScript 61.9%
HTML 37.2%
SQL 0.7%
1import { mkdirSync, writeFileSync } from 'node:fs';2import path from 'node:path';3import { and, desc, eq, inArray, sql } from 'drizzle-orm';4import { connectorBackfills, connectorRuns, connectors as connectorsTable, rawRecords } from '@rareindex/database';5import { createCrawlContext, duplicateExplosion, loadConnector, missingRequirements, type BackfillProgress, type CrawlOptions, type RawRecordInput } from '@rareindex/connectors';6import { env, logger, newId, sha256, toDateOnly } 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';1213export interface CrawlRunResult {14 runId: string;15 connectorId: string;16 status: 'success' | 'partial' | 'failed' | 'skipped';17 recordsRaw: number;18 recordsDuplicate: number;19 pagesAttempted: number;20 pagesSuccess: number;21 error: string | null;22 durationMs: number;23 backfillId?: string | null;24}2526const SNAPSHOT_THRESHOLD = 8 * 1024;27const BACKFILL_MAX_ERRORS = 8;2829function snapshotPath(connectorId: string, hash: string): string {30 const dir = path.resolve(env().RI_DATA_DIR, 'raw', connectorId);31 mkdirSync(dir, { recursive: true });32 return path.join(dir, `${hash}.html`);33}3435/** Active (running/paused) backfill campaign for a connector, if any. */36export async function activeBackfill(connectorId: string) {37 const [row] = await db().select().from(connectorBackfills).where(and(eq(connectorBackfills.connectorId, connectorId), inArray(connectorBackfills.status, ['running', 'paused']))).orderBy(desc(connectorBackfills.startedAt)).limit(1);38 return row ?? null;39}4041/** Start (or reset) a backfill campaign. Existing running campaigns are resumed unless `reset` is set. */42export async function startBackfill(connectorId: string, opts: { reset?: boolean; startDate?: string | null; endDate?: string | null } = {}) {43 const existing = await activeBackfill(connectorId);44 if (existing && !opts.reset) {45 if (existing.status === 'paused') await db().update(connectorBackfills).set({ status: 'running', updatedAt: new Date() }).where(eq(connectorBackfills.id, existing.id));46 return existing;47 }48 if (existing) await db().update(connectorBackfills).set({ status: 'failed', lastError: 'reset by operator', finishedAt: new Date(), updatedAt: new Date() }).where(eq(connectorBackfills.id, existing.id));49 const id = newId('backfill');50 const [row] = await db().insert(connectorBackfills).values({ id, connectorId, status: 'running', startedAt: new Date(), backfillStartDate: opts.startDate ?? null, backfillEndDate: opts.endDate ?? null }).returning();51 return row!;52}5354export async function pauseBackfill(connectorId: string): Promise<boolean> {55 const b = await activeBackfill(connectorId);56 if (!b) return false;57 await db().update(connectorBackfills).set({ status: 'paused', updatedAt: new Date() }).where(eq(connectorBackfills.id, b.id));58 return true;59}6061/**62 * Run one connector crawl (§108 raw layer): bookkeeping row in connector_runs, routed fetches with63 * budget + cost tracking, immutable raw_records with content-hash dedupe, cursor persistence.64 * Backfill mode is resumable (SPEC §9): the cursor comes from the active connector_backfills row and65 * progress is written back after every checkpoint, so an interrupted campaign continues where it stopped.66 */67export async function runCrawl(connectorId: string, options: Partial<CrawlOptions> & { trigger?: string; resetBackfill?: boolean; onRaw?: (ids: string[]) => Promise<void> } = {}): Promise<CrawlRunResult> {68 const started = Date.now();69 const log = logger.child({ component: 'crawler', connector: connectorId });70 const connector = await loadConnector(connectorId);71 const meta = connector.meta;72 const [state] = await db().select().from(connectorsTable).where(eq(connectorsTable.id, connectorId)).limit(1);73 const skip = (error: string): CrawlRunResult => ({ runId: '', connectorId, status: 'skipped', recordsRaw: 0, recordsDuplicate: 0, pagesAttempted: 0, pagesSuccess: 0, error, durationMs: 0 });74 if (state && state.status !== 'active' && options.trigger !== 'manual') {75 log.info({ status: state.status }, 'connector not active; skipping');76 return skip(`connector ${state.status}`);77 }78 const missing = missingRequirements(meta);79 if (missing.length) {80 log.warn({ missing }, 'connector requirements missing; skipping (gated source)');81 await db().update(connectorsTable).set({ nextRunAt: new Date(Date.now() + 24 * 3600_000), updatedAt: new Date() }).where(eq(connectorsTable.id, connectorId));82 return skip(`missing env: ${missing.join(', ')}`);83 }84 const runId = newId('job');85 const mode = options.mode ?? 'incremental';86 const savedCursor = (state?.config as { cursor?: Record<string, unknown> } | undefined)?.cursor;8788 // Backfill campaign bookkeeping (resumable)89 let backfill = mode === 'backfill' ? await startBackfill(connectorId, { reset: options.resetBackfill }) : null;90 if (backfill && backfill.status === 'paused' && options.trigger !== 'manual') return skip('backfill paused');91 const backfillCursor = backfill && Object.keys(backfill.lastCursor ?? {}).length ? (backfill.lastCursor as Record<string, unknown>) : undefined;92 if (backfillCursor && (backfillCursor as { done?: boolean }).done) {93 await db().update(connectorBackfills).set({ status: 'completed', finishedAt: new Date(), updatedAt: new Date(), percent: 100 }).where(eq(connectorBackfills.id, backfill!.id));94 return skip('backfill already completed');95 }96 const crawlOptions: CrawlOptions = { mode, limit: options.limit, categories: options.categories, seeds: options.seeds, cursor: mode === 'backfill' ? backfillCursor : savedCursor };97 await db().insert(connectorRuns).values({ id: runId, connectorId, trigger: options.trigger ?? 'schedule', startedAt: new Date(), status: 'running', cursor: crawlOptions.cursor ?? {} });98 if (backfill) await db().update(connectorBackfills).set({ runs: sql`${connectorBackfills.runs} + 1`, status: 'running', updatedAt: new Date() }).where(eq(connectorBackfills.id, backfill.id));99100 const router = getRouter(connectorId);101 const anomalies: string[] = [];102 let cursor: Record<string, unknown> | null = null;103 let lastProgress: BackfillProgress | null = null;104 const ctx = createCrawlContext({105 router,106 meta,107 options: crawlOptions,108 log,109 budget: createBudgetStore({ connectorId }),110 onCursor: async (c) => {111 cursor = c;112 await db().update(connectorRuns).set({ cursor: c }).where(eq(connectorRuns.id, runId));113 if (backfill) await db().update(connectorBackfills).set({ lastCursor: c, updatedAt: new Date() }).where(eq(connectorBackfills.id, backfill.id));114 },115 onAnomaly: (kind, detail) => anomalies.push(detail ? `${kind}: ${detail}` : kind),116 onProgress: async (p) => {117 if (!backfill) return;118 lastProgress = p;119 const percent = p.totalPages && p.page ? Math.min(100, Math.round((p.page / p.totalPages) * 1000) / 10) : null;120 await db()121 .update(connectorBackfills)122 .set({123 pagesProcessed: p.page !== undefined ? sql`greatest(${connectorBackfills.pagesProcessed}, ${p.page})` : sql`${connectorBackfills.pagesProcessed} + 1`,124 lastSuccessfulPage: p.page ?? sql`${connectorBackfills.lastSuccessfulPage}`,125 totalPages: p.totalPages ?? sql`${connectorBackfills.totalPages}`,126 itemsProcessed: p.itemsProcessed !== undefined ? sql`greatest(${connectorBackfills.itemsProcessed}, ${p.itemsProcessed})` : sql`${connectorBackfills.itemsProcessed}`,127 reachedDate: p.reachedDate ? toDateOnly(p.reachedDate) : sql`${connectorBackfills.reachedDate}`,128 percent: percent ?? sql`${connectorBackfills.percent}`,129 updatedAt: new Date(),130 })131 .where(eq(connectorBackfills.id, backfill.id));132 },133 });134135 let raw = 0;136 let dupes = 0;137 let error: string | null = null;138 let batch: Array<typeof rawRecords.$inferInsert> = [];139 const newIds: string[] = [];140141 async function flush() {142 if (batch.length === 0) return;143 const rows = batch;144 batch = [];145 const inserted = await db()146 .insert(rawRecords)147 .values(rows)148 .onConflictDoNothing({ target: [rawRecords.connectorId, rawRecords.contentHash] })149 .returning({ id: rawRecords.id });150 raw += inserted.length;151 dupes += rows.length - inserted.length;152 for (const r of inserted) newIds.push(r.id);153 if (options.onRaw && inserted.length) await options.onRaw(inserted.map((r) => r.id));154 }155156 // Time budget per run: very large sources (100k+ pages) would otherwise monopolise crawl slots for hours.157 // The connector's cursor is persisted at its checkpoints, so a time-boxed run simply resumes on the next tick.158 const maxRunMs = Number(process.env.RI_MAX_RUN_MINUTES ?? 40) * 60_000;159 let timeBoxed = false;160 try {161 for await (const rec of connector.crawl(ctx)) {162 if (mode !== 'probe' && Date.now() - started > maxRunMs) {163 timeBoxed = true;164 break;165 }166 const input: RawRecordInput = rec;167 const payloadText = typeof input.payload === 'string' ? input.payload : JSON.stringify(input.payload ?? null);168 const contentHash = sha256(`${input.kind}|${input.externalId ?? input.url}|${payloadText}`);169 let snapshotRef: string | null = null;170 if (input.snapshot && input.snapshot.length > SNAPSHOT_THRESHOLD) {171 const p = snapshotPath(connectorId, contentHash);172 writeFileSync(p, input.snapshot);173 snapshotRef = p;174 }175 batch.push({176 id: newId('raw'),177 connectorId,178 sourceId: meta.sourceId,179 runId,180 engine: input.engine,181 url: input.url,182 externalId: input.externalId ?? null,183 kind: input.kind,184 fetchedAt: input.fetchedAt ?? new Date(),185 contentHash,186 httpStatus: input.httpStatus ?? null,187 payload: input.snapshot && !snapshotRef && typeof input.payload === 'object' && input.payload ? { ...(input.payload as object), snapshot: input.snapshot } : (input.payload as object),188 snapshotRef,189 parserVersion: connector.parserVersion,190 connectorVersion: connector.version,191 });192 if (batch.length >= 200) await flush();193 if (ctx.signal?.aborted) break;194 }195 await flush();196 } catch (err) {197 error = err instanceof Error ? `${err.message}` : String(err);198 log.error({ err }, 'crawl failed');199 try {200 await flush();201 } catch (e2) {202 log.error({ err: e2 }, 'flush after failure failed');203 }204 }205206 const stats = ctx.engineStats;207 const attempted = Object.values(stats).reduce((a, s) => a + s.attempts, 0);208 const success = Object.values(stats).reduce((a, s) => a + s.success, 0);209 const credits = Object.values(stats).reduce((a, s) => a + s.credits, 0);210 const blocked = Object.values(stats).reduce((a, s) => a + (s.blocked ?? 0), 0);211 const refused = Object.values(stats).reduce((a, s) => a + (s.circuitOpen ?? 0), 0);212 const dupAnomaly = duplicateExplosion(raw + dupes, dupes);213 if (dupAnomaly && mode !== 'backfill') anomalies.push(dupAnomaly);214 if (attempted > 0 && success === 0) anomalies.push('all_pages_failed');215 if (blocked > 0) anomalies.push(`challenge_events: ${blocked}`);216 if (refused > 0) anomalies.push(`circuit_open: ${refused} requests refused`);217 if (timeBoxed) anomalies.push('time_budget_reached');218 const status: CrawlRunResult['status'] = error ? (raw > 0 ? 'partial' : 'failed') : timeBoxed ? 'partial' : 'success';219 const finishedAt = new Date();220 const finalCursor: Record<string, unknown> | undefined = cursor ?? (crawlOptions.cursor as Record<string, unknown> | undefined) ?? undefined;221 await db()222 .update(connectorRuns)223 .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: finalCursor ?? {} })224 .where(eq(connectorRuns.id, runId));225226 // Backfill campaign state (SPEC §9): completed when the connector signalled `done`, or when a run227 // finished naturally (not time-boxed, no error); otherwise it stays running and resumes next tick.228 if (backfill) {229 const done = Boolean((finalCursor as { done?: boolean } | undefined)?.done) || (status === 'success' && !timeBoxed);230 const failedHard = status === 'failed';231 const [bf] = await db().select({ errors: connectorBackfills.errors }).from(connectorBackfills).where(eq(connectorBackfills.id, backfill.id)).limit(1);232 const errors = (bf?.errors ?? 0) + (error ? 1 : 0);233 await db()234 .update(connectorBackfills)235 .set({236 status: done ? 'completed' : failedHard && errors >= BACKFILL_MAX_ERRORS ? 'failed' : 'running',237 finishedAt: done || (failedHard && errors >= BACKFILL_MAX_ERRORS) ? finishedAt : null,238 errors,239 retryCount: error ? sql`${connectorBackfills.retryCount} + 1` : sql`${connectorBackfills.retryCount}`,240 lastError: error,241 lastCursor: finalCursor ?? {},242 itemsProcessed: sql`${connectorBackfills.itemsProcessed} + ${raw}`,243 percent: done ? 100 : sql`${connectorBackfills.percent}`,244 updatedAt: finishedAt,245 })246 .where(eq(connectorBackfills.id, backfill.id));247 if (done) log.info({ backfillId: backfill.id }, 'backfill completed');248 }249250 const refresh = state?.refreshFrequencyMinutes ?? meta.refreshFrequencyMinutes;251 // A time-boxed run resumes quickly (cursor kept); failures back off ×2.252 const nextRunAt = timeBoxed ? new Date(finishedAt.getTime() + 3 * 60_000) : new Date(finishedAt.getTime() + refresh * 60_000 * (status === 'failed' ? 2 : 1));253 await db()254 .update(connectorsTable)255 .set({256 lastRunAt: finishedAt,257 lastSuccessAt: status === 'failed' ? sql`${connectorsTable.lastSuccessAt}` : finishedAt,258 nextRunAt,259 // incremental cursors live in connectors.config; backfill cursors live on the campaign row260 config: cursor && mode !== 'backfill' ? sql`${connectorsTable.config} || ${JSON.stringify({ cursor })}::jsonb` : sql`${connectorsTable.config}`,261 updatedAt: finishedAt,262 })263 .where(and(eq(connectorsTable.id, connectorId)));264 await flushCosts();265 await emit('connector_run_finished', { type: 'connector_run', id: runId }, { connectorId, status, mode, raw, dupes, attempted, success, error, backfillId: backfill?.id ?? null, progress: lastProgress });266 log.info({ runId, status, mode, raw, dupes, attempted, success, credits, blocked, ms: Date.now() - started }, 'crawl finished');267 return { runId, connectorId, status, recordsRaw: raw, recordsDuplicate: dupes, pagesAttempted: attempted, pagesSuccess: success, error, durationMs: Date.now() - started, backfillId: backfill?.id ?? null };268}269