api/batch_jobs/batch_job_db.py

"""
Postgres persistence for YouTube pipeline batch jobs (real-time dashboard).

Table: ``bronze.youtube_batch_job_runs`` (migration 073).
"""

from __future__ import annotations

import json
import logging
import os
from pathlib import Path
from typing import Any, Dict, List, Optional

from psycopg2.extras import Json, RealDictCursor

from api.batch_jobs.batch_job_status import BatchJob, BatchJobStore, list_batches

logger = logging.getLogger(__name__)

_ENSURED = False


def _use_db() -> bool:
    return os.getenv("BATCH_JOBS_USE_DB", "1").strip().lower() not in (
        "0",
        "false",
        "no",
        "off",
    )


def get_db_connection():
    import psycopg2

    from core_lib.db import resolve_target_database_url

    url = resolve_target_database_url()
    for bad in ("&channel_binding=require", "channel_binding=require&", "channel_binding=require"):
        url = url.replace(bad, "")
    url = url.replace("&&", "&").rstrip("?&")
    return psycopg2.connect(
        url,
        connect_timeout=int(os.getenv("PGCONNECT_TIMEOUT", "10")),
        options=os.getenv("PGOPTIONS", "-c statement_timeout=60000"),
    )


def ensure_batch_job_tables(conn: Any) -> None:
    global _ENSURED
    if _ENSURED:
        return
    cur = conn.cursor()
    try:
        cur.execute(
            """
            CREATE TABLE IF NOT EXISTS bronze.youtube_batch_job_runs (
                batch_id    VARCHAR(128) PRIMARY KEY,
                step        VARCHAR(32)  NOT NULL,
                status      VARCHAR(32)  NOT NULL,
                started_at  TIMESTAMPTZ,
                updated_at  TIMESTAMPTZ  NOT NULL DEFAULT CURRENT_TIMESTAMP,
                finished_at TIMESTAMPTZ,
                config      JSONB        NOT NULL DEFAULT '{}',
                summary     JSONB        NOT NULL DEFAULT '{}',
                payload     JSONB        NOT NULL
            )
            """
        )
        cur.execute(
            """
            CREATE INDEX IF NOT EXISTS idx_youtube_batch_job_runs_updated
                ON bronze.youtube_batch_job_runs (updated_at DESC)
            """
        )
        cur.execute(
            """
            CREATE INDEX IF NOT EXISTS idx_youtube_batch_job_runs_running
                ON bronze.youtube_batch_job_runs (status)
                WHERE status = 'running'
            """
        )
    finally:
        cur.close()
    conn.commit()
    _ENSURED = True


def upsert_batch_job(conn: Any, job: BatchJob) -> None:
    ensure_batch_job_tables(conn)
    payload = job.to_dict()
    cur = conn.cursor()
    try:
        cur.execute(
            """
            INSERT INTO bronze.youtube_batch_job_runs (
                batch_id, step, status, started_at, updated_at, finished_at,
                config, summary, payload
            ) VALUES (
                %(batch_id)s, %(step)s, %(status)s,
                %(started_at)s::timestamptz, %(updated_at)s::timestamptz,
                NULLIF(%(finished_at)s, '')::timestamptz,
                %(config)s, %(summary)s, %(payload)s
            )
            ON CONFLICT (batch_id) DO UPDATE SET
                step = EXCLUDED.step,
                status = EXCLUDED.status,
                started_at = EXCLUDED.started_at,
                updated_at = EXCLUDED.updated_at,
                finished_at = EXCLUDED.finished_at,
                config = EXCLUDED.config,
                summary = EXCLUDED.summary,
                payload = EXCLUDED.payload
            """,
            {
                "batch_id": job.batch_id,
                "step": job.step,
                "status": job.status,
                "started_at": job.started_at or None,
                "updated_at": job.updated_at,
                "finished_at": job.finished_at or "",
                "config": Json(job.config or {}),
                "summary": Json(job.summary or {}),
                "payload": Json(payload),
            },
        )
    finally:
        cur.close()
    conn.commit()


def sync_batch_job_to_db(job: BatchJob) -> None:
    if not _use_db():
        return
    try:
        with get_db_connection() as conn:
            upsert_batch_job(conn, job)
    except Exception as exc:
        logger.warning("batch job DB sync failed for %s: %s", job.batch_id, exc)


def list_batch_jobs_from_db(*, limit: int = 100) -> List[BatchJob]:
    with get_db_connection() as conn:
        ensure_batch_job_tables(conn)
        with conn.cursor(cursor_factory=RealDictCursor) as cur:
            cur.execute(
                """
                SELECT payload
                FROM bronze.youtube_batch_job_runs
                ORDER BY updated_at DESC NULLS LAST
                LIMIT %s
                """,
                (limit,),
            )
            rows = cur.fetchall()
    jobs: List[BatchJob] = []
    for row in rows:
        payload = row["payload"]
        if isinstance(payload, str):
            payload = json.loads(payload)
        jobs.append(BatchJob.from_dict(payload))
    return jobs


def reap_stale_running_batches(*, limit: int = 30) -> int:
    """Cancel ``running`` batch rows whose recorded activity has gone stale.

    The dashboard counts ``status = 'running'`` rows, but a row stays ``running``
    after its worker dies or starts skipping already-done work — the per-job
    stale-cancel (``apply_batch_lifecycle`` → ``_maybe_stale_cancel_batch``,
    BATCH_JOB_INACTIVITY_SECONDS, default 3600s) only fires when a job is
    otherwise touched, which never happens for an abandoned run. Sweeping it here,
    on the dashboard read path, keeps ``totals.running`` honest instead of pinning
    it at a phantom count. Returns the number of rows whose status changed.
    """
    if not _use_db():
        return 0
    # Lazy import: batch_job_status imports this module, so import its lifecycle
    # helpers at call time to avoid a circular import at module load.
    from api.batch_jobs.batch_job_status import apply_batch_lifecycle, persist_batch_job

    reaped = 0
    try:
        with get_db_connection() as conn:
            ensure_batch_job_tables(conn)
            with conn.cursor(cursor_factory=RealDictCursor) as cur:
                cur.execute(
                    """
                    SELECT payload
                    FROM bronze.youtube_batch_job_runs
                    WHERE status = 'running'
                    ORDER BY updated_at DESC NULLS LAST
                    LIMIT %s
                    """,
                    (limit,),
                )
                rows = cur.fetchall()
        for row in rows:
            payload = row["payload"]
            if isinstance(payload, str):
                payload = json.loads(payload)
            job = BatchJob.from_dict(payload)
            if apply_batch_lifecycle(job):
                persist_batch_job(job)  # writes cancelled/completed status back
                reaped += 1
    except Exception as exc:
        logger.warning("stale batch reap failed: %s", exc)
    return reaped


