feat(db): backfill, DB-backed web scans, firewall snapshot view (R001/S001/F001)

- R001: app/db/backfill.py imports git-history blocklist -> ledger, firewall
  snapshots, and per-source requests.csv (idempotent; ran live migrating 705
  historical blocklist entries).
- S001: web_scan.latest_scans() now merges SQLite 'scans' records so the
  dashboard surfaces CLI/restart scans (_db_scans).
- F001: /api/firewall/snapshots + 'Snapshot history' panel; added
  recent_firewall_snapshots() query with entry counts.

app/db/backfill.py

@@ -0,0 +1,196 @@
+#!/usr/bin/env python3
+"""R001 — one-time backfill of legacy runtime data into the SQLite store.
+
+Imports existing JSON/CSV/git-history data so the DB becomes the authoritative
+source without losing anything already accumulated on a deployed instance:
+
+- cumulative blocklist from the git-history reconstruction (old source of
+  truth) -> ``blocklist_entries`` (grow-only ledger)
+- firewall snapshots from ``data/mikrotik/*.json`` -> ``firewall_snapshots``
+- per-source requests from ``reports/<source>/data/requests.csv`` (best effort)
+
+Idempotent: skips rows already present. Safe to run repeatedly.
+
+Usage:
+    python -m app.db.backfill [--dry-run]
+"""
+
+from __future__ import annotations
+
+import argparse
+import csv
+import json
+import os
+import sys
+from datetime import datetime, timezone
+
+from ..utils.constants import (
+    APP_REPORT_DATA_DIR,
+    DATA_DIR,
+    MIKROTIK_DATA_DIR,
+    NPM_REPORT_DIR,
+    COPYPARTY_REPORT_DIR,
+)
+
+
+def utcnow() -> str:
+    return datetime.now(timezone.utc).isoformat(timespec="seconds")
+
+
+def _read_cidr_lines(path: str) -> set[str]:
+    try:
+        with open(path, "r", encoding="utf-8") as f:
+            return {line.strip() for line in f if line.strip()}
+    except OSError:
+        return set()
+
+
+def backfill_blocklist(conn, dry_run: bool = False) -> dict:
+    """Import the legacy cumulative blocklist into the ledger."""
+    from . import repositories as repo
+    from ..utils.blocklist_history import (
+        committed_lines, current_file_lines,
+    )
+
+    v4_path = os.path.join(APP_REPORT_DATA_DIR, "blocklist_ipv4.txt")
+    v6_path = os.path.join(APP_REPORT_DATA_DIR, "blocklist_ipv6.txt")
+    # Legacy source of truth: union of all git revisions + working file +
+    # (there is no earlier scan target to add here).
+    complete_v4 = committed_lines(v4_path) | current_file_lines(v4_path)
+    complete_v6 = committed_lines(v6_path) | current_file_lines(v6_path)
+    cidrs = sorted(complete_v4 | complete_v6)
+    if not cidrs:
+        return {"blocklist": {"cidrs": 0, "added": 0}}
+    if dry_run:
+        return {"blocklist": {"cidrs": len(cidrs), "added": None, "dry_run": True}}
+    result = repo.record_blocklist(conn, cidrs, added_by="backfill")
+    return {"blocklist": result}
+
+
+def backfill_firewall(conn, dry_run: bool = False) -> dict:
+    """Import the latest firewall snapshot from ``data/mikrotik/*.json``."""
+    from . import repositories as repo
+
+    def _load(name: str) -> list:
+        try:
+            with open(os.path.join(MIKROTIK_DATA_DIR, name), "r",
+                      encoding="utf-8") as f:
+                data = json.load(f)
+            return data if isinstance(data, list) else []
+        except (OSError, ValueError):
+            return []
+
+    addr_v4 = _load("address_list_ipv4.json")
+    addr_v6 = _load("address_list_ipv6.json")
+    fw_v4 = _load("firewall_ipv4.json")
+    fw_v6 = _load("firewall_ipv6.json")
+    if not (addr_v4 or addr_v6 or fw_v4 or fw_v6):
+        return {"firewall": {"dry_run": dry_run, "snapshot": None}}
+    # Idempotent: if a firewall snapshot already exists (previous backfill),
+    # don't create another copy of the same latest state.
+    n = conn.execute(
+        "SELECT COUNT(*) FROM firewall_snapshots").fetchone()[0]
+    if n:
+        return {"firewall": {"dry_run": dry_run, "snapshot": "exists",
+                              "skipped": True}}
+    if dry_run:
+        return {"firewall": {"dry_run": True, "snapshot": None}}
+    sid = repo.save_firewall_snapshot(
+        conn, address_lists_v4=addr_v4, address_lists_v6=addr_v6,
+        rules_v4=fw_v4, rules_v6=fw_v6)
+    return {"firewall": {"snapshot": sid}}
+
+
+def backfill_requests(conn, dry_run: bool = False) -> dict:
+    """Best-effort import of per-source ``requests.csv`` files.
+
+    Only ingests rows that don't already exist (keyed on source+log_file+ts+ip).
+    Legacy CSV columns vary by source; we keep a conservative mapping and never
+    fail the whole backfill over a single bad file.
+    """
+    from . import repositories as repo
+
+    seen = set()
+    for r in conn.execute(
+            "SELECT source, log_file, ts, ip FROM requests"):
+        seen.add((r["source"], r["log_file"], r["ts"], r["ip"]))
+
+    candidates = [
+        (os.path.join(COPYPARTY_REPORT_DIR, "data", "requests.csv"), "copyparty"),
+        (os.path.join(NPM_REPORT_DIR, "data", "requests.csv"), "npm"),
+    ]
+    total = 0
+    for path, source in candidates:
+        if not os.path.exists(path):
+            continue
+        try:
+            with open(path, "r", encoding="utf-8", errors="replace") as f:
+                reader = csv.DictReader(f)
+                rows = []
+                for row in reader:
+                    ip = (row.get("ip") or "").strip()
+                    ts = row.get("ts") or row.get("timestamp") or ""
+                    log_file = row.get("log_file") or row.get("file") or ""
+                    if not ip:
+                        continue
+                    key = (source, log_file, ts, ip)
+                    if key in seen:
+                        continue
+                    seen.add(key)
+                    rows.append({
+                        "source": source, "scan_id": None, "log_file": log_file,
+                        "ts": ts, "ip": ip,
+                        "version": 4 if ":" not in ip else 6,
+                        "method": row.get("method"), "path": row.get("path"),
+                        "status": _int_or_none(row.get("status")),
+                        "upstream_status": _int_or_none(
+                            row.get("upstream_status")),
+                        "host": row.get("host"), "user": row.get("user"),
+                        "ua": row.get("ua") or row.get("user_agent"),
+                        "referer": row.get("referer"),
+                        "denied": _bool_str(row.get("denied")),
+                        "banned": _bool_str(row.get("banned")),
+                        "is_error": _bool_str(row.get("is_error")),
+                        "extra": None,
+                    })
+                    total += 1
+            if not dry_run and total:
+                repo.insert_requests(conn, rows)
+        except Exception as exc:
+            print(f"  WARN: requests.csv backfill {source}: {exc}")
+    return {"requests": {"imported": total, "dry_run": dry_run}}
+
+
+def _int_or_none(v):
+    try:
+        if v in (None, "", "-"):
+            return None
+        return int(v)
+    except (TypeError, ValueError):
+        return None
+
+
+def _bool_str(v):
+    if isinstance(v, str):
+        return int(v.strip().lower() in ("1", "true", "yes", "t"))
+    return int(bool(v))
+
+
+def main(argv: list[str] | None = None) -> None:
+    p = argparse.ArgumentParser(prog="python -m app.db.backfill",
+                                description="Backfill legacy runtime data into SQLite.")
+    p.add_argument("--dry-run", action="store_true",
+                   help="report what would be imported without writing")
+    opts = p.parse_args(argv)
+
+    from . import get_connection
+    conn = get_connection()
+    print(f"== backfill (dry_run={opts.dry_run}) ==")
+    print("blocklist:", backfill_blocklist(conn, dry_run=opts.dry_run))
+    print("firewall :", backfill_firewall(conn, dry_run=opts.dry_run))
+    print("requests :", backfill_requests(conn, dry_run=opts.dry_run))
+    print("== done ==")
+
+
+if __name__ == "__main__":
+    sys.exit(main())

