`lib/db.js`

58 lines
// Shared pg pool — local dw-style Postgres on the /tmp socket.
import pg from 'pg';

export const pool = new pg.Pool({
  host: process.env.PGHOST || '/tmp',
  database: process.env.PGDATABASE || 'ca_donations',
  max: 8,
});

export async function q(text, params) {
  const res = await pool.query(text, params);
  return res.rows;
}

// Run a query with a bounded statement_timeout on a dedicated client, so a heavy
// aggregation (e.g. the public political rollup over the full table) cancels
// cleanly instead of tying up a connection. The timeout is session-local and the
// client is reset before returning to the pool.
export async function qWithTimeout(text, params, ms = 8000) {
  const client = await pool.connect();
  try {
    await client.query(`SET statement_timeout = ${Number(ms)}`);
    const res = await client.query(text, params);
    return res.rows;
  } finally {
    try { await client.query('SET statement_timeout = 0'); } catch (_) {}
    client.release();
  }
}

// Begin/finish an ingest_runs audit row. Returns { id, done(rowsIn, rowsUpsert, status, detail) }.
export async function startRun(sourceSlug) {
  const [row] = await q(
    `INSERT INTO ingest_runs (source_slug) VALUES ($1) RETURNING id`,
    [sourceSlug],
  );
  return {
    id: row.id,
    async done(rowsIn, rowsUpsert, status = 'ok', detail = null) {
      await q(
        `UPDATE ingest_runs SET finished_at=now(), rows_in=$2, rows_upsert=$3, status=$4, detail=$5 WHERE id=$1`,
        [row.id, rowsIn, rowsUpsert, status, detail],
      );
    },
  };
}

export async function upsertSource(s) {
  await q(
    `INSERT INTO sources (slug,name,url,jurisdiction,family,granularity,access,notes)
     VALUES ($1,$2,$3,$4,$5,$6,$7,$8)
     ON CONFLICT (slug) DO UPDATE SET name=EXCLUDED.name, url=EXCLUDED.url,
       jurisdiction=EXCLUDED.jurisdiction, family=EXCLUDED.family,
       granularity=EXCLUDED.granularity, access=EXCLUDED.access, notes=EXCLUDED.notes`,
    [s.slug, s.name, s.url, s.jurisdiction, s.family, s.granularity, s.access, s.notes],
  );
}