def policy_event_counts_24h(conn: Any) -> Dict[str, Any]:
    """
    Pipeline counts from the per-event bronze stamps (migration 083):

    - ``analysis`` / ``reports`` / ``*_errors`` — events in the last 24h.
    - ``*_total`` — all-time count of transcripts/analyses/reports (one live,
      de-duplicated source for the "on disk" cards and progress %, instead of
      summing per-batch disk scans, which double-counts transcripts across
      overlapping batches and goes stale for analysis/reports).
    - ``last_*_at`` — most recent stamp per step; drives the "ago" cards and,
      unlike the batch ``updated_at`` clock, reflects standalone analyze runs.

    Returns zeros/empty strings if the columns do not exist yet (migration not
    applied), so the dashboard falls back to the batch-summary counters.
    """
    out: Dict[str, Any] = {
        "analysis": 0,
        "reports": 0,
        "analysis_errors": 0,
        "reports_errors": 0,
        "transcripts_total": 0,
        "analysis_total": 0,
        "reports_total": 0,
        "last_transcript_at": "",
        "last_analysis_at": "",
        "last_report_at": "",
    }
    try:
        with conn.cursor() as cur:
            cur.execute(
                """
                SELECT
                    COUNT(*) FILTER (
                        WHERE policy_analysis_at >= now() - interval '24 hours'
                    )::int AS analysis,
                    COUNT(*) FILTER (
                        WHERE policy_report_at >= now() - interval '24 hours'
                    )::int AS reports,
                    COUNT(*) FILTER (
                        WHERE policy_analysis_error IS NOT NULL
                          AND last_updated >= now() - interval '24 hours'
                    )::int AS analysis_errors,
                    COUNT(*) FILTER (
                        WHERE policy_report_error IS NOT NULL
                          AND last_updated >= now() - interval '24 hours'
                    )::int AS reports_errors,
                    COUNT(*) FILTER (WHERE transcript_download_at IS NOT NULL)::int
                        AS transcripts_total,
                    COUNT(*) FILTER (WHERE policy_analysis_at IS NOT NULL)::int
                        AS analysis_total,
                    COUNT(*) FILTER (WHERE policy_report_at IS NOT NULL)::int
                        AS reports_total,
                    MAX(transcript_download_at) AS last_transcript_at,
                    MAX(policy_analysis_at) AS last_analysis_at,
                    MAX(policy_report_at) AS last_report_at
                FROM bronze.bronze_event_youtube
                """
            )
            row = cur.fetchone()
        if row:
            out["analysis"] = int(row[0] or 0)
            out["reports"] = int(row[1] or 0)
            out["analysis_errors"] = int(row[2] or 0)
            out["reports_errors"] = int(row[3] or 0)
            out["transcripts_total"] = int(row[4] or 0)
            out["analysis_total"] = int(row[5] or 0)
            out["reports_total"] = int(row[6] or 0)
            out["last_transcript_at"] = row[7].isoformat() if row[7] is not None else ""
            out["last_analysis_at"] = row[8].isoformat() if row[8] is not None else ""
            out["last_report_at"] = row[9].isoformat() if row[9] is not None else ""
    except Exception as exc:
        # Columns missing (pre-083) or transient DB error — degrade to zeros so the
        # dashboard falls back to the disk-scan counters.
        conn.rollback()
        logger.debug("policy_event_counts_24h unavailable: %s", exc)
    return out


# Pipeline stages, in funnel order. Each (scope, stage) pair is one report row,
# so adding a metric is a field and adding a dimension (state, later jurisdiction)
# is more rows — not more columns.
PIPELINE_STAGES = ("discover", "videos", "transcripts", "analyses", "reports")


def _stage_timing(conn: Any) -> Dict[str, Any]:
    """Per-stage cadence: median seconds between recent completions (~throughput
    per file) and the most recent output file path. Column names are fixed (not
    user input), so the f-string is safe. Empty on pre-083 DBs.

    ``avg_seconds`` is the median gap between *completed* rows, so it freezes
    during a stall — no new row means no new gap, and gaps >= 1h are filtered
    out anyway. To keep ``/hr`` and ETA honest, we also return:

    - ``stale_seconds``: the open trailing gap (``now() - last_at``), i.e. how
      long we have already been waiting for the next completion.
    - ``effective_seconds``: ``max(avg_seconds, stale_seconds)`` — once the next
      file is overdue, the best estimate of the current per-file pace is at
      least how long we have waited. It degrades while stalled and snaps back to
      ``avg_seconds`` the moment a file lands. Consumers showing a live rate for
      an idle (not-running) stage should ignore ``stale_seconds``.
    """

    def _one(
        ts_col: str, path_col: str, table: str = "bronze.bronze_event_youtube"
    ) -> Dict[str, Any]:
        out: Dict[str, Any] = {
            "avg_seconds": None,
            "last_path": "",
            "last_at": "",
            "stale_seconds": None,
            "effective_seconds": None,
        }
        try:
            with conn.cursor() as cur:
                cur.execute(
                    f"""
                    WITH recent AS (
                        SELECT {ts_col} AS ts, {path_col} AS path
                        FROM {table}
                        WHERE {ts_col} IS NOT NULL
                        ORDER BY {ts_col} DESC LIMIT 150
                    ),
                    gaps AS (
                        SELECT EXTRACT(EPOCH FROM (ts - LAG(ts) OVER (ORDER BY ts))) AS gap
                        FROM recent
                    ),
                    latest AS (
                        SELECT ts, path FROM recent ORDER BY ts DESC LIMIT 1
                    )
                    SELECT
                        (SELECT percentile_cont(0.5) WITHIN GROUP (ORDER BY gap)
                           FROM gaps WHERE gap > 0 AND gap < 3600) AS med_gap,
                        (SELECT path FROM latest) AS last_path,
                        (SELECT ts FROM latest) AS last_at,
                        EXTRACT(EPOCH FROM (now() - (SELECT ts FROM latest))) AS stale_seconds
                    """
                )
                row = cur.fetchone()
            if row:
                out["avg_seconds"] = round(float(row[0]), 1) if row[0] is not None else None
                out["last_path"] = str(row[1] or "")
                out["last_at"] = row[2].isoformat() if row[2] is not None else ""
                out["stale_seconds"] = (
                    round(float(row[3]), 1) if row[3] is not None else None
                )
                avg = out["avg_seconds"]
                stale = out["stale_seconds"]
                if avg is not None:
                    out["effective_seconds"] = (
                        round(max(avg, stale), 1) if stale is not None else avg
                    )
        except Exception:
            conn.rollback()
        return out

    vids = _one("transcript_download_at", "transcript_file_path")
    # Transcripts cadence/last-file come from bronze_event_youtube_transcript (where the
    # backfill actually writes), not the YouTube download stamp — otherwise the
    # dashboard reads "stalled · 0/hr · last file <days ago>" while a DB-only
    # backfill is inserting fine. video_id stands in for the missing file path.
    transcripts = _one(
        "COALESCE(last_updated, created_at)",
        "video_id",
        table="bronze.bronze_event_youtube_transcript",
    )
    return {
        "videos": vids,
        "transcripts": transcripts,
        "analyses": _one("policy_analysis_at", "policy_analysis_path"),
        "reports": _one("policy_report_at", "policy_report_path"),
    }