app/templates/firewall.html

@@ -23,6 +23,11 @@ endblock %} {% block content %}
   <div class="panel-head"><h2>Firewall rules</h2></div>
   <div id="firewall-rules"></div>
 </div>
+
+<div class="panel">
+  <div class="panel-head"><h2>Snapshot history (SQLite)</h2></div>
+  <div id="fw-snapshots"></div>
+</div>
 {% endblock %} {% block scripts %}
 <script>
   async function loadState() {
@@ -35,6 +40,45 @@ endblock %} {% block content %}
     }
   }
 
+  async function loadSnapshots() {
+    const box = document.getElementById("fw-snapshots");
+    try {
+      const d = await api("/api/firewall/snapshots?limit=10");
+      const snaps = d.snapshots || [];
+      if (!snaps.length) {
+        box.innerHTML =
+          '<p class="muted">No SQLite snapshots yet — they are recorded on each <code>--update-firewall</code> run.</p>';
+        return;
+      }
+      box.innerHTML = "";
+      const rows = snaps.map((s) =>
+        el(
+          "tr",
+          el("td", { class: "mono", text: `#${s.id}` }),
+          el("td", { text: fmtTime(s.captured_at) }),
+          el("td", { text: s.kind }),
+          el("td", { text: s.trigger }),
+          el("td", { class: "num", text: s.addr_entries }),
+          el("td", { class: "num", text: s.rule_entries }),
+        ),
+      );
+      box.append(
+        el(
+          "table",
+          { class: "table" },
+          el(
+            "thead",
+            el("tr", {}, el("th", { text: "ID" }), el("th", { text: "Captured" }), el("th", { text: "Kind" }), el("th", { text: "Trigger" }), el("th", { text: "Addr" }), el("th", { text: "Rules" })),
+          ),
+          el("tbody", {}, ...rows),
+        ),
+      );
+    } catch (e) {
+      box.innerHTML=
+        '<p class="muted">Snapshot history unavailable (SQLite).</p>';
+    }
+  }
+
   function renderAddressLists(addr) {
     const box = document.getElementById("address-lists");
     box.innerHTML = "";
@@ -172,5 +216,6 @@ endblock %} {% block content %}
     .getElementById("apply-btn")
     .addEventListener("click", () => sync(false));
   loadState();
