[object Object]

← back to Ca Donations

ca-donations: materialized k-anon political rollup (political_agg) — agg endpoint fast + Basic-Auth gated (not public), refreshes on ingest

e807f8e51c25ebde7cc462a2c9d7cebd0fc0b97b · 2026-08-24 12:17:09 -0700 · Steve Abrams

Files touched

Diff

commit e807f8e51c25ebde7cc462a2c9d7cebd0fc0b97b
Author: Steve Abrams <steve@designerwallcoverings.com>
Date:   Mon Aug 24 12:17:09 2026 -0700

    ca-donations: materialized k-anon political rollup (political_agg) — agg endpoint fast + Basic-Auth gated (not public), refreshes on ingest
---
 RUNBOOK-GOLIVE.md                      |  25 ++++++---
 data/latest.json                       |  20 +++----
 db/schema.sql                          |  19 +++++++
 package.json                           |   3 +-
 scripts/ingest-calaccess-political.mjs |   5 ++
 scripts/ingest-fec-ca.mjs              |   5 ++
 scripts/refresh-political-agg.mjs      | 100 +++++++++++++++++++++++++++++++++
 server.js                              |  67 +++++++++++-----------
 8 files changed, 191 insertions(+), 53 deletions(-)

diff --git a/RUNBOOK-GOLIVE.md b/RUNBOOK-GOLIVE.md
index b40dc6d..b282bfa 100644
--- a/RUNBOOK-GOLIVE.md
+++ b/RUNBOOK-GOLIVE.md
@@ -13,11 +13,20 @@ charitable_orgs 161K, charitable_grants 40.5K.
 
 ---
 