def pipeline_stage_report(conn: Any) -> Dict[str, Any]:
    """
    Long-format per-state pipeline coverage from the bronze per-event stamps.

    Returns ``{"states": [...], "rows": [{scope, stage, done, total, failed,
    last_at}, ...]}`` where ``scope`` is a 2-letter state code or ``"ALL"`` (the
    national rollup). All four stages are derived from one ``GROUP BY state_code``
    over ``bronze_event_youtube`` so per-state and overall numbers are consistent
    and live (covering standalone/parallel analyze runs). Empty on pre-083 DBs.
    """
    out: Dict[str, Any] = {"states": [], "rows": []}

    def _iso(v: Any) -> str:
        return v.isoformat() if v is not None else ""

    def _max_iso(a: str, b: str) -> str:
        return a if a >= b else b

    try:
        with conn.cursor() as cur:
            cur.execute(
                """
                SELECT
                    COALESCE(NULLIF(state_code, ''), '??')                     AS st,
                    COUNT(*) FILTER (WHERE transcript_download_at IS NOT NULL)::int AS vids_ok,
                    COUNT(*) FILTER (WHERE transcript_file_error IS NOT NULL)::int  AS vids_fail,
                    COUNT(*) FILTER (WHERE policy_analysis_at IS NOT NULL)::int     AS analyses,
                    COUNT(*) FILTER (WHERE policy_analysis_error IS NOT NULL)::int  AS analysis_err,
                    COUNT(*) FILTER (WHERE policy_report_at IS NOT NULL)::int       AS reports,
                    COUNT(*) FILTER (WHERE policy_report_error IS NOT NULL)::int    AS report_err,
                    MAX(transcript_download_at)                                 AS last_video,
                    MAX(policy_analysis_at)                                     AS last_analysis,
                    MAX(policy_report_at)                                       AS last_report
                FROM bronze.bronze_event_youtube
                GROUP BY 1
                """
            )
            db_rows = cur.fetchall()
    except Exception as exc:
        conn.rollback()
        logger.debug("pipeline_stage_report unavailable: %s", exc)
        return out

    # Stage 0 (channel discovery) lives in the scraped-jurisdiction tables, not in
    # bronze_event_youtube: per state, how many jurisdictions have a YouTube channel
    # found (done) out of all scraped (total); failed = still missing a channel.
    # Grouped by entity (counties vs municipalities) so the Discover cell can split
    # the combined coverage; the per-entity rows sum back to the combined total.
    DISCOVER_ENTITIES = ("counties", "municipalities")
    discover_by_state: Dict[str, Dict[str, Any]] = {}
    try:
        with conn.cursor() as cur:
            cur.execute(
                """
                SELECT usps AS st, entity,
                    COUNT(*)::int AS total,
                    COUNT(*) FILTER (
                        WHERE COALESCE(NULLIF(youtube_channel_id, ''),
                                       NULLIF(youtube_channel_url, '')) IS NOT NULL
                    )::int AS done,
                    MAX(discovered_at) AS last_at
                FROM (
                    SELECT usps, youtube_channel_id, youtube_channel_url, discovered_at,
                           'counties' AS entity
                      FROM bronze.bronze_jurisdictions_counties_scraped
                    UNION ALL
                    SELECT usps, youtube_channel_id, youtube_channel_url, discovered_at,
                           'municipalities' AS entity
                      FROM bronze.bronze_jurisdictions_municipalities_scraped
                ) j
                WHERE COALESCE(usps, '') <> ''
                GROUP BY usps, entity
                """
            )
            for st, entity, total, done, last_at in cur.fetchall():
                rec = discover_by_state.setdefault(
                    str(st), {"total": 0, "done": 0, "last_at": "", "breakdown": {}}
                )
                rec["total"] += int(total)
                rec["done"] += int(done)
                rec["last_at"] = _max_iso(rec["last_at"], _iso(last_at))
                rec["breakdown"][str(entity)] = {
                    "entity": str(entity), "total": int(total), "done": int(done),
                }
    except Exception:
        conn.rollback()  # scraped tables absent — discover stage degrades to zeros

    # Transcript coverage per state over the int_events_union candidate set
    # (CivicSearch + YouTube-API + LocalView): total = candidate videos, done =
    # candidates with a transcript landed in bronze_event_youtube_transcript.
    # Counting both from the same population keeps the bar honest — counting
    # done from the transcript table alone gave it no denominator and reported
    # done/done = 100% even with a large backlog (the "118k / 118k" artifact).
    # Videos with no state_code (most CivicSearch school meetings) group under
    # '??'; they get no per-state row but are folded into the ALL total below so
    # the headline coverage isn't overstated.
    tx_by_state: Dict[str, Dict[str, Any]] = {}
    tx_unknown_total = 0
    tx_unknown_done = 0
    try:
        with conn.cursor() as cur:
            cur.execute(
                """
                SELECT
                    COALESCE(NULLIF(u.state_code, ''), '??')   AS st,
                    COUNT(*)::int                              AS total,
                    COUNT(t.video_id)::int                     AS done,
                    MAX(COALESCE(t.last_updated, t.created_at)) AS last_at
                FROM intermediate.int_events_union u
                LEFT JOIN bronze.bronze_event_youtube_transcript t
                    ON t.video_id = u.video_id
                GROUP BY 1
                """
            )
            for st, total, done, last_at in cur.fetchall():
                if str(st) == "??":
                    tx_unknown_total += int(total)
                    tx_unknown_done += int(done)
                else:
                    tx_by_state[str(st)] = {
                        "done": int(done), "total": int(total), "last_at": _iso(last_at),
                    }
    except Exception:
        conn.rollback()  # int_events_union absent — transcript coverage degrades to zeros

    def _discover_row(scope: str, disc: Optional[Dict[str, Any]]) -> Dict[str, Any]:
        d = disc or {"total": 0, "done": 0, "last_at": "", "breakdown": {}}
        bd = d.get("breakdown") or {}
        breakdown = [
            {
                "entity": e,
                "done": int(bd[e]["done"]),
                "total": int(bd[e]["total"]),
                "failed": max(0, int(bd[e]["total"]) - int(bd[e]["done"])),
            }
            for e in DISCOVER_ENTITIES
            if e in bd
        ]
        return {
            "scope": scope, "stage": "discover", "done": d["done"], "total": d["total"],
            "failed": max(0, d["total"] - d["done"]), "last_at": d["last_at"],
            "breakdown": breakdown,
        }

    def _stage_rows(
        scope: str,
        rec: Dict[str, Any],
        disc: Optional[Dict[str, Any]],
        tx: Dict[str, Any],
    ) -> List[Dict[str, Any]]:
        ok = int(rec["vids_ok"])
        fail = int(rec["vids_fail"])
        attempted = ok + fail
        lv, la, lr = rec["last_video"], rec["last_analysis"], rec["last_report"]
        # videos: YouTube download stamps (ok / attempted). transcripts: actual
        # landed coverage from bronze_event_youtube_transcript over the int_events_union
        # candidate set (done = with transcript, total = candidates) — reaches
        # LocalView/union videos the YouTube stamp never counted. analyses/reports
        # sit below transcripts in the funnel, so they are denominated by the
        # transcripts that exist (clamped so a stamp count can't exceed total).
        tx_done = int(tx.get("done", 0))
        tx_total = max(int(tx.get("total", 0)), tx_done)
        tx_last = tx.get("last_at") or ""
        analyses = int(rec["analyses"])
        reports = int(rec["reports"])
        return [
            _discover_row(scope, disc),
            {"scope": scope, "stage": "videos", "done": ok, "total": attempted,
             "failed": fail, "last_at": lv},
            {"scope": scope, "stage": "transcripts", "done": tx_done, "total": tx_total,
             "failed": fail, "last_at": tx_last or lv},
            {"scope": scope, "stage": "analyses", "done": analyses,
             "total": max(tx_done, analyses),
             "failed": int(rec["analysis_err"]), "last_at": la},
            {"scope": scope, "stage": "reports", "done": reports,
             "total": max(tx_done, reports),
             "failed": int(rec["report_err"]), "last_at": lr},
        ]

    cols = ("st", "vids_ok", "vids_fail", "analyses", "analysis_err", "reports",
            "report_err", "last_video", "last_analysis", "last_report")
    yt_by_state: Dict[str, Dict[str, Any]] = {}
    for r in db_rows:
        rec = dict(zip(cols, r))
        for tk in ("last_video", "last_analysis", "last_report"):
            rec[tk] = _iso(rec[tk])
        yt_by_state[str(rec["st"])] = rec

    _zero_rec = {
        "vids_ok": 0, "vids_fail": 0, "analyses": 0, "analysis_err": 0,
        "reports": 0, "report_err": 0,
        "last_video": "", "last_analysis": "", "last_report": "",
    }

    rows: List[Dict[str, Any]] = []
    states: List[str] = []
    totals = {k: 0 for k in ("vids_ok", "vids_fail", "analyses", "analysis_err",
                             "reports", "report_err")}
    last = {"last_video": "", "last_analysis": "", "last_report": ""}
    tx_totals = {"done": 0, "total": 0, "last_at": ""}

    # Surface every state that has entered the pipeline (any attempted YouTube
    # video) OR that has at least one landed transcript — the latter covers
    # LocalView/union states with no bronze_event_youtube rows.
    candidate_states = {
        st for st, rec in yt_by_state.items()
        if int(rec["vids_ok"]) + int(rec["vids_fail"]) > 0
    } | {
        st for st, tx in tx_by_state.items()
        if int(tx.get("done", 0)) > 0 or int(tx.get("total", 0)) > 0
    }
    candidate_states.discard("??")

    for st in sorted(candidate_states):
        rec = yt_by_state.get(st, dict(_zero_rec))
        tx = tx_by_state.get(st, {"done": 0, "total": 0, "last_at": ""})
        for nk in totals:
            totals[nk] += int(rec[nk])
        for tk in last:
            last[tk] = _max_iso(last[tk], rec[tk])
        tx_totals["done"] += int(tx.get("done", 0))
        tx_totals["total"] += int(tx.get("total", 0))
        tx_totals["last_at"] = _max_iso(tx_totals["last_at"], tx.get("last_at") or "")
        states.append(st)
        rows.extend(_stage_rows(st, rec, discover_by_state.get(st), tx))

    disc_all: Dict[str, Any] = {"total": 0, "done": 0, "last_at": "", "breakdown": {}}
    for st in states:
        d = discover_by_state.get(st)
        if d:
            disc_all["total"] += d["total"]
            disc_all["done"] += d["done"]
            disc_all["last_at"] = _max_iso(disc_all["last_at"], d["last_at"])
            for e, b in (d.get("breakdown") or {}).items():
                agg = disc_all["breakdown"].setdefault(
                    e, {"entity": e, "total": 0, "done": 0}
                )
                agg["total"] += int(b["total"])
                agg["done"] += int(b["done"])
    # Stateless candidates (CivicSearch school meetings etc. carry no state_code)
    # have no per-state row; fold their coverage into the ALL transcripts numbers
    # so the headline reflects the whole candidate set.
    tx_totals["total"] += tx_unknown_total
    tx_totals["done"] += tx_unknown_done
    rows = _stage_rows("ALL", {**totals, **last}, disc_all, tx_totals) + rows
    out["states"] = sorted(states)
    out["rows"] = rows
    out["timing"] = _stage_timing(conn)
    return out


