`scripts/canary-freshness.js`

123 lines
// LA-feed freshness + row-count canary (READ-ONLY, $0 local).
// Baseline-aware, per Steve's canary doctrine: flag a REGRESSION (row count dropped
// below the stored baseline, or a feed's newest record aged past its max) — not
// absolute thresholds. First run establishes the baseline and reports OK.
// Writes tmp/canary/latest.json (heartbeat). Report-only: it alerts nobody and
// writes no source data. Alerting (CNCP/George) is a separate, gated follow-up.
import { q, pool } from '../src/db.js';
import fs from 'fs';
import path from 'path';
import { fileURLToPath } from 'url';

const __dirname = path.dirname(fileURLToPath(import.meta.url));
const BASE = path.resolve(__dirname, '../data/canary-baseline.json');
const OUT = path.resolve(__dirname, '../tmp/canary');

// table -> { where?, freshCol?, maxAgeDays? }  (freshCol omitted = no freshness check)
const MANIFEST = {
  la_building_permits_raw:      { freshCol: 'issue_date', where: "dataset_id='pi9x-tg5x'", maxAgeDays: 10 },
  la_assessor_parcels_raw:      {},
  la_parcel_geom:               {},
  la_gis_features:              {},
  la_code_enforcement_raw:      { freshCol: 'add_dttm', maxAgeDays: 10 },
  la_business_registrations_raw:{}, // location_start_date is future-dated; skip freshness
  cslb_raw:                     {},
};
const DROP_TOLERANCE = 0.97; // count below baseline*0.97 = regression