+  loadSnapshots();
 </script>
 {% endblock %}

app/web/routes.py

@@ -265,6 +265,19 @@ def register_routes(app: Flask,
             "snapshot_dir": MIKROTIK_DATA_DIR,
         })
 
+    @app.get("/api/firewall/snapshots")
+    def firewall_snapshots():
+        """Recent firewall snapshots persisted in the SQLite store (F001)."""
+        try:
+            from ..db import get_connection
+            from ..db import repositories as repo
+            conn = get_connection()
+            limit = request.args.get("limit", 10, type=int)
+            return jsonify({"snapshots": repo.recent_firewall_snapshots(
+                conn, limit=limit)})
+        except Exception as exc:
+            return jsonify({"error": str(exc), "snapshots": []}), 503
+
     @app.post("/api/firewall/sync")
     def firewall_sync():
         return _sync_firewall(dry_run=True)

app/web/web_scan.py

@@ -258,23 +258,71 @@ def latest_file_scan() -> dict[str, Any] | None:
     }
 
 
+def _db_scans(limit: int = 10) -> list[dict[str, Any]]:
+    """Recent scan records from the SQLite ``scans`` table (S001).
+
+    Converts DB rows into the web job-dict shape so the dashboard can surface
+    scans that were persisted by ``run_scan`` / analyzers even if they were
+    never run through the in-memory job store (e.g. CLI scans or restarts).
+    Returns [] when the DB is unavailable.
+    """
+    try:
+        from ..db import get_connection
+        from ..db import repositories as repo
+        conn = get_connection()
+        rows = repo.latest_scans(conn, limit=limit)
+    except Exception:
+        return []
+    out = []
+    for r in rows:
+        try:
+            result_json = r.get("result_json")
+            result = json.loads(result_json) if result_json else {}
+        except (TypeError, ValueError):
+            result = {}
+        out.append({
+            "job_id": f"db-scan-{r['id']}",
+            "kind": r.get("kind") or "consolidated",
+            "state": r.get("status") or "done",
+            "started": r.get("started_at"),
+            "finished": r.get("finished_at"),
+            "error": r.get("error"),
+            "busy": False,
+            "details": {},
+            "result": result or None,
+        })
+    return out
+
+
 def latest_scans(limit: int = 10) -> list[dict[str, Any]]:
     """Return the most recently started jobs, newest first.
 
-    Appends a file-derived "latest scan" entry when the newest report files on
+    Merges three sources, newest-first:
-    disk reflect a scan newer than any known job (e.g. after a restart or a
+      1. the in-memory job store (currently running / last-run jobs),
-    CLI run), so the dashboard always shows the latest completed scan.
+      2. the SQLite ``scans`` table (persisted by run_scan/analyzers — S001),
+      3. a file-derived "latest scan" fallback for pre-DB runs.
+    Any DB-backed scan not already represented in memory is appended so the
+    dashboard shows scans from CLI runs and across restarts.
     """
     load_persisted()
     with _jobs_lock:
         jobs = sorted(_jobs.values(),
                       key=lambda j: j.started or "", reverse=True)
     entries = [j.to_dict() for j in jobs]
+
+    # Add DB scan records that aren't already in the in-memory store. Key on
+    # (kind, started_at) to avoid duplicating a job that run_scan also logged.
+    known = {(e.get("kind"), e.get("started")) for e in entries}
+    for db_row in _db_scans(limit=limit * 2):
+        if (db_row.get("kind"), db_row.get("started")) not in known:
+            entries.append(db_row)
+
     file_entry = latest_file_scan()
     if file_entry is not None:
-        newest = jobs[0] if jobs else None
+        newest = (entries or [None])[0]
-        newest_ts = _iso_key(
+        newest_key = (
-            (newest.finished or newest.started) if newest else None)
+            (newest.get("finished") or newest.get("started")) if newest else None)
+        newest_ts = _iso_key(newest_key)
         # Only surface the file-derived entry when it is genuinely newer than
         # every known job (tolerance avoids duplicating the run that produced
         # it, where collected_at lands a moment after the job finished).