def dashboard_stage_report() -> Dict[str, Any]:
    """Open a connection and return the per-state pipeline report (empty on error)."""
    try:
        with get_db_connection() as conn:
            ensure_batch_job_tables(conn)
            return pipeline_stage_report(conn)
    except Exception as exc:
        logger.debug("dashboard_stage_report unavailable: %s", exc)
        return {"states": [], "rows": []}


def aggregate_dashboard_totals_from_db(*, limit: int = 30) -> Dict[str, Any]:
    """Sum numeric fields from ``summary`` JSONB across recent batches (no payload)."""
    # Reap abandoned ``running`` rows first so the ``running`` count below (and any
    # per-batch meta read for the same dashboard tick) reflects real liveness.
    reap_stale_running_batches(limit=limit)
    with get_db_connection() as conn:
        ensure_batch_job_tables(conn)
        with conn.cursor(cursor_factory=RealDictCursor) as cur:
            cur.execute(
                """
                WITH recent AS (
                    SELECT status, config, summary, updated_at
                    FROM bronze.youtube_batch_job_runs
                    ORDER BY updated_at DESC NULLS LAST
                    LIMIT %s
                )
                SELECT
                    COUNT(*)::int AS batches,
                    COUNT(*) FILTER (WHERE status = 'running')::int AS running,
                    COALESCE(SUM((summary->>'processed_jurisdictions')::int), 0)
                        AS processed_jurisdictions,
                    COALESCE(SUM((summary->>'failed_jurisdictions')::int), 0)
                        AS failed_jurisdictions,
                    COALESCE(SUM((summary->>'remaining_jurisdictions')::int), 0)
                        AS remaining_jurisdictions,
                    COALESCE(SUM((summary->>'videos_ok')::int), 0) AS videos_ok,
                    COALESCE(SUM((summary->>'videos_fail')::int), 0) AS videos_fail,
                    COALESCE(SUM((summary->>'files_transcripts')::int), 0)
                        AS files_transcripts,
                    COALESCE(SUM((summary->>'files_transcripts_disk')::int), 0)
                        AS files_transcripts_disk,
                    COALESCE(SUM((summary->>'bronze_download_rows')::int), 0)
                        AS bronze_download_rows,
                    COALESCE(SUM((summary->>'files_analysis')::int), 0) AS files_analysis,
                    COALESCE(SUM((summary->>'files_reports')::int), 0) AS files_reports,
                    COALESCE(SUM((summary->>'transcript_seconds')::float), 0)
                        AS transcript_seconds,
                    MAX(updated_at) AS last_updated
                FROM recent
                """,
                (limit,),
            )
            row = cur.fetchone() or {}
            recent_events = policy_event_counts_24h(conn)
    totals = {
        "batches": int(row.get("batches") or 0),
        "running": int(row.get("running") or 0),
        "states": 0,
        "states_planned": 0,
        "states_started": 0,
        "states_completed": 0,
        "processed_jurisdictions": int(row.get("processed_jurisdictions") or 0),
        "failed_jurisdictions": int(row.get("failed_jurisdictions") or 0),
        "remaining_jurisdictions": int(row.get("remaining_jurisdictions") or 0),
        "videos_ok": int(row.get("videos_ok") or 0),
        "videos_fail": int(row.get("videos_fail") or 0),
        "videos_attempted": 0,
        "files_transcripts": int(row.get("files_transcripts") or 0),
        "files_transcripts_disk": int(row.get("files_transcripts_disk") or 0),
        "transcript_hours": round(float(row.get("transcript_seconds") or 0) / 3600.0, 2),
        "bronze_download_rows": int(row.get("bronze_download_rows") or 0),
        "files_analysis": int(row.get("files_analysis") or 0),
        "files_reports": int(row.get("files_reports") or 0),
        # Rolling 24h throughput from per-event bronze stamps (migration 083); covers
        # both batch and standalone runs, unlike the batch-scoped disk-scan counters.
        "files_analysis_recent": int(recent_events.get("analysis") or 0),
        "files_reports_recent": int(recent_events.get("reports") or 0),
        "files_analysis_errors_recent": int(recent_events.get("analysis_errors") or 0),
        "files_reports_errors_recent": int(recent_events.get("reports_errors") or 0),
        # Most recent stamp per step (all time) for the dashboard "ago" cards.
        "last_transcript_at": recent_events.get("last_transcript_at") or "",
        "last_analysis_at": recent_events.get("last_analysis_at") or "",
        "last_report_at": recent_events.get("last_report_at") or "",
    }
    # Prefer the live, de-duplicated bronze-stamp totals for the pipeline cards and
    # progress %: summing per-batch disk scans inflates transcripts (overlapping
    # batches) and goes stale for analysis/reports. Only override when the stamps
    # have data, so a pre-083 DB still shows the batch-summary fallback.
    transcripts_total = int(recent_events.get("transcripts_total") or 0)
    analysis_total = int(recent_events.get("analysis_total") or 0)
    reports_total = int(recent_events.get("reports_total") or 0)
    if transcripts_total > 0:
        totals["files_transcripts_disk"] = transcripts_total
    if analysis_total > 0:
        totals["files_analysis"] = analysis_total
    if reports_total > 0:
        totals["files_reports"] = reports_total
    last_updated = row.get("last_updated")
    totals["last_activity_at"] = (
        last_updated.isoformat() if last_updated is not None else ""
    )
    return totals


