← 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
M RUNBOOK-GOLIVE.mdM data/latest.jsonM db/schema.sqlM package.jsonM scripts/ingest-calaccess-political.mjsM scripts/ingest-fec-ca.mjsA scripts/refresh-political-agg.mjsM server.js
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 →