`src/ingest/run.ts`

54 lines
/**
 * Shared ingest harness: every job opens an ingest_runs row, downloads to
 * data/downloads/ (sha256 recorded), and closes the run with counts.
 * HTTP != 200 or a 0-row parse must end as status='failed' + non-zero exit —
 * URL drift shows up as a red run, never silent staleness.
 */
import { createHash } from 'node:crypto';
import { mkdirSync, writeFileSync, readFileSync, existsSync } from 'node:fs';
import { join, dirname } from 'node:path';
import { fileURLToPath } from 'node:url';
import { query } from '../../db/pool.ts';

const __dirname = dirname(fileURLToPath(import.meta.url));
export const DOWNLOADS = join(__dirname, '..', '..', 'data', 'downloads');

export async function openRun(source: string, url?: string): Promise<number> {
  const r = await query<{ id: number }>(
    `INSERT INTO ingest_runs (source, url) VALUES ($1, $2) RETURNING id`,
    [source, url ?? null],
  );
  return r.rows[0].id;
}

export async function closeRun(
  id: number,
  status: 'ok' | 'failed',
  counts: { upserted?: number; skipped?: number; notes?: string; fileHash?: string } = {},
): Promise<void> {
  await query(
    `UPDATE ingest_runs SET finished_at = NOW(), status = $2,
       rows_upserted = COALESCE($3, rows_upserted),
       rows_skipped = COALESCE($4, rows_skipped),
       notes = COALESCE($5, notes),
       file_hash = COALESCE($6, file_hash)
     WHERE id = $1`,
    [id, status, counts.upserted ?? null, counts.skipped ?? null, counts.notes ?? null, counts.fileHash ?? null],
  );
}

/** Download url -> data/downloads/<filename>. Returns { path, sha256 }. Throws on non-200. */
export async function download(url: string, filename: string, opts: { reuseCached?: boolean; headers?: Record<string, string> } = {}): Promise<{ path: string; sha256: string }> {
  mkdirSync(DOWNLOADS, { recursive: true });
  const dest = join(DOWNLOADS, filename);
  if (opts.reuseCached && existsSync(dest)) {
    const buf = readFileSync(dest);
    return { path: dest, sha256: createHash('sha256').update(buf).digest('hex') };
  }
  const res = await fetch(url, { redirect: 'follow', headers: opts.headers });
  if (!res.ok) throw new Error(`download failed ${res.status} ${res.statusText}: ${url}`);
  const buf = Buffer.from(await res.arrayBuffer());
  writeFileSync(dest, buf);
  return { path: dest, sha256: createHash('sha256').update(buf).digest('hex') };
}