def list_jurisdiction_rows_from_db(
    batch_id: str,
    state_code: str,
) -> List[Dict[str, Any]]:
    """
    Extract slim jurisdiction rows for one state via JSONB (no per-video arrays).
    """
    st = (state_code or "").strip().upper()
    if not st:
        return []
    with get_db_connection() as conn:
        ensure_batch_job_tables(conn)
        with conn.cursor(cursor_factory=RealDictCursor) as cur:
            cur.execute(
                """
                SELECT COALESCE(
                    jsonb_agg(sub.row ORDER BY sub.sort_name),
                    '[]'::jsonb
                ) AS jurisdictions
                FROM (
                    SELECT
                        jsonb_build_object(
                            'state_code', elem->>'state_code',
                            'jurisdiction_id', elem->>'jurisdiction_id',
                            'jurisdiction_name', elem->>'jurisdiction_name',
                            'status', COALESCE(NULLIF(elem->>'status', ''), 'pending'),
                            'started_at', COALESCE(elem->>'started_at', ''),
                            'updated_at', COALESCE(elem->>'updated_at', ''),
                            'finished_at', COALESCE(elem->>'finished_at', ''),
                            'elapsed_seconds',
                                COALESCE((elem->>'elapsed_seconds')::float, 0),
                            'exit_code', COALESCE((elem->>'exit_code')::int, 0),
                            'stats', COALESCE(elem->'stats', '{}'::jsonb),
                            'file_counts', COALESCE(elem->'file_counts', '{}'::jsonb),
                            'current_video_id', COALESCE(elem->>'current_video_id', ''),
                            'current_video_title', COALESCE(elem->>'current_video_title', ''),
                            'current_video_started_at',
                                COALESCE(elem->>'current_video_started_at', ''),
                            'videos', '[]'::jsonb
                        ) AS row,
                        lower(
                            COALESCE(
                                elem->>'jurisdiction_name',
                                elem->>'jurisdiction_id',
                                ''
                            )
                        ) AS sort_name
                    FROM bronze.youtube_batch_job_runs b,
                         jsonb_array_elements(
                             CASE
                                 WHEN jsonb_typeof(b.payload->'jurisdictions') = 'array'
                                 THEN b.payload->'jurisdictions'
                                 ELSE '[]'::jsonb
                             END
                         ) elem
                    WHERE b.batch_id = %s
                      AND UPPER(COALESCE(elem->>'state_code', '')) = %s
                ) sub
                """,
                (batch_id, st),
            )
            row = cur.fetchone()
    raw = row.get("jurisdictions") if row else []
    if isinstance(raw, str):
        raw = json.loads(raw)
    if not isinstance(raw, list):
        return []
    return [dict(x) for x in raw if isinstance(x, dict)]


