diff --git a/backend/app/metrics.py b/backend/app/metrics.py index 89cf31af..b96bbef9 100644 --- a/backend/app/metrics.py +++ b/backend/app/metrics.py @@ -130,7 +130,12 @@ run_exports = Counter( "spire_codex_run_exports_total", - "Bulk run data export downloads", + "Bulk run data export downloads (unbounded full dumps)", +) + +run_export_pages = Counter( + "spire_codex_run_export_pages_total", + "Paginated run export page fetches (a bounded ?limit= request)", ) # ── Auth ──────────────────────────────────────────────────── diff --git a/backend/app/routers/exports.py b/backend/app/routers/exports.py index ff33b99e..d5ed1956 100644 --- a/backend/app/routers/exports.py +++ b/backend/app/routers/exports.py @@ -1,16 +1,19 @@ +import base64 import gzip import io import json import os import zipfile +from datetime import datetime, timezone from pathlib import Path -from fastapi import APIRouter, Request +from fastapi import APIRouter, HTTPException, Query, Request from fastapi.responses import StreamingResponse +from pymongo import ASCENDING from slowapi import Limiter from ..dependencies import VALID_LANGUAGES, client_ip -from ..metrics import data_exports, run_exports +from ..metrics import data_exports, run_export_pages, run_exports from ..services.data_service import DATA_DIR router = APIRouter(prefix="/api/exports", tags=["Exports"]) @@ -43,26 +46,120 @@ OFFICIAL_CHARACTERS = {"IRONCLAD", "SILENT", "DEFECT", "NECROBINDER", "REGENT"} +# Upper bound on a single page so one request can't ask for the whole corpus +# while still claiming the cheap (paginated) rate-limit cost. +MAX_PAGE_LIMIT = 50000 -def _official_run_hashes() -> list[str]: - """Get hashes of runs with official characters from Mongo.""" + +def _parse_iso(value: str, field: str) -> datetime: + """Parse an ISO-8601 timestamp param into an aware UTC datetime (400 on + malformed input). Naive timestamps are assumed UTC.""" + try: + dt = datetime.fromisoformat(value) + except ValueError: + raise HTTPException( + status_code=400, detail=f"invalid {field}: expected an ISO-8601 timestamp" + ) + if dt.tzinfo is None: + dt = dt.replace(tzinfo=timezone.utc) + return dt + + +def _encode_cursor(submitted_at, run_hash: str) -> str: + """Encode a keyset position as an opaque, URL-safe token. Runs missing a + submitted_at sort first, encoded with an empty timestamp.""" + sa = submitted_at.isoformat() if submitted_at is not None else "" + return base64.urlsafe_b64encode(f"{sa}|{run_hash}".encode("utf-8")).decode("ascii") + + +def _decode_cursor(token: str): + """Decode an X-Next-Cursor token back into (submitted_at|None, run_hash). + 400 on a malformed token. binascii.Error subclasses ValueError, so a bad + base64 payload is covered too.""" + try: + raw = base64.urlsafe_b64decode(token.encode("ascii")).decode("utf-8") + sa_str, run_hash = raw.split("|", 1) + except (ValueError, UnicodeDecodeError): + raise HTTPException(status_code=400, detail="invalid cursor") + submitted_at = _parse_iso(sa_str, "cursor") if sa_str else None + return submitted_at, run_hash + + +def _build_match(start, end, cursor) -> dict: + """Mongo filter for the export: official characters, an optional + half-open [start, end) submitted_at window, and an optional keyset + continuation strictly after `cursor`'s (submitted_at, _id).""" + clauses: list[dict] = [{"character": {"$in": list(OFFICIAL_CHARACTERS)}}] + + range_q: dict = {} + if start is not None: + range_q["$gte"] = start + if end is not None: + range_q["$lt"] = end + if range_q: + clauses.append({"submitted_at": range_q}) + + if cursor is not None: + sa, run_hash = cursor + if sa is None: + # Still inside the leading null/missing-submitted_at block: take + # the remaining nulls by _id, then everything with a timestamp. + clauses.append( + { + "$or": [ + {"submitted_at": None, "_id": {"$gt": run_hash}}, + {"submitted_at": {"$ne": None}}, + ] + } + ) + else: + clauses.append( + { + "$or": [ + {"submitted_at": {"$gt": sa}}, + {"submitted_at": sa, "_id": {"$gt": run_hash}}, + ] + } + ) + + return clauses[0] if len(clauses) == 1 else {"$and": clauses} + + +def _page_hashes(start, end, cursor, limit): + """Return (ordered_hashes, next_cursor). Runs are ordered by + (submitted_at, _id); next_cursor is None unless a bounded page is full + and at least one more run follows it.""" from ..services.runs_db_mongo import _get_collection coll = _get_collection() - cursor = coll.find( - {"character": {"$in": list(OFFICIAL_CHARACTERS)}}, - {"_id": 1}, + finder = coll.find( + _build_match(start, end, cursor), + {"_id": 1, "submitted_at": 1}, no_cursor_timeout=True, - ) + ).sort([("submitted_at", ASCENDING), ("_id", ASCENDING)]) + if limit is not None: + finder = finder.limit(limit + 1) # one extra row probes for a next page try: - return [doc["_id"] for doc in cursor] + docs = list(finder) finally: - cursor.close() + finder.close() + + next_cursor = None + if limit is not None and len(docs) > limit: + docs = docs[:limit] + last = docs[-1] + next_cursor = _encode_cursor(last.get("submitted_at"), last["_id"]) + return [doc["_id"] for doc in docs], next_cursor + +def _export_cost(request: Request) -> int: + """Rate-limit cost: a bounded (paginated) pull is cheap; an unbounded full + dump stays rare. 60 against the 120/hour bucket reproduces the historical + 2/hour ceiling for the full export.""" + return 1 if request.query_params.get("limit") else 60 -def _stream_runs_jsonl(): - hashes = _official_run_hashes() +def _stream_runs_jsonl(hashes): buf = io.BytesIO() gz = gzip.GzipFile(fileobj=buf, mode="wb") @@ -92,22 +189,75 @@ def _stream_runs_jsonl(): # Declared BEFORE the /{lang} route so FastAPI matches the literal # path "runs" instead of treating it as a language code. @router.get("/runs") -@limiter.limit("2/hour") -def export_runs(request: Request): - """Bulk export of all submitted runs as gzipped JSONL. +@limiter.limit("120/hour", cost=_export_cost) +def export_runs( + request: Request, + limit: int | None = Query( + None, + ge=1, + le=MAX_PAGE_LIMIT, + description="Max runs in this page. Omit for the full (unbounded) export.", + ), + start: str | None = Query( + None, description="Inclusive lower bound on submitted_at (ISO-8601)." + ), + end: str | None = Query( + None, description="Exclusive upper bound on submitted_at (ISO-8601)." + ), + cursor: str | None = Query( + None, + description="Opaque keyset token from a prior page's X-Next-Cursor header.", + ), +): + """Bulk export of submitted runs as gzipped JSONL. Each line is the full raw game JSON as submitted by the client, including players, map_point_history, acts, deck, relics, and card_choices. Only runs with official characters are included. + + Runs are ordered by ``(submitted_at, _id)``. With no params the response + is the full corpus (unchanged behaviour). To pull it in reliable chunks: + + * ``limit=N`` bounds the page to N runs. When more runs follow, the + response carries an ``X-Next-Cursor`` header; pass it back as + ``cursor=`` to fetch the next page. The ascending order means runs + submitted *during* a long bootstrap sort after the cursor, so a + forward pager never misses them. + * ``start`` / ``end`` restrict to a half-open ``[start, end)`` + submitted_at window (e.g. an incremental "everything since my last + sync"). Combine with ``limit`` to also bound each page. + + Consumer notes: + + * Terminate a paged walk on the **absence of ``X-Next-Cursor``**, never on + "fewer lines than ``limit``": a page is ``limit`` *runs*, but the streamed + line count can differ (missing/unreadable run files are skipped, and a + multiplayer run emits one raw line per player). + * The cursor does **not** embed the window, so keep ``start`` / ``end`` + constant across a paged sequence. + * ``start`` / ``end`` filter on ``submitted_at``, so they exclude legacy + runs that have no ``submitted_at``. Omit both to receive the whole corpus. + * Browser clients reading ``X-Next-Cursor`` need it in the CORS + ``Access-Control-Expose-Headers`` list; non-browser clients are unaffected. """ - run_exports.inc() + start_dt = _parse_iso(start, "start") if start else None + end_dt = _parse_iso(end, "end") if end else None + cursor_key = _decode_cursor(cursor) if cursor else None + + (run_export_pages if limit is not None else run_exports).inc() + hashes, next_cursor = _page_hashes(start_dt, end_dt, cursor_key, limit) + + headers = { + "Content-Disposition": 'attachment; filename="spire-codex-runs.jsonl.gz"', + "Cache-Control": "no-store", + } + if next_cursor is not None: + headers["X-Next-Cursor"] = next_cursor + return StreamingResponse( - _stream_runs_jsonl(), + _stream_runs_jsonl(hashes), media_type="application/gzip", - headers={ - "Content-Disposition": 'attachment; filename="spire-codex-runs.jsonl.gz"', - "Cache-Control": "no-store", - }, + headers=headers, ) diff --git a/backend/app/services/runs_db_mongo.py b/backend/app/services/runs_db_mongo.py index ba066ce2..c36baa84 100644 --- a/backend/app/services/runs_db_mongo.py +++ b/backend/app/services/runs_db_mongo.py @@ -23,11 +23,15 @@ {character: 1} {username: 1, submitted_at: -1} {submitted_at: -1} + {submitted_at: 1, _id: 1} (keyset-paginated run export) {character: 1, win: 1, ascension: 1} {build_id: 1} {"deck.id": 1} (multikey) {relics.id: 1} (multikey) {killed_by: 1} + +Schema validation (applied on import — idempotent, level "moderate"): + submitted_at must be a BSON date — see _ensure_run_validator. """ from __future__ import annotations @@ -140,6 +144,7 @@ def _get_collection(): # Database name comes from the connection string's default db. _coll = _client.get_default_database().runs _ensure_indexes(_coll) + _ensure_run_validator(_coll) return _coll @@ -149,6 +154,11 @@ def _ensure_indexes(coll) -> None: coll.create_index([("character", ASCENDING)]) coll.create_index([("username", ASCENDING), ("submitted_at", DESCENDING)]) coll.create_index([("submitted_at", DESCENDING)]) + # Ascending (submitted_at, _id) backs the keyset-paginated run export + # (GET /api/exports/runs ordered by (submitted_at, _id)). The {submitted_at: -1} + # index above is descending and lacks the _id tiebreaker, so it can't serve + # the ordered scan the cursor walks. + coll.create_index([("submitted_at", ASCENDING), ("_id", ASCENDING)]) coll.create_index( [("character", ASCENDING), ("win", ASCENDING), ("ascension", ASCENDING)] ) @@ -245,6 +255,59 @@ def _ensure_indexes(coll) -> None: ) +def _ensure_run_validator(coll) -> None: + """Require every run doc to carry a BSON-date ``submitted_at``. Idempotent; + applied on first collection access alongside the indexes, so a deploy turns + it on with no manual step. + + Why: the keyset run export (GET /api/exports/runs) orders by + ``(submitted_at, _id)``, and a forward pager only reliably sees new runs + because every new run sorts at the *end* — i.e. is stamped with a real + ``submitted_at`` (submit_run always sets ``now()``). A run inserted *without* + one would sort into the leading null block and be silently dropped from the + export for any pager that had already advanced past it. This makes that + invariant enforced rather than assumed: a future code path that tried to + insert a run with a missing / null / non-date ``submitted_at`` fails loudly + here instead of quietly corrupting the export. + + ``validationLevel="moderate"`` so the rule gates all new inserts but spares + *updates* to any pre-existing legacy doc whose ``submitted_at`` is null / + missing (the duplicate-key linking ``$set``, backfill_user_runs, claim_runs + keep working). Best-effort: if collMod can't run (permissions, or a + deliberate bulk-import window) it logs and continues — the export stays + correct as long as the submit path keeps stamping ``submitted_at``, which it + does; this is defense in depth, not a load-bearing dependency. + + NB for a bulk re-import of untimestamped legacy runs: ``moderate`` still + validates *inserts*, so such an import would now be rejected. Stamp + ``submitted_at`` in the import, or temporarily set ``validationAction`` + to ``"warn"`` for that window. + """ + try: + coll.database.command( + "collMod", + coll.name, + validator={ + "$jsonSchema": { + "required": ["submitted_at"], + "properties": { + "submitted_at": { + "bsonType": "date", + "description": ( + "must be a BSON date; keyset export ordering " + "depends on it" + ), + } + }, + } + }, + validationLevel="moderate", + validationAction="error", + ) + except Exception as e: + print(f"Warning: could not apply runs submitted_at validator: {e}") + + # ── helpers (mirrors of the sqlite module) ────────────────────────────── def clean_id(raw_id: str) -> str: """Strip CARD./RELIC./MONSTER./etc. prefixes (matches runs_db.clean_id).""" diff --git a/backend/pytest.ini b/backend/pytest.ini new file mode 100644 index 00000000..c7b23ecb --- /dev/null +++ b/backend/pytest.ini @@ -0,0 +1,3 @@ +[pytest] +pythonpath = . +testpaths = tests diff --git a/backend/requirements-dev.txt b/backend/requirements-dev.txt new file mode 100644 index 00000000..964a047e --- /dev/null +++ b/backend/requirements-dev.txt @@ -0,0 +1,5 @@ +# Dev/test-only dependencies. Install on top of requirements.txt: +# pip install -r requirements.txt -r requirements-dev.txt +# Run the tests from the backend/ dir: +# pytest +pytest>=8,<10 diff --git a/backend/tests/test_export_helpers.py b/backend/tests/test_export_helpers.py new file mode 100644 index 00000000..55d7eb0c --- /dev/null +++ b/backend/tests/test_export_helpers.py @@ -0,0 +1,140 @@ +"""Unit tests for the run-export pagination helpers. + +Pure-function coverage of the cursor codec, the keyset/range Mongo match +builder, ISO parsing, and the rate-limit cost - none of which touch Mongo +(the router imports `_get_collection` lazily, so these run with no database +and no FastAPI app). Run from the backend/ dir: `pytest`. +""" + +from datetime import datetime, timezone + +import pytest +from fastapi import HTTPException + +from app.routers.exports import ( + OFFICIAL_CHARACTERS, + _build_match, + _decode_cursor, + _encode_cursor, + _export_cost, + _parse_iso, +) + + +# --- _parse_iso ------------------------------------------------------------ + +def test_parse_iso_aware_passthrough(): + dt = _parse_iso("2026-06-22T19:51:28+00:00", "start") + assert dt == datetime(2026, 6, 22, 19, 51, 28, tzinfo=timezone.utc) + + +def test_parse_iso_naive_assumed_utc(): + dt = _parse_iso("2026-06-22T00:00:00", "start") + assert dt.tzinfo == timezone.utc + + +def test_parse_iso_rejects_garbage(): + with pytest.raises(HTTPException) as err: + _parse_iso("not-a-date", "start") + assert err.value.status_code == 400 + + +# --- cursor codec ---------------------------------------------------------- + +def test_cursor_roundtrip_with_timestamp(): + dt = datetime(2026, 6, 22, 19, 51, 28, 664000, tzinfo=timezone.utc) + token = _encode_cursor(dt, "aaaabbbbcccc1111") + sa, run_hash = _decode_cursor(token) + assert sa == dt + assert run_hash == "aaaabbbbcccc1111" + + +def test_cursor_roundtrip_null_submitted_at(): + # Legacy/null block: submitted_at encodes as empty and decodes back to None. + token = _encode_cursor(None, "aaaabbbbcccc1111") + sa, run_hash = _decode_cursor(token) + assert sa is None + assert run_hash == "aaaabbbbcccc1111" + + +def test_decode_cursor_rejects_bad_base64(): + # "x" is a malformed base64 payload (1 char, an invalid length), so the + # decode raises binascii.Error (a ValueError subclass) before any "|" + # split - exercising the base64 guard specifically, not the UTF-8 one. + with pytest.raises(HTTPException) as err: + _decode_cursor("x") + assert err.value.status_code == 400 + + +def test_decode_cursor_rejects_missing_separator(): + import base64 + + # Valid base64 but no "|" separator -> the unpack fails -> 400. + token = base64.urlsafe_b64encode(b"noseparator").decode("ascii") + with pytest.raises(HTTPException) as err: + _decode_cursor(token) + assert err.value.status_code == 400 + + +# --- _build_match ---------------------------------------------------------- + +def _char_clause(match): + """Pull the official-character clause out of a match (order-independent).""" + clause = match if "character" in match else next( + c for c in match["$and"] if "character" in c + ) + return set(clause["character"]["$in"]) + + +def test_build_match_no_params_is_just_character_filter(): + match = _build_match(None, None, None) + # No range/cursor -> the single character clause, not wrapped in $and. + assert "$and" not in match + assert _char_clause(match) == OFFICIAL_CHARACTERS + + +def test_build_match_half_open_range(): + start = datetime(2026, 6, 1, tzinfo=timezone.utc) + end = datetime(2026, 6, 20, tzinfo=timezone.utc) + match = _build_match(start, end, None) + range_clause = next(c for c in match["$and"] if "submitted_at" in c) + assert range_clause["submitted_at"] == {"$gte": start, "$lt": end} + + +def test_build_match_start_only(): + start = datetime(2026, 6, 1, tzinfo=timezone.utc) + match = _build_match(start, None, None) + range_clause = next(c for c in match["$and"] if "submitted_at" in c) + assert range_clause["submitted_at"] == {"$gte": start} + + +def test_build_match_cursor_inside_null_block(): + # Cursor with sa=None: take remaining nulls by _id, then everything timed. + match = _build_match(None, None, (None, "abc")) + or_clause = next(c for c in match["$and"] if "$or" in c)["$or"] + assert {"submitted_at": None, "_id": {"$gt": "abc"}} in or_clause + assert {"submitted_at": {"$ne": None}} in or_clause + + +def test_build_match_cursor_past_nulls(): + sa = datetime(2026, 6, 10, tzinfo=timezone.utc) + match = _build_match(None, None, (sa, "abc")) + or_clause = next(c for c in match["$and"] if "$or" in c)["$or"] + # Strictly after (sa, abc) in (submitted_at, _id) order. + assert {"submitted_at": {"$gt": sa}} in or_clause + assert {"submitted_at": sa, "_id": {"$gt": "abc"}} in or_clause + + +# --- _export_cost ---------------------------------------------------------- + +class _FakeRequest: + def __init__(self, query): + self.query_params = query + + +def test_export_cost_paged_is_cheap(): + assert _export_cost(_FakeRequest({"limit": "5000"})) == 1 + + +def test_export_cost_full_dump_is_expensive(): + assert _export_cost(_FakeRequest({})) == 60