feat(db): chronicle and ledger on native Postgres — #251

backend/benchmarks/ledger_promotion.py

@@ -0,0 +1,148 @@
+"""Times a ledger day promotion and `/ledger/series` on Postgres at the factions cap.
+
+    DATABASE_URL=postgresql://… python benchmarks/ledger_promotion.py [--days 90] [--factions 2000]
+
+Point it at a scratch database: it migrates it and writes ledger rows for the
+map `bench`. Every day carries `--factions` factions (the cap is 2,000) and 200
+guilds. The first `--promote` days go through `ingest.promote_day` (raw file
+read, daily file write, index), the rest through `ingest.index_snapshot` only,
+to build history quickly. Then it times `/ledger/series` the way the viewer
+asks for it: the default (top 12 by wealth, core), 40 chosen factions with
+`fields=full`, and the index.
+"""
+from __future__ import annotations
+
+import argparse
+import os
+import random
+import statistics
+import sys
+import tempfile
+import time
+from datetime import date, timedelta
+from pathlib import Path
+
+BACKEND = Path(__file__).resolve().parents[1]
+sys.path[:0] = [str(BACKEND), str(BACKEND / "src")]
+os.environ.setdefault("SKINS_DEV", "1")
+
+MAP = "bench"
+
+
+def _faction(index: int, day: int, rng: random.Random) -> dict:
+    wealth = 1000.0 + index * 10 + day + rng.random()
+    return {
+        "id": f"nation{index:04d}", "founded_at": f"2026-01-01T00:00:{index % 60:02d}.{index:06d}Z",
+        "name": f"Nation {index}", "rgb": "#%06x" % (index * 2654435761 % 0xFFFFFF),
+        "overlord": f"nation{index - 1:04d}" if index % 7 == 0 and index else None,
+        "subjects": [f"nation{index + 1:04d}"] if index % 7 == 6 else [],
+        "wealth": wealth, "bank": wealth * 0.1, "vassal_wealth": 0.0, "net_income": rng.uniform(-50, 50),
+        "inflation_delta": rng.uniform(-1, 1), "trade_power": rng.uniform(0, 20), "prestige": wealth / 2,
+        "wealth_breakdown": {"Bank": wealth * 0.1, "Nodes": wealth * 0.5, f"guild{index:04d}": wealth * 0.4},
+        "prestige_breakdown": {"wealth": wealth / 4, "war": wealth / 4},
+        "rank": "Kingdom", "rank_level": 3, "rank_up_at": 600.0, "rank_down_at": 400.0,
+        "prestige_position": index + 1, "wealth_position": index + 1, "provinces": 12, "realm_size": 12,
+        "tier": "king", "tier_index": 4, "highest_title": f"King of {index}", "members": 5,
+        "members_with_vassals": 5, "settlements": 3, "population": 340, "installations": 2, "forts": 1,
+        "wars": [{"enemy": f"nation{(index + 3) % 2000:04d}", "since": "2026-01-01"}] if index % 11 == 0 else [],
+    }
+
+
+def _payload(day: date, factions: int, rng: random.Random) -> dict:
+    fs = [_faction(i, day.toordinal() % 1000, rng) for i in range(factions)]
+    guilds = [
+        {"id": f"guild{i:04d}", "faction_id": f"nation{i:04d}", "name": f"Guild {i}", "type": "CRAFT",
+         "wealth": 40.0, "bank": 10.0, "expansions": 2, "trade_power": 1.5, "credit_score": 0.8, "size": 6}
+        for i in range(200)
+    ]
+    return {
+        "schema_version": 1, "map_id": MAP, "captured_at": f"{day.isoformat()}T12:00:00.123456789Z",
+        "server_day": day.toordinal() % 1000, "day_progress_seconds": 900, "complete": True,
+        "global": {"faction_count": factions, "guild_count": len(guilds), "claimed_provinces": 12000,
+                   "population": 340000, "active_wars": 3, "max_wealth_prestige": 900.0,
+                   "faction_wealth": 1e7, "pouch_wealth": 5e5, "player_bank_wealth": 2e5, "liquid_wealth": 7e5,
+                   "guild_liquid_wealth": 1e4, "node_wealth": 5e3, "expansion_wealth": 2e3, "guild_income": 3.0},
+        "factions": fs, "guilds": guilds, "events": [],
+    }
+
+
+def _timed(work, repeat: int) -> list[float]:
+    times = []
+    for _ in range(repeat):
+        start = time.perf_counter()
+        work()
+        times.append((time.perf_counter() - start) * 1000)
+    return times
+
+
+def _report(label: str, times: list[float]) -> None:
+    print(f"{label:<44} median {statistics.median(times):8.1f} ms   max {max(times):8.1f} ms   (n={len(times)})")
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
+    parser.add_argument("--days", type=int, default=90)
+    parser.add_argument("--factions", type=int, default=2000)
+    parser.add_argument("--promote", type=int, default=5, help="days timed through promote_day")
+    parser.add_argument("--repeat", type=int, default=7)
+    args = parser.parse_args()
+
+    from fastapi import FastAPI
+    from fastapi.testclient import TestClient
+
+    from src.api import map_registry
+    from src.database.migrate import migrate
+    from src.scripts.ledger import ingest, store
+    from src.scripts.ledger.schema import normalize_snapshot
+
+    tmp = Path(tempfile.mkdtemp(prefix="ledger-bench-"))
+    (tmp / "maps.yml").write_text(
+        f"maps:\n  - id: {MAP}\n    public: true\n    display_name: Bench\n    realm_id: {MAP}\n", encoding="utf-8"
+    )
+    os.environ["MAP_REGISTRY_PATH"] = str(tmp / "maps.yml")
+    map_registry.clear_map_registry_cache()
+    store.OUTPUT_DIR = str(tmp / "output")
+    migrate()
+
+    rng = random.Random(7)
+    first = date.today() - timedelta(days=args.days)
+    promote, index = [], []
+    for offset in range(args.days):
+        day = first + timedelta(days=offset)
+        start = time.perf_counter()
+        snapshot = normalize_snapshot(_payload(day, args.factions, rng), MAP)
+        normalised = (time.perf_counter() - start) * 1000
+        if offset < args.promote:
+            ingest.store_raw(MAP, snapshot)
+            promote += _timed(lambda: ingest.promote_day(MAP, snapshot["day"]), 1)
+        else:
+            index += _timed(lambda: ingest.index_snapshot(MAP, snapshot), 1)
+    print(f"{args.days} days x {args.factions} factions + 200 guilds; normalise one payload: {normalised:.0f} ms")
+    _report("promote_day (raw read, daily write, index)", promote)
+    if index:
+        _report("index_snapshot (index only)", index)
+    _report("re-promote one day (replace rows)", _timed(lambda: ingest.promote_day(MAP, first.isoformat()), 3))
+
+    app = FastAPI()
+    from src.api.ledger_routes import ledger_router
+
+    app.include_router(ledger_router)
+    keys = [row["faction_key"] for row in store.list_registry(MAP)][:40]
+    with TestClient(app) as client:
+        def get(url: str):
+            response = client.get(url)
+            assert response.status_code == 200, response.text
+            return response
+
+        for label, url in (
+            ("/ledger/series (default: top 12, core)", f"/{MAP}/ledger/series"),
+            ("/ledger/series (40 factions, full)", f"/{MAP}/ledger/series?fields=full&factions={','.join(keys)}"),
+            ("/ledger/index", f"/{MAP}/ledger/index"),
+        ):
+            get(url)
+            _report(label, _timed(lambda: get(url), args.repeat))
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

backend/src/api/chronicle_routes.py

@@ -25,6 +25,7 @@
     get_snapshot,
     is_valid_day,
     list_day_rows,
+    read_view,
     resolve_stored_file,
 )
 from ..scripts.loader.markers import (
@@ -165,7 +166,7 @@ def _incomplete_days(day_manifests: list[tuple[str, dict]]) -> list[dict]:
 
     Takes the rows the caller already fetched rather than looking each day up
     again: the manifest lives in the index row, and a get_snapshot per day was a
-    fresh SQLite connect per day of history on every index request.
+    database connection per day of history on every index request.
     """
     out: list[dict] = []
     for day, manifest in day_manifests:
@@ -225,8 +226,8 @@ def get_chronicle_index(
     cached, since a stale copy silently hides newly captured days. The ETag
     still turns an unchanged index into a bodiless 304.
 
-    Declared `def`, not `async def`: every line below is blocking - SQLite and a
-    file stat - and an `async def` handler runs on the event loop itself, so one
+    Declared `def`, not `async def`: every line below is blocking - database
+    reads and a file stat - and an `async def` handler runs on the event loop itself, so one
     slow index request stalls every other connection on the box. Starlette runs
     a sync handler in its threadpool instead.
     """
@@ -240,7 +241,11 @@ def get_chronicle_index(
 
     # One query for both fields: `days` is the row order this returns, so the
     # response ordering is unchanged from the old list_days + per-day lookup.
-    day_rows = list_day_rows(map_id)
+    # One snapshot for the rows and the wipe stamp, so a wipe committing in
+    # between cannot report days that are gone alongside the wipe that took them.
+    with read_view() as conn:
+        day_rows = list_day_rows(map_id, conn=conn)
+        wiped_at = last_wiped_at(map_id, conn=conn)
     days = [day for day, _manifest, _geometry in day_rows]
     incomplete = _incomplete_days([(day, manifest) for day, manifest, _ in day_rows])
     live_geometry_version = _cached_geometry_version(map_id)
@@ -266,7 +271,7 @@ def get_chronicle_index(
             # and the viewer told users the first thing in both cases. This is
             # the unix stamp of the newest staff wipe of this map (null when it
             # has never been wiped), so the two can be told apart.
-            "last_wiped_at": last_wiped_at(map_id),
+            "last_wiped_at": wiped_at,
         },
         if_none_match=if_none_match,
     )
@@ -394,7 +399,7 @@ def get_chronicle_markers(
     That is accepted — snapshotting images is out of scope.
 
     Sync body, so `def` rather than `async def` - see get_chronicle_index. This
-    is the heaviest of the three: three SQLite connects, two gunzip+parse passes
+    is the heaviest of the three: three database reads, two gunzip+parse passes
     and the centroid load, which a timelapse build issues once per day.
     """
     entry = ensure_map_access(map_name, authorization)

backend/src/api/data_routes.py

@@ -24,13 +24,10 @@
 )
 from ..scripts.chronicle.chapter_identity import write_chapter_identity
 from ..scripts.util.dirs import input_file, defines_file, validate_map
-# Imported at module scope on purpose, unlike capture_if_due below. A lazy
-# `from ... import` inside the handler re-runs the import machinery per request,
-# and src/skins/test_drinks.py permanently aliases sys.modules["src.skins.db"];
-# a lazy bind after those tests have run would hand the ledger a *different*
-# module object with an unpatched DB_PATH, silently reading and writing the
-# wrong database while reporting success. These modules import nothing that
-# imports this one, so there is no cycle to dodge.
+# Imported at module scope on purpose, unlike capture_if_due below: a lazy
+# `from ... import` inside the handler re-runs the import machinery per request.
+# These modules import nothing that imports this one, so there is no cycle to
+# dodge.
 from ..scripts.ledger import ingest as ledger_ingest
 from ..scripts.ledger.schema import (
     LEDGER_UPLOAD_MODE,

backend/src/api/ledger_routes.py

@@ -4,7 +4,7 @@
 `src/scripts/ledger/__init__.py` for why the two names exist side by side.
 
 Everything here is a plain `def`, not `async def`: every body is synchronous
-SQLite plus JSON encoding, so FastAPI runs it in the threadpool instead of
+database reads plus JSON encoding, so FastAPI runs it in the threadpool instead of
 blocking the event loop (same reasoning as `chronicle_routes.get_chronicle_index`).
 
 Shapes are frozen against `frontend/app/lib/map/ledgerData.ts`. The series
@@ -438,15 +438,19 @@ def get_ledger_index(
     """
     map_id = ensure_map_access(map_name, authorization).id
 
-    day_rows = store.list_days(map_id)
+    # One snapshot for the three reads, so the days and the registry agree.
+    with store.read_view() as conn:
+        day_rows = store.list_days(map_id, conn=conn)
+        latest_complete = store.latest_complete_day(map_id, conn=conn)
+        registry_rows = store.list_registry(map_id, conn=conn)
     days = [row["day"] for row in day_rows]
     server_days = [row["server_day"] for row in day_rows if row["server_day"] is not None]
 
     payload = {
         "days": days,
         "first": days[0] if days else None,
         "last": days[-1] if days else None,
-        "latest_complete_day": store.latest_complete_day(map_id),
+        "latest_complete_day": latest_complete,
         # A day is incomplete when its canonical snapshot never reported
         # `complete: true`. Absence on such a day is not deletion, and a client
         # must not read a dip in faction_count there as factions disbanding.
@@ -472,7 +476,7 @@ def get_ledger_index(
                 "deleted_day": row["deleted_day"],
                 "deleted_at": row["deleted_at"],
             }
-            for row in store.list_registry(map_id)
+            for row in registry_rows
         ],
     }
     return conditional_json_response(payload, if_none_match=if_none_match)
@@ -504,12 +508,11 @@ def get_ledger_series(
 
     requested_keys = _parse_faction_keys(factions)
 
-    # One connection for the whole request, passed through every store call
-    # below. Each of them used to open its own (plus its PRAGMA), which was
-    # seven connects for one response; sharing one also makes the day axis and
-    # the faction rows a single consistent view of the database.
-    conn = store.open_connection()
-    try:
+    # One connection and one snapshot for the whole request, passed through
+    # every store call below. Each of them used to open its own, which was
+    # seven connects for one response; sharing one also makes the day axis, the
+    # ETag inputs and the faction rows a single consistent view of the database.
+    with store.read_view() as conn:
         all_days = [row["day"] for row in store.list_days(map_id, conn=conn)]
         range_start, range_end, truncated = _resolve_range(all_days, start, end)
 
@@ -569,8 +572,6 @@ def get_ledger_series(
             map_id, range_start, range_end, keys, full=full, conn=conn
         )
         grouped = _group_by_key(rows)
-    finally:
-        conn.close()
 
     payload = {
         "days": days,
@@ -611,28 +612,38 @@ def get_ledger_faction(
     """
     map_id = ensure_map_access(map_name, authorization).id
 
-    all_days = [row["day"] for row in store.list_days(map_id)]
-    range_start, range_end, _ = _resolve_range(all_days, start, end)
-
-    registry = {row["faction_key"]: row for row in store.list_registry(map_id)}
-    if key not in registry and all_days:
-        # An indexed-row probe, not a range read: this only distinguishes
-        # "unknown key" from "known key with nothing in the requested window",
-        # and reading `all_days[0]..all_days[-1]` to answer it materialised the
-        # faction's entire history, sidestepping MAX_RANGE_DAYS on a route that
-        # otherwise enforces it.
-        if not store.faction_has_any_row(map_id, key):
-            return add_no_cache(JSONResponse({"error": "Faction not found"}, 404))
+    # One snapshot for every read below; see get_ledger_series.
+    with store.read_view() as conn:
+        all_days = [row["day"] for row in store.list_days(map_id, conn=conn)]
+        range_start, range_end, _ = _resolve_range(all_days, start, end)
 
-    day_rows = store.read_global_days(map_id, range_start, range_end) if all_days else []
+        registry = {
+            row["faction_key"]: row for row in store.list_registry(map_id, conn=conn)
+        }
+        if key not in registry and all_days:
+            # An indexed-row probe, not a range read: this only distinguishes
+            # "unknown key" from "known key with nothing in the requested
+            # window", and reading `all_days[0]..all_days[-1]` to answer it
+            # materialised the faction's entire history, sidestepping
+            # MAX_RANGE_DAYS on a route that otherwise enforces it.
+            if not store.faction_has_any_row(map_id, key, conn=conn):
+                return add_no_cache(JSONResponse({"error": "Faction not found"}, 404))
+
+        day_rows = (
+            store.read_global_days(map_id, range_start, range_end, conn=conn)
+            if all_days
+            else []
+        )
+        rows = (
+            store.read_faction_days(
+                map_id, range_start, range_end, [key], full=True, conn=conn
+            )
+            if all_days
+            else []
+        )
     days, index = _day_axis(day_rows)
     size = len(days)
 
-    rows = (
-        store.read_faction_days(map_id, range_start, range_end, [key], full=True)
-        if all_days
-        else []
-    )
     block = _faction_block(rows, index, size, full=True, registry=registry, key=key)
 
     overlord: list = [None] * size

backend/src/api/maps_routes.py

@@ -24,8 +24,10 @@ def _with_chronicle_day_flag(public: dict) -> dict:
     return out
 
 
+# `def`, not `async def`: archived chapters read the chronicle index from
+# Postgres, which must not block the event loop.
 @maps_router.get("/maps/accessible")
-async def get_accessible_maps(
+def get_accessible_maps(
     authorization: str | None = Header(default=None),
 ):
     return add_revalidate(JSONResponse({

backend/src/api/test_chronicle_archive_routes.py

@@ -56,7 +56,7 @@
 
 
 @pytest.fixture
-def archive_http_env(tmp_path, monkeypatch):
+def archive_http_env(tmp_path, monkeypatch, postgres):
     registry = tmp_path / "maps.yml"
     registry.write_text(TEST_REGISTRY, encoding="utf-8")
     monkeypatch.setenv("MAP_REGISTRY_PATH", str(registry))