def running_batch_activity_from_db() -> Optional[Dict[str, Any]]:
    """Active in-flight rows for the running batch (JSONB projection; no full payload parse)."""
    with get_db_connection() as conn:
        ensure_batch_job_tables(conn)
        with conn.cursor(cursor_factory=RealDictCursor) as cur:
            cur.execute(
                """
                SELECT
                    b.batch_id,
                    COALESCE(
                        jsonb_agg(sub.row ORDER BY sub.sort_key),
                        '[]'::jsonb
                    ) AS jurisdictions
                FROM bronze.youtube_batch_job_runs b
                LEFT JOIN LATERAL (
                    SELECT
                        jsonb_build_object(
                            'state_code', elem->>'state_code',
                            'jurisdiction_id', elem->>'jurisdiction_id',
                            'jurisdiction_name', elem->>'jurisdiction_name',
                            'status', COALESCE(elem->>'status', 'running'),
                            'updated_at', COALESCE(elem->>'updated_at', ''),
                            'current_video_id', COALESCE(elem->>'current_video_id', ''),
                            'current_video_title', COALESCE(elem->>'current_video_title', ''),
                            'current_video_started_at',
                                COALESCE(elem->>'current_video_started_at', ''),
                            'videos', '[]'::jsonb
                        ) AS row,
                        COALESCE(elem->>'current_video_started_at', elem->>'started_at', '')
                            AS sort_key
                    FROM jsonb_array_elements(
                        CASE
                            WHEN jsonb_typeof(b.payload->'jurisdictions') = 'array'
                            THEN b.payload->'jurisdictions'
                            ELSE '[]'::jsonb
                        END
                    ) elem
                    WHERE LOWER(COALESCE(elem->>'status', '')) = 'running'
                       OR COALESCE(elem->>'current_video_id', '') <> ''
                ) sub ON TRUE
                WHERE b.status = 'running'
                GROUP BY b.batch_id
                ORDER BY MAX(b.updated_at) DESC NULLS LAST
                LIMIT 1
                """
            )
            row = cur.fetchone()
    if not row:
        return None
    jurs = row.get("jurisdictions") or []
    if isinstance(jurs, str):
        jurs = json.loads(jurs)
    if not isinstance(jurs, list) or not jurs:
        return None
    return {"batch_id": row["batch_id"], "jurisdictions": [dict(x) for x in jurs if isinstance(x, dict)]}


_FAILED_VIDEO_STATUSES = frozenset(
    {"fail", "failed", "tombstoned", "empty", "rate_limit", "error"}
)