async function main() {
  const baseline = fs.existsSync(BASE) ? JSON.parse(fs.readFileSync(BASE, 'utf8')) : {};
  const firstRun = Object.keys(baseline).length === 0;
  const results = [];

  // Rolling history for SLOW-BLEED detection (Cody c1 #2): the ratchet floor catches a
  // big drop but hides a slow net decline. Compare each table's count to its EARLIEST
  // observation (widest window); flag BLEED on a sustained shrink the floor would miss.
  const HIST = path.resolve(__dirname, '../data/canary-history.jsonl');
  const history = fs.existsSync(HIST) ? fs.readFileSync(HIST, 'utf8').trim().split('\n').filter(Boolean).map(l => JSON.parse(l)) : [];
  const nowTs = (await q(`SELECT now() n`)).rows[0].n;
  // TIME-anchored bleed reference (Cody c3): the newest history entry OLDER than 30d
  // (else the earliest available). Run-count anchoring let a slow decline walk the
  // floor up; a time anchor holds regardless of run cadence / the 120-run trim.
  const cutoff = new Date(new Date(nowTs).getTime() - 30 * 86400 * 1000).toISOString();
  const older = history.filter(h => String(h.ts) <= cutoff);
  const anchor = ((older.length ? older[older.length - 1] : history[0]) || {}).counts || {};
  const curCounts = {};

  for (const [table, cfg] of Object.entries(MANIFEST)) {
    const whereSql = cfg.where ? `WHERE ${cfg.where}` : '';
    const count = Number((await q(`SELECT count(*) c FROM ${table} ${whereSql}`)).rows[0].c);

    let ageDays = null, freshest = null;
    if (cfg.freshCol) {
      const r = (await q(`SELECT max(${cfg.freshCol}) m, EXTRACT(EPOCH FROM (now() - max(${cfg.freshCol})))/86400 age FROM ${table} ${whereSql}`)).rows[0];
      freshest = r.m; ageDays = r.age == null ? null : Math.round(Number(r.age) * 10) / 10;
    }

    const base = baseline[table] ?? count;
    const floor = Math.floor(base * DROP_TOLERANCE);
    curCounts[table] = count;
    const ref = anchor[table]; // time-anchored (~30d) reference count
    let status = 'OK', reason = '';
    if (count < floor) { status = 'LOW'; reason = `count ${count.toLocaleString()} < floor ${floor.toLocaleString()} (baseline ${base.toLocaleString()})`; }
    else if (ref != null && history.length >= 1 && count < Math.floor(ref * 0.99)) { status = 'BLEED'; reason = `count ${count.toLocaleString()} < earliest observed ${Number(ref).toLocaleString()} (slow decline the floor misses)`; }
    else if (ageDays != null && ageDays > cfg.maxAgeDays) { status = 'STALE'; reason = `newest record ${ageDays}d old (> ${cfg.maxAgeDays}d)`; }

    results.push({ table, count, baseline: base, ageDays, freshest, status, reason });
    // baseline ratchets UP (never down) so a bad reload can't quietly reset the floor
    baseline[table] = Math.max(base, count);
  }

  // SOURCE staleness from la_ingest_state — the real "when did we last ingest / did it
  // error" signal. Covers the count-only tables the row check alone can't see stale, and
  // surfaces any source whose last run errored (Cody c1: count-only = false comfort).
  const SRC_MAX_AGE = 14; // days; generous until a scheduled refresh job exists
  // Known-broken upstreams: demoted to WARNING (BROKEN) so one unfixable source can't
  // hold the whole canary red forever (Cody c3: a permanently-FAIL canary is ignored).
  // Document the reason + date; remove the entry when the source is fixed.
  const KNOWN_BROKEN = {
    gis_zoning: 'NavigateLA ArcGIS gateway 502s past offset 44k; ~75% partial load; full LA zoning is fragmented across per-community Zone_Builder services (noted 2026-08-11)',
  };
  const srcRows = (await q(`SELECT source, last_status, EXTRACT(EPOCH FROM (now()-last_run))/86400 age FROM la_ingest_state`)).rows;
  const sources = srcRows.map(r => {
    const age = r.age == null ? null : Math.round(Number(r.age) * 10) / 10;
    const errored = String(r.last_status || '').startsWith('error');
    const status = errored ? (KNOWN_BROKEN[r.source] ? 'BROKEN' : 'ERRORED')
      : (age != null && age > SRC_MAX_AGE ? 'STALE' : 'OK');
    return { source: r.source, ageDays: age, last_status: r.last_status, status, note: KNOWN_BROKEN[r.source] };
  }).sort((a, b) => (b.ageDays || 0) - (a.ageDays || 0));

  fs.mkdirSync(path.dirname(BASE), { recursive: true });
  fs.writeFileSync(BASE, JSON.stringify(baseline, null, 2));
  // append this run to rolling history (keep last 120 runs)
  fs.appendFileSync(HIST, JSON.stringify({ ts: nowTs, counts: curCounts }) + '\n');
  const trimmed = fs.readFileSync(HIST, 'utf8').trim().split('\n').filter(Boolean);
  if (trimmed.length > 120) fs.writeFileSync(HIST, trimmed.slice(-120).join('\n') + '\n');
  fs.mkdirSync(OUT, { recursive: true });
  const anyErr = sources.some(s => s.status === 'ERRORED'); // excludes known-broken
  const anyWarnSrc = sources.some(s => s.status === 'STALE' || s.status === 'BROKEN');
  const anyBleed = results.some(r => r.status === 'BLEED');
  const worst = (results.some(r => r.status === 'LOW') || anyErr) ? 'FAIL'
    : (results.some(r => r.status === 'STALE') || anyWarnSrc || anyBleed) ? 'WARN' : 'OK';
  fs.writeFileSync(path.join(OUT, 'latest.json'), JSON.stringify({ verdict: worst, firstRun, results, sources }, null, 2));

  // report
  console.log(`\nLA data canary — ${firstRun ? 'BASELINE established' : 'verdict ' + worst}\n`);
  console.log('  ROW COUNTS + freshness:');
  for (const r of results) {
    const mark = r.status === 'OK' ? '✔' : r.status === 'STALE' ? '⏳' : r.status === 'BLEED' ? '↓' : '✖';
    const fresh = r.ageDays != null ? `  fresh ${r.ageDays}d` : '';
    console.log(`  ${mark} ${r.table.padEnd(30)} ${r.count.toLocaleString().padStart(12)}${fresh}${r.reason ? '  — ' + r.reason : ''}`);
  }
  console.log('\n  SOURCE last-ingest (la_ingest_state):');
  for (const s of sources) {
    const mark = s.status === 'OK' ? '✔' : s.status === 'STALE' ? '⏳' : s.status === 'BROKEN' ? '⚠' : '✖';
    const age = s.ageDays != null ? `${s.ageDays}d ago` : 'never';
    const detail = s.status === 'OK' ? '' : '  — ' + (s.status === 'BROKEN' ? `known-broken: ${s.note}` : s.status === 'ERRORED' ? s.last_status : 'stale');
    console.log(`  ${mark} ${s.source.padEnd(30)} ${age.padStart(10)}${detail}`);
  }
  console.log('');
}

main().catch(e => { console.error('canary error:', e.message); process.exitCode = 1; }).finally(() => pool.end());