-## 0. PRE-FLIGHT — materialize the political rollup BEFORE any public-political flag
-`PUBLIC_POLITICAL_AGG` stays OFF at go-live (charitable-only public per the conditioned plan).
-Do NOT enable it until this exists — the live k-anon agg is a 15.7M-row full scan (~174s, guarded to 503).
-When/if you want aggregate political public, build a materialized rollup refreshed on ingest first
-(TK-10782 follow-up). Charitable go-live does NOT need this.
+## 0. Political rollup — EXISTS, refreshes on ingest (TK-10782, done)
+The materialized k-anon rollup `political_agg` now EXISTS (schema.sql) and is rebuilt
+ATOMICALLY on every political ingest (`scripts/refresh-political-agg.mjs`, called at the
+end of ingest-calaccess-political.mjs + ingest-fec-ca.mjs). It bakes the k-anon floor in
+(`HAVING count(distinct donor_name) >= 5`), so `/api/political/agg` reads an indexed table
+(sub-second, ~0.4s — the old 503 from the 15.7M-row live scan is gone). ~39.4K groups across
+recipient/employer/city/jurisdiction; min(distinct_donors) = 5, zero groups < 5.
+
+**Go-live posture (Steve's decision):** aggregate political is served behind the standard
+un/pw (`admin` / `DW2024!`), NOT public. The public surface is **charitable-only**.
+`/api/political/agg` is NOT in the public allowlist — it returns 401 without creds even
+when `PUBLIC_TIER=1`, and 200 (fast) WITH creds, exactly like raw `/api/political`. The
+`PUBLIC_POLITICAL_AGG` flag has been retired — the aggregate is simply an authenticated
+feature now. There is no "enable political agg publicly" step; do NOT set any such flag.
 
 ## 1. DB migration (Mac2 → Kamatera)
 ```sh
@@ -41,7 +50,8 @@ ssh root@45.61.58.125 'cd /root/public-projects/ca-donations && \
   PUBLIC_TIER=1 PORT=9926 pm2 start server.js --name ca-donations --update-env || \
   (pm2 set ca-donations && PUBLIC_TIER=1 pm2 restart ca-donations --update-env); \
   pm2 save'
-# NOTE: leave PUBLIC_POLITICAL_AGG UNSET (off). Set BASIC_AUTH_PASS to a fresh value for the still-gated political tier.
+# NOTE: PUBLIC_POLITICAL_AGG is retired — political (raw AND aggregate) stays Basic-Auth gated regardless.
+# Set BASIC_AUTH_PASS to a fresh value for the gated political tier.
 ```
 
 ## 3. nginx (IP-bound) + webroot cert — per standing rules
@@ -92,7 +102,8 @@ launchctl list | grep ca-donations-loader
 ---
 
 ### Rollback (if a go-live step misbehaves)
-- App code: `git -C ~/Projects/ca-donations revert 80ce7b1 b285cb4` (returns to fully-gated).
+- App code (public tier): `git -C ~/Projects/ca-donations revert 80ce7b1 b285cb4` (returns to fully-gated).
+- Political rollup (TK-10782): `git -C ~/Projects/ca-donations revert <sha>` then `psql -h /tmp -d ca_donations -c 'DROP TABLE political_agg'` (the endpoint reverts to the live-scan/503 version).
 - Public origin: `ssh root@45.61.58.125 'pm2 restart ca-donations' ` after unsetting PUBLIC_TIER, or `pm2 delete`.
 - DNS: repoint the `donations` A record away from 45.61.58.125.
 - Loader job: `launchctl bootout gui/$(id -u)/com.steve.ca-donations-loader`.
diff --git a/data/latest.json b/data/latest.json
index 355140e..04b03c4 100644
--- a/data/latest.json
+++ b/data/latest.json
@@ -1,24 +1,24 @@
 {
   "service": "ca-donations-freshness",
-  "verdict": "WARN",
-  "status": "WARN",
-  "checked_at": "2026-08-24T18:50:23.657Z",
+  "verdict": "PASS",
+  "status": "PASS",
+  "checked_at": "2026-08-24T19:16:22.765Z",
   "findings": [
     {
       "source": "calaccess",
       "table": "political_contributions",
-      "rows": 6947645,
-      "last_ok": "2026-08-21T15:30:22.566Z",
-      "age_days": 3.1,
-      "verdict": "WARN",
-      "why": "stale: last ok 3.1d ago (> 3d)"
+      "rows": 6947672,
+      "last_ok": "2026-08-24T18:59:33.650Z",
+      "age_days": 0,
+      "verdict": "PASS",
+      "why": ""
     },
     {
       "source": "fec_bulk",
       "table": "political_contributions",
       "rows": 8750471,
       "last_ok": "2026-08-23T00:53:43.362Z",
-      "age_days": 1.7,
+      "age_days": 1.8,
       "verdict": "PASS",
       "why": ""
     },
@@ -27,7 +27,7 @@
       "table": "charitable_grants",
       "rows": 40514,
       "last_ok": "2026-08-21T15:28:40.378Z",
-      "age_days": 3.1,
+      "age_days": 3.2,
       "verdict": "PASS",
       "why": ""
     },
diff --git a/db/schema.sql b/db/schema.sql
index 48fb3c1..04fbe05 100644
--- a/db/schema.sql
+++ b/db/schema.sql
@@ -100,3 +100,22 @@ CREATE TABLE IF NOT EXISTS charitable_grants (
 CREATE INDEX IF NOT EXISTS cg_grantor_idx ON charitable_grants (grantor_ein);
 CREATE INDEX IF NOT EXISTS cg_grantee_idx ON charitable_grants USING gin (to_tsvector('simple', coalesce(grantee_name,'')));
 CREATE INDEX IF NOT EXISTS cg_year_idx    ON charitable_grants (tax_year);
+
+-- ---------------------------------------------------------------------------
+-- POLITICAL AGG — materialized k-anon rollup of political_contributions.
+-- Precomputed so /api/political/agg reads an indexed table (sub-second) instead
+-- of a 15.7M-row full-table GROUP BY (~174s, guarded to 503). refresh-political-agg.mjs
+-- rebuilds it atomically on every ingest. The k-anon floor is BAKED IN: the
+-- populate step enforces HAVING count(distinct donor_name) >= 5, so this table
+-- can NEVER contain a group that singles out fewer than 5 donors.
+-- ---------------------------------------------------------------------------
+CREATE TABLE IF NOT EXISTS political_agg (
+  dimension          TEXT NOT NULL,           -- recipient | employer | city | jurisdiction
+  group_key          TEXT NOT NULL,
+  total_amount       NUMERIC(14,2),
+  contribution_count BIGINT,
+  distinct_donors    INT,
+  refreshed_at       TIMESTAMPTZ,
+  UNIQUE (dimension, group_key)
+);
+CREATE INDEX IF NOT EXISTS pa_dim_amount_idx ON political_agg (dimension, total_amount DESC);
diff --git a/package.json b/package.json
index c9143d6..eaf71d2 100644
--- a/package.json
+++ b/package.json
@@ -9,7 +9,8 @@
     "start": "node server.js",
     "schema": "psql -h /tmp -d ca_donations -f db/schema.sql",
     "ingest:propublica": "node scripts/ingest-propublica-ca-orgs.mjs",
-    "ingest:ca-ag": "node scripts/ingest-ca-ag-status.mjs"
+    "ingest:ca-ag": "node scripts/ingest-ca-ag-status.mjs",
+    "refresh:political-agg": "node scripts/refresh-political-agg.mjs"
   },
   "dependencies": {
     "adm-zip": "^0.6.0",
diff --git a/scripts/ingest-calaccess-political.mjs b/scripts/ingest-calaccess-political.mjs
index 2fa0db0..6893c52 100644
--- a/scripts/ingest-calaccess-political.mjs
+++ b/scripts/ingest-calaccess-political.mjs
@@ -25,6 +25,7 @@ import readline from 'node:readline';
 import { fileURLToPath } from 'node:url';
 import unzipper from 'unzipper';
 import { q, startRun, upsertSource, pool } from '../lib/db.js';
+import { refreshPoliticalAgg } from './refresh-political-agg.mjs';
 
 const __dirname = path.dirname(fileURLToPath(import.meta.url));
 const RAW = path.join(__dirname, '..', 'data', 'raw');
@@ -229,6 +230,10 @@ async function main() {
 
     if (batch.length) upserted += await flush(batch);
 
+    console.log('Refreshing political_agg rollup...');
+    const aggCounts = await refreshPoliticalAgg();
+    for (const r of aggCounts) console.log(`  agg ${r.dimension}: ${r.groups} groups (min_donors=${r.min_donors})`);
+
     await run.done(seen, upserted, 'ok',
       `kept ${kept} (>=${SINCE}); skippedOld ${skippedOld}; skippedNoTran ${skippedNoTran}`);
     console.log(`CAL-ACCESS RCPT: seen ${seen}, kept ${kept} (>=${SINCE}), upserted ${upserted}. $0 (local bulk).`);
diff --git a/scripts/ingest-fec-ca.mjs b/scripts/ingest-fec-ca.mjs
index aa486df..29ac7a5 100644
--- a/scripts/ingest-fec-ca.mjs
+++ b/scripts/ingest-fec-ca.mjs
@@ -20,6 +20,7 @@ import readline from 'node:readline';
 import { fileURLToPath } from 'node:url';
 import unzipper from 'unzipper';
 import { q, startRun, upsertSource, pool } from '../lib/db.js';
+import { refreshPoliticalAgg } from './refresh-political-agg.mjs';
 
 const __dirname = path.dirname(fileURLToPath(import.meta.url));
 const RAW = path.join(__dirname, '..', 'data', 'raw');
@@ -175,6 +176,10 @@ async function main() {
 
     if (batch.length) upserted += await flush(batch);
 
+    console.log('Refreshing political_agg rollup...');
+    const aggCounts = await refreshPoliticalAgg();
+    for (const r of aggCounts) console.log(`  agg ${r.dimension}: ${r.groups} groups (min_donors=${r.min_donors})`);
+
     await run.done(seen, upserted, 'ok', `CA kept ${kept} of ${seen}; skippedNoSub ${skippedNoSub}`);
     console.log(`FEC indiv${CYCLE}: seen ${seen}, CA kept ${kept}, upserted ${upserted}. $0 (free bulk).`);
   } catch (e) {
diff --git a/scripts/refresh-political-agg.mjs b/scripts/refresh-political-agg.mjs
new file mode 100644
index 0000000..8b18643
--- /dev/null
+++ b/scripts/refresh-political-agg.mjs
@@ -0,0 +1,100 @@
+// 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();
+    }
+  })();
+}
diff --git a/server.js b/server.js
index 51cf97a..93c437e 100644
--- a/server.js
+++ b/server.js
@@ -2,7 +2,7 @@
 // Two families: charitable (orgs + grants) and political (donor-level contributions).
 import express from 'express';
 import path from 'node:path';
-import { q, qWithTimeout } from './lib/db.js';
+import { q } from './lib/db.js';
 
 const app = express();
 const PORT = process.env.PORT || 9926;
@@ -10,10 +10,11 @@ const USER = process.env.BASIC_AUTH_USER || 'admin';
 const PASS = process.env.BASIC_AUTH_PASS || 'DW2024!';
 
 // Public tier is OFF by default — when unset/'0' the server is byte-for-byte the
-// original fail-closed Basic-Auth app. Political AGG is a SECOND independent flag:
-// even with the tier on, aggregate political stays dark until it too is enabled.
+// original fail-closed Basic-Auth app. The public surface is charitable-only:
+// ALL political endpoints (raw /api/political AND aggregate /api/political/agg)
+// require Basic-Auth in every mode. The aggregate is fast (reads political_agg)
+// but authenticated — not public.
 const PUBLIC_TIER = process.env.PUBLIC_TIER === '1';
-const PUBLIC_POLITICAL_AGG = process.env.PUBLIC_POLITICAL_AGG === '1';
 
 // STRICT POSITIVE ALLOWLIST. There is deliberately NO "anything not under /api/"
 // fallback: the gate opens a path only when it EXACTLY matches a known-public API
@@ -21,14 +22,13 @@ const PUBLIC_POLITICAL_AGG = process.env.PUBLIC_POLITICAL_AGG === '1';
 // falls through returns false and stays Basic-Auth gated — so a future file like
 // public/political-export.json can never be silently served to the public.
 
-// Exact-match public charitable/org API endpoints (raw political donor rows are
-// never on this list). /api/political/agg is exact-match AND still gated behind
-// PUBLIC_POLITICAL_AGG. Org detail is the one prefix (/api/org/:ein).
+// Exact-match public charitable/org API endpoints. NO political endpoint is on
+// this list — raw /api/political AND aggregate /api/political/agg both require
+// Basic-Auth in all modes. Org detail is the one prefix (/api/org/:ein).
 const PUBLIC_API_EXACT = new Set(['/api/stats', '/api/orgs', '/api/grants']);
 const publicApiPath = (p) =>
   PUBLIC_API_EXACT.has(p) ||
-  p.startsWith('/api/org/') ||
-  (PUBLIC_POLITICAL_AGG && p === '/api/political/agg');
+  p.startsWith('/api/org/');
 
 // Genuine static assets only: no /api prefix, a known-safe extension, and the
 // resolved path must stay inside public/ (blocks traversal / encoded escapes).
@@ -173,38 +173,35 @@ app.get('/api/political', async (req, res) => {
   } catch (e) { res.status(500).json({ error: e.message }); }
 });
 
-// Aggregate-only political rollups — NO per-donor rows ever leave here. Each row
-// is a group SUM(amount) + COUNT(*), and the HAVING floor drops any group with
-// fewer than 5 distinct donors so a group can never single out one person.
-// Gated behind BOTH flags: only reachable when PUBLIC_TIER + PUBLIC_POLITICAL_AGG.
-const AGG_GROUP = {
-  recipient:    'recipient_name',
-  employer:     'donor_employer',
-  city:         'donor_city',
-  jurisdiction: 'jurisdiction',
+// Aggregate-only political rollups — NO per-donor rows ever leave here. Served
+// from the materialized political_agg table (indexed, sub-second), which the
+// ingest bakes with a HAVING count(distinct donor_name) >= 5 k-anon floor so a
+// group can never single out fewer than 5 people. AUTHENTICATED, not public:
+// this endpoint is NOT in the allowlist above, so it requires Basic-Auth in every
+// mode exactly like raw /api/political — the public surface is charitable-only.
+const AGG_DIMENSIONS = {
+  recipient:    1,
+  employer:     1,
+  city:         1,
+  jurisdiction: 1,
 };
+const AGG_SORT = { amount: 'total_amount DESC NULLS LAST', count: 'contribution_count DESC NULLS LAST' };
 app.get('/api/political/agg', async (req, res) => {
   try {
     const by = req.query.by;
-    if (typeof by !== 'string' || !Object.hasOwn(AGG_GROUP, by))
-      return res.status(400).json({ error: 'by must be one of: ' + Object.keys(AGG_GROUP).join('|') });
-    const col = AGG_GROUP[by];
+    if (typeof by !== 'string' || !Object.hasOwn(AGG_DIMENSIONS, by))
+      return res.status(400).json({ error: 'by must be one of: ' + Object.keys(AGG_DIMENSIONS).join('|') });
+    const sort = typeof req.query.sort === 'string' && Object.hasOwn(AGG_SORT, req.query.sort)
+      ? req.query.sort : 'amount';
     const limit = Math.min(+req.query.limit || 100, 500);
-    // Bounded — a full-table GROUP BY over ~15M rows must cancel cleanly, not hang.
-    const rows = await qWithTimeout(
-      `SELECT ${col} AS group_key,
-              sum(amount)::numeric(14,2)     AS total_amount,
-              count(*)::int                  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
-       ORDER BY sum(amount) DESC NULLS LAST
-       LIMIT $1`, [limit], 8000);
-    res.json({ by: req.query.by, k_anonymity_floor: 5, rows });
+    const rows = await q(
+      `SELECT group_key, total_amount, contribution_count, distinct_donors
+       FROM political_agg
+       WHERE dimension = $1
+       ORDER BY ${AGG_SORT[sort]}
+       LIMIT $2`, [by, limit]);
+    res.json({ by, sort, k_anonymity_floor: 5, rows });
   } catch (e) {
-    if (e.code === '57014') return res.status(503).json({ error: 'aggregate timed out' });
     res.status(500).json({ error: e.message });
   }
 });

← 15aa128 ca-donations: gated go-live runbook (Kamatera migration + ng  ·  back to Ca Donations  ·  auto-data-snapshot: 2026-08-25T09:32:22 (1 data files) — dat 0ec0654 →