def list_failed_videos_from_db(
    *,
    batch_id: Optional[str] = None,
    limit: int = 500,
    batch_limit: int = 25,
) -> Dict[str, Any]:
    """
    Extract per-video failure rows from stored batch payloads (JSONB).

    Returns ``rows`` plus ``total_fail_in_summaries`` (sum of ``summary.videos_fail``)
    which may exceed ``len(rows)`` when failures were counted in stats but not logged
    per video, or when ``limit`` truncates.
    """
    bid = (batch_id or "").strip() or None
    lim = max(1, min(int(limit), 2000))
    with get_db_connection() as conn:
        ensure_batch_job_tables(conn)
        with conn.cursor(cursor_factory=RealDictCursor) as cur:
            if bid:
                cur.execute(
                    """
                    SELECT COALESCE((summary->>'videos_fail')::int, 0) AS n
                    FROM bronze.youtube_batch_job_runs
                    WHERE batch_id = %s
                    """,
                    (bid,),
                )
            else:
                cur.execute(
                    """
                    SELECT COALESCE(SUM((summary->>'videos_fail')::int), 0)::int AS n
                    FROM (
                        SELECT summary
                        FROM bronze.youtube_batch_job_runs
                        ORDER BY updated_at DESC NULLS LAST
                        LIMIT %s
                    ) recent
                    """,
                    (batch_limit,),
                )
            summary_row = cur.fetchone() or {}
            total_fail = int(summary_row.get("n") or 0)

            cur.execute(
                """
                SELECT
                    b.batch_id,
                    b.step AS batch_step,
                    j.elem->>'state_code' AS state_code,
                    j.elem->>'jurisdiction_id' AS jurisdiction_id,
                    j.elem->>'jurisdiction_name' AS jurisdiction_name,
                    v.elem->>'video_id' AS video_id,
                    v.elem->>'title' AS title,
                    v.elem->>'status' AS status,
                    v.elem->>'error' AS error,
                    v.elem->>'transcript_source' AS transcript_source,
                    v.elem->>'finished_at' AS finished_at,
                    v.elem->>'duration_seconds' AS duration_seconds
                FROM bronze.youtube_batch_job_runs b
                CROSS JOIN LATERAL jsonb_array_elements(
                    CASE
                        WHEN jsonb_typeof(b.payload->'jurisdictions') = 'array'
                        THEN b.payload->'jurisdictions'
                        ELSE '[]'::jsonb
                    END
                ) AS j(elem)
                CROSS JOIN LATERAL jsonb_array_elements(
                    CASE
                        WHEN jsonb_typeof(j.elem->'videos') = 'array'
                        THEN j.elem->'videos'
                        ELSE '[]'::jsonb
                    END
                ) AS v(elem)
                WHERE (%s::text IS NULL OR b.batch_id = %s)
                  AND COALESCE(v.elem->>'video_id', '') <> ''
                  AND (
                    LOWER(COALESCE(v.elem->>'status', '')) = ANY(%s)
                    OR (
                      LOWER(COALESCE(v.elem->>'status', '')) NOT IN (
                        'ok', 'pending', 'skipped', 'noop', ''
                      )
                    )
                  )
                ORDER BY b.updated_at DESC NULLS LAST,
                         v.elem->>'finished_at' DESC NULLS LAST
                LIMIT %s
                """,
                (
                    bid,
                    bid,
                    list(_FAILED_VIDEO_STATUSES),
                    lim + 1,
                ),
            )
            raw_rows = cur.fetchall()
    truncated = len(raw_rows) > lim
    rows_out: List[Dict[str, Any]] = []
    for row in raw_rows[:lim]:
        dur = row.get("duration_seconds")
        try:
            dur_f = float(dur) if dur not in (None, "") else None
        except (TypeError, ValueError):
            dur_f = None
        rows_out.append(
            {
                "batch_id": row["batch_id"],
                "batch_step": row.get("batch_step") or "",
                "state_code": row.get("state_code") or "",
                "jurisdiction_id": row.get("jurisdiction_id") or "",
                "jurisdiction_name": row.get("jurisdiction_name") or "",
                "video": {
                    "video_id": row.get("video_id") or "",
                    "title": row.get("title") or "",
                    "status": row.get("status") or "",
                    "error": row.get("error") or "",
                    "transcript_source": row.get("transcript_source") or "",
                    "finished_at": row.get("finished_at") or "",
                    "duration_seconds": dur_f,
                },
            }
        )
    return {
        "rows": rows_out,
        "total_fail_in_summaries": total_fail,
        "truncated": truncated,
    }


def list_batch_job_meta_from_db(*, limit: int = 30) -> List[Dict[str, Any]]:
    """Lightweight batch rows (no ``payload``) for fast dashboard summary."""
    with get_db_connection() as conn:
        ensure_batch_job_tables(conn)
        with conn.cursor(cursor_factory=RealDictCursor) as cur:
            cur.execute(
                """
                SELECT
                    batch_id,
                    step,
                    status,
                    started_at,
                    updated_at,
                    finished_at,
                    config,
                    summary
                FROM bronze.youtube_batch_job_runs
                ORDER BY updated_at DESC NULLS LAST
                LIMIT %s
                """,
                (limit,),
            )
            rows = cur.fetchall()
    out: List[Dict[str, Any]] = []
    for row in rows:
        started = row.get("started_at")
        updated = row.get("updated_at")
        finished = row.get("finished_at")
        out.append(
            {
                "batch_id": row["batch_id"],
                "step": row["step"],
                "status": row["status"],
                "started_at": started.isoformat() if started else "",
                "updated_at": updated.isoformat() if updated else "",
                "finished_at": finished.isoformat() if finished else "",
                "config": row.get("config") or {},
                "summary": row.get("summary") or {},
                "jurisdictions": [],
            }
        )
    return out


def sync_json_batches_to_db(*, limit: int = 100) -> int:
    """Import recent JSON batch files into Postgres (one-time / backfill)."""
    if not _use_db():
        return 0
    jobs = list_batches(limit=limit)
    if not jobs:
        return 0
    with get_db_connection() as conn:
        for job in jobs:
            upsert_batch_job(conn, job)
    return len(jobs)


def enrich_transcript_counts_from_bronze(conn: Any, job: BatchJob) -> None:
    """Set jurisdiction ``bronze_download_rows`` from bronze_event_youtube."""
    jids = [j.jurisdiction_id for j in job.jurisdictions if j.jurisdiction_id]
    if not jids:
        return
    cur = conn.cursor(cursor_factory=RealDictCursor)
    try:
        cur.execute(
            """
            SELECT
                jurisdiction_id,
                COUNT(*) FILTER (WHERE transcript_download_at IS NOT NULL)::int AS transcripts,
                COUNT(*) FILTER (
                    WHERE transcript_file_error IS NOT NULL
                      AND BTRIM(transcript_file_error) <> ''
                )::int AS transcript_errors
            FROM bronze.bronze_event_youtube
            WHERE jurisdiction_id = ANY(%s)
            GROUP BY jurisdiction_id
            """,
            (jids,),
        )
        by_jid = {r["jurisdiction_id"]: r for r in cur.fetchall()}
    finally:
        cur.close()

    for j in job.jurisdictions:
        row = by_jid.get(j.jurisdiction_id) or {}
        tx = int(row.get("transcripts") or 0)
        j.file_counts = dict(j.file_counts or {})
        j.file_counts["bronze_download_rows"] = tx
        j.file_counts["bronze_transcript_errors"] = int(
            row.get("transcript_errors") or 0
        )


def enrich_transcript_seconds_from_bronze(conn: Any, job: BatchJob) -> None:
    """
    Set ``job.summary['transcript_seconds']`` from ``bronze_event_youtube.duration_minutes``.

    Batch payloads often have per-video stats without ``duration_seconds`` (older runs or
    DB copies). Prefer catalog duration for ok video ids in ``j.videos``; if none, sum
    transcripts downloaded in processed jurisdictions since ``job.started_at``.
    """
    from api.batch_jobs.batch_job_status import (
        transcript_seconds_from_job_videos,
    )

    ok_ids = [
        v.video_id
        for j in job.jurisdictions
        for v in j.videos or []
        if (v.status or "").strip().lower() == "ok" and (v.video_id or "").strip()
    ]
    cur = conn.cursor()
    try:
        if ok_ids:
            cur.execute(
                """
                SELECT COALESCE(SUM(duration_minutes), 0) * 60.0
                FROM bronze.bronze_event_youtube
                WHERE video_id = ANY(%s)
                  AND duration_minutes IS NOT NULL
                  AND duration_minutes > 0
                """,
                (ok_ids,),
            )
            row = cur.fetchone()
            secs = float(row[0] or 0) if row else 0.0
        else:
            active_jids = [
                j.jurisdiction_id
                for j in job.jurisdictions
                if j.jurisdiction_id
                and j.status in ("completed", "failed", "running")
                and (
                    int((j.stats or {}).get("ok") or 0) > 0
                    or any(
                        (v.status or "").strip().lower() == "ok" for v in j.videos or []
                    )
                )
            ]
            if not active_jids:
                secs = transcript_seconds_from_job_videos(job)
            elif job.started_at:
                cur.execute(
                    """
                    SELECT COALESCE(SUM(duration_minutes), 0) * 60.0
                    FROM bronze.bronze_event_youtube
                    WHERE jurisdiction_id = ANY(%s)
                      AND transcript_download_at IS NOT NULL
                      AND transcript_download_at >= %s::timestamptz
                      AND duration_minutes IS NOT NULL
                      AND duration_minutes > 0
                    """,
                    (active_jids, job.started_at),
                )
                row = cur.fetchone()
                secs = float(row[0] or 0) if row else 0.0
            else:
                cur.execute(
                    """
                    SELECT COALESCE(SUM(duration_minutes), 0) * 60.0
                    FROM bronze.bronze_event_youtube
                    WHERE jurisdiction_id = ANY(%s)
                      AND transcript_download_at IS NOT NULL
                      AND duration_minutes IS NOT NULL
                      AND duration_minutes > 0
                    """,
                    (active_jids,),
                )
                row = cur.fetchone()
                secs = float(row[0] or 0) if row else 0.0
    finally:
        cur.close()

    job.summary = dict(job.summary or {})
    job.summary["transcript_seconds"] = round(secs, 1)


def enrich_disk_file_counts(job: BatchJob, *, cache_root: Path | None = None) -> None:
    """Merge on-disk policy cache file counts into each jurisdiction's ``file_counts``."""
    from api.batch_jobs.batch_job_status import (
        count_policy_files_for_jurisdiction,
        policy_disk_file_counts,
    )

    root = cache_root or (
        Path(__file__).resolve().parents[3] / "data" / "cache" / "gemini_transcript_policy"
    )
    for j in job.jurisdictions:
        if not j.jurisdiction_id:
            continue
        scanned = count_policy_files_for_jurisdiction(
            root,
            state_code=j.state_code,
            jurisdiction_id=j.jurisdiction_id,
        )
        j.file_counts = dict(j.file_counts or {})
        j.file_counts.update(policy_disk_file_counts(scanned))


def enrich_jobs_from_bronze(
    jobs: List[BatchJob],
    *,
    only_running_jurisdictions: bool = False,
    enrich_disk: bool = True,
) -> None:
    if not jobs:
        return
    try:
        with get_db_connection() as conn:
            for job in jobs:
                if only_running_jurisdictions:
                    pending = [
                        j
                        for j in job.jurisdictions
                        if j.status in ("running", "pending")
                    ]
                    if not pending:
                        continue
                    stub = BatchJob(
                        batch_id=job.batch_id,
                        step=job.step,
                        jurisdictions=pending,
                    )
                    enrich_transcript_counts_from_bronze(conn, stub)
                    enrich_transcript_seconds_from_bronze(conn, stub)
                    if enrich_disk:
                        enrich_disk_file_counts(stub)
                    by_jid = {j.jurisdiction_id: j for j in pending}
                    for j in job.jurisdictions:
                        if j.jurisdiction_id in by_jid:
                            j.file_counts = dict(j.file_counts or {})
                            j.file_counts.update(by_jid[j.jurisdiction_id].file_counts)
                else:
                    enrich_transcript_counts_from_bronze(conn, job)
                    enrich_transcript_seconds_from_bronze(conn, job)
                    if enrich_disk:
                        enrich_disk_file_counts(job)
    except Exception as exc:
        logger.warning("bronze transcript count enrich failed: %s", exc)


def load_batch_job_from_db(batch_id: str) -> Optional[BatchJob]:
    with get_db_connection() as conn:
        ensure_batch_job_tables(conn)
        with conn.cursor(cursor_factory=RealDictCursor) as cur:
            cur.execute(
                """
                SELECT payload FROM bronze.youtube_batch_job_runs
                WHERE batch_id = %s
                """,
                (batch_id,),
            )
            row = cur.fetchone()
    if not row:
        return None
    payload = row["payload"]
    if isinstance(payload, str):
        payload = json.loads(payload)
    return BatchJob.from_dict(payload)


def latest_dashboard_revision() -> Optional[str]:
    """Cheap change detector for SSE (max updated_at + running count)."""
    try:
        with get_db_connection() as conn:
            ensure_batch_job_tables(conn)
            with conn.cursor() as cur:
                cur.execute(
                    """
                    SELECT
                        COALESCE(MAX(updated_at)::text, ''),
                        COUNT(*) FILTER (WHERE status = 'running')::int
                    FROM bronze.youtube_batch_job_runs
                    """
                )
                updated_at, running = cur.fetchone()
        return f"{updated_at}|{running}"
    except Exception:
        return None