`scripts/refresh-political-agg.mjs`

101 lines
// Rebuild the materialized k-anon political rollup (political_agg) from
// political_contributions. Precomputes SUM/COUNT/distinct-donor per group for all
// four dimensions so /api/political/agg reads an indexed table instead of a
// 15.7M-row full scan. The k-anon floor is BAKED IN via HAVING count(distinct
// donor_name) >= 5, so political_agg can never contain a group of < 5 donors.
//
// ATOMIC: builds every dimension into a temp table, then swaps into political_agg
// inside ONE transaction (TRUNCATE + INSERT ... SELECT), so a reader never sees a
// half-built table. Runs standalone (`node scripts/refresh-political-agg.mjs`) and
// is called at the END of every political ingest.
//
// $0 — local aggregation, no paid API.

import { pool } from '../lib/db.js';

const POL_CLEAN =
  `donor_name IS NOT NULL AND donor_name <> '' AND (contribution_date IS NULL OR contribution_date <= CURRENT_DATE)`;

// dimension -> source column. MUST match server.js AGG_GROUP.
const DIMENSIONS = {
  recipient:    'recipient_name',
  employer:     'donor_employer',
  city:         'donor_city',
  jurisdiction: 'jurisdiction',
};

export async function refreshPoliticalAgg(client) {
  const own = !client;
  const c = client || (await pool.connect());
  try {
    await c.query('BEGIN');
    // Build every dimension into a temp table first (heavy scans stay outside the
    // swap window as much as possible; the swap itself is TRUNCATE + INSERT).
    await c.query(`DROP TABLE IF EXISTS political_agg_build`);
    await c.query(`
      CREATE TEMP TABLE political_agg_build (
        dimension          TEXT NOT NULL,
        group_key          TEXT NOT NULL,
        total_amount       NUMERIC(14,2),
        contribution_count BIGINT,
        distinct_donors    INT
      ) ON COMMIT DROP`);

    for (const [dimension, col] of Object.entries(DIMENSIONS)) {
      await c.query(
        `INSERT INTO political_agg_build
           (dimension, group_key, total_amount, contribution_count, distinct_donors)
         SELECT $1 AS dimension,
                ${col} AS group_key,
                sum(amount)::numeric(14,2)      AS total_amount,
                count(*)::bigint                AS contribution_count,
                count(distinct donor_name)::int AS distinct_donors
         FROM political_contributions
         WHERE ${POL_CLEAN} AND ${col} IS NOT NULL AND ${col} <> ''
         GROUP BY ${col}
         HAVING count(distinct donor_name) >= 5`,
        [dimension],
      );
    }

    // Swap: clear the served table and repopulate from the build table + stamp.
    await c.query(`TRUNCATE political_agg`);
    await c.query(`
      INSERT INTO political_agg
        (dimension, group_key, total_amount, contribution_count, distinct_donors, refreshed_at)
      SELECT dimension, group_key, total_amount, contribution_count, distinct_donors, now()
      FROM political_agg_build`);
    await c.query('COMMIT');
  } catch (e) {
    try { await c.query('ROLLBACK'); } catch (_) {}
    throw e;
  } finally {
    if (own) c.release();
  }

  const counts = await pool.query(
    `SELECT dimension, count(*)::int AS groups, min(distinct_donors)::int AS min_donors
     FROM political_agg GROUP BY dimension ORDER BY dimension`);
  return counts.rows;
}

// Standalone entrypoint.
if (import.meta.url === `file://${process.argv[1]}`) {
  (async () => {
    try {
      const t0 = Date.now();
      const rows = await refreshPoliticalAgg();
      console.log(`political_agg refreshed in ${((Date.now() - t0) / 1000).toFixed(1)}s:`);
      for (const r of rows) {
        console.log(`  ${r.dimension.padEnd(12)} ${String(r.groups).padStart(8)} groups   min_donors=${r.min_donors}`);
      }
      console.log('$0 (local aggregation).');
    } catch (e) {
      console.error('FAILED:', e.stack || e.message);
      process.exitCode = 1;
    } finally {
      await pool.end();
    }
  })();
}