diff --git a/ai/tasks/done/TASK-POSTGRES-HISTORICAL-PHASE-2.md b/ai/tasks/done/TASK-POSTGRES-HISTORICAL-PHASE-2.md new file mode 100644 index 0000000..2bbe699 --- /dev/null +++ b/ai/tasks/done/TASK-POSTGRES-HISTORICAL-PHASE-2.md @@ -0,0 +1,135 @@ +--- +id: TASK-POSTGRES-HISTORICAL-PHASE-2 +title: Migrate displayed historical data to PostgreSQL +status: done +type: backend +team: Backend Senior +supporting_teams: + - Arquitecto de Base de Datos + - Arquitecto Python +roadmap_item: foundation +priority: high +--- + +# TASK-POSTGRES-HISTORICAL-PHASE-2 - Migrate displayed historical data to PostgreSQL + +## Goal + +Make PostgreSQL authoritative for historical and server data currently displayed +by the frontend, add an idempotent SQLite/file migration command, and close the +phase-1 datetime serialization and materialized detail lookup regressions. + +## Context + +Phase 1 moved the live RCON historical write path to PostgreSQL, but historical +frontend pages still read older SQLite and file-backed data for ranking, +scoreboard fallback, snapshots, server cache, and some match-detail continuity. +PostgreSQL reads also expose native datetime values that can terminate JSON API +responses. The phase-2 migration must preserve existing historical URLs and safe +scoreboard links without making SQLite the active source of truth again. + +## Steps + +1. Inspect the PostgreSQL RCON schema, SQLite historical schema, snapshot/server + reads, and affected API routes. +2. Add PostgreSQL-backed displayed read models plus deterministic migration from + existing SQLite/files. +3. Fix JSON-safe datetime handling and recent/detail materialized match + consistency with focused regression tests. +4. Update diagnostics and docs, then run the requested local and Docker checks + where available. + +## Files to Read First + +- `ai/architecture-index.md` +- `ai/repo-context.md` +- `backend/app/postgres_rcon_storage.py` +- `backend/app/historical_api.py` +- `backend/app/historical_snapshots.py` +- `backend/app/storage_diagnostics.py` + +## Expected Files to Modify + +- `backend/app/postgres_rcon_storage.py` +- `backend/app/sqlite_to_postgres_migration.py` +- `backend/app/storage_diagnostics.py` +- backend historical/server storage and API modules needed to move displayed + SQLite/file reads to PostgreSQL +- focused backend tests for datetime and detail lookup regressions +- `backend/README.md` +- `docs/decisions.md` + +## Constraints + +- Do not return displayed historical data to SQLite as the main fix. +- Preserve old match ids, match keys, public scoreboard safe URLs, and existing + historical data where present. +- Do not re-enable Comunidad Hispana #03 as an active configured target. +- Keep Elo/MMR out of phase 2 unless a visible API requires its PostgreSQL data. +- Do not modify unrelated frontend behavior or commit runtime database files. + +## Validation + +- Run the requested compile, diagnostics, migration, Node syntax, RCON pipeline, + integration, Docker Compose, container diagnostics, materialization, and HTTP + smoke commands where the local environment permits them. +- Review `git diff --name-only` and confirm changed files match this task. + +## Outcome + +Implemented PostgreSQL phase 2 for the displayed historical surface. The new +displayed PostgreSQL schema stores migrated public-scoreboard match/player +tables, historical snapshot payloads, live server cache rows and player-event +ledger rows alongside the phase-1 RCON tables. API storage facades now read +those PostgreSQL stores when `HLL_BACKEND_DATABASE_URL` is configured; SQLite +and snapshot JSON remain migration inputs or explicit local fixtures. + +`python -m app.sqlite_to_postgres_migration` now discovers legacy SQLite files +and historical snapshot JSON under the configured backend data directory, copies +stable IDs/keys with idempotent conflict handling, advances PostgreSQL +sequences, skips visible `comunidad-hispana-03` legacy scope, preserves safe +scoreboard URLs and prints JSON row counts/errors. A repeat container run read +the existing source rows and reported zero inserted rows with no errors. + +Required fixes included global HTTP JSON encoding for PostgreSQL `date` and +`datetime` values, a materialized detail lookup guard for match keys that embed +the requested server key, reduced repeated RCON PostgreSQL read-time DDL to +avoid schema-lock chains during materialization/diagnostics, and focused tests +for JSON encoding plus recent/detail match-id consistency. + +Post-migration container diagnostics showed PostgreSQL counts including: + +- `admin_log_events=21242` +- `materialized_matches=58` +- `player_stats=3824` +- `public_scoreboard_historical_matches=10030` +- ranking source stats `1090704` +- displayed snapshots `51` +- player-event ledger `14715` +- safe scoreboard candidates `400` + +Remaining SQLite/file-backed boundaries are non-visible phase-3 scope only: +public-scoreboard import run/backfill bookkeeping and paused Elo/MMR tables. + +Validation completed: + +- `python -m compileall backend/app` +- `python -m unittest discover -s tests -p test_json_serialization.py` + from `backend/` +- `python -m app.storage_diagnostics` locally and in the backend container +- `node --check` for the four requested frontend JavaScript files +- `powershell -ExecutionPolicy Bypass -File scripts/run-rcon-data-pipeline-tests.ps1` +- `powershell -ExecutionPolicy Bypass -File scripts/run-integration-tests.ps1` +- requested advanced Docker Compose down/up/build/ps flow +- container migration, diagnostics, AdminLog ingestion and materialization +- requested HTTP probes for `/health`, recent matches and both known detail keys +- direct PostgreSQL-backed snapshot and direct monthly leaderboard HTTP probes + +The RCON pipeline script still prints existing SQLite `ResourceWarning` noise +from its unittest fallback suites, but the script and suites passed. + +## Change Budget + +This migration intentionally crosses backend schema, read models, diagnostics, +tests, and docs. Keep each change tied to displayed data ownership or the +required migration/serialization/detail consistency fixes. diff --git a/backend/README.md b/backend/README.md index bd1d7d1..5e5ff73 100644 --- a/backend/README.md +++ b/backend/README.md @@ -1579,6 +1579,41 @@ Estado real a fecha de esta fase: RCON-backed primario y usar `public-scoreboard` solo como suplemento/fallback para estadisticas por jugador sin paridad RCON +## PostgreSQL Phase 2 Displayed Data Migration + +Cuando `HLL_BACKEND_DATABASE_URL` esta configurado, los endpoints visibles de +historico y el cache mostrado por `/api/servers` leen PostgreSQL. SQLite y los +JSON legacy quedan como fuente de migracion o fixture explicito con `db_path`. + +Migracion idempotente: + +```powershell +cd backend +python -m app.sqlite_to_postgres_migration +python -m app.storage_diagnostics +``` + +La salida JSON de `sqlite_to_postgres_migration` lista rutas fuente, dominios y +tablas migradas, filas leidas, insertadas, actualizadas, omitidas y errores. +La migracion conserva `external_match_id`, IDs legacy y `match_key` RCON para +que URLs de detalle existentes sigan resolviendo. Tambien copia candidatos y +URLs seguras de scoreboard; no vuelve a activar filas visibles de +`comunidad-hispana-03`. + +Paridad minima a revisar en `storage_diagnostics`: + +- `admin_log_events`, `materialized_matches`, `player_stats` +- `public_scoreboard_historical_matches` +- fuentes de rankings semanales y mensuales +- `server_summary_cache`, `server_snapshots`, `player_event_ledger` +- `scoreboard_candidates` +- ultimas partidas materializadas y ultimos eventos AdminLog `match_end` + +Fuera de phase 2 quedan checkpoints/runs de ingesta publica que no se muestran +en frontend y Elo/MMR pausado. Si un endpoint de mantenimiento recibe un +`db_path` explicito, sigue trabajando contra SQLite para migracion, tests o +compatibilidad operativa controlada. + ## Elo/MMR Monthly Ranking Se añade una primera base operativa inspirada en el documento diff --git a/backend/app/historical_snapshot_storage.py b/backend/app/historical_snapshot_storage.py index f3f596a..d81706c 100644 --- a/backend/app/historical_snapshot_storage.py +++ b/backend/app/historical_snapshot_storage.py @@ -6,7 +6,7 @@ import json from datetime import datetime, timezone from pathlib import Path -from .config import get_storage_path +from .config import get_storage_path, use_postgres_rcon_storage from .historical_models import HistoricalSnapshotRecord from .historical_snapshots import validate_snapshot_identity @@ -46,6 +46,22 @@ def persist_historical_snapshot( raise ValueError("server_key is required for historical snapshots.") validate_snapshot_identity(snapshot_type=snapshot_type, metric=metric) + if use_postgres_rcon_storage(explicit_sqlite_path=db_path): + from .postgres_display_storage import persist_snapshot_record + + return persist_snapshot_record( + { + "server_key": normalized_server_key, + "snapshot_type": snapshot_type, + "metric": metric, + "window": window, + "generated_at": generated_at or datetime.now(timezone.utc), + "source_range_start": source_range_start, + "source_range_end": source_range_end, + "is_stale": is_stale, + "payload": payload, + } + ) snapshots_root = initialize_historical_snapshot_storage(db_path=db_path) generated_at_value = _as_utc(generated_at or datetime.now(timezone.utc)) payload_json = json.dumps(payload, ensure_ascii=True) @@ -148,6 +164,15 @@ def get_historical_snapshot( ) -> dict[str, object] | None: """Return one persisted snapshot and decoded payload, if present.""" validate_snapshot_identity(snapshot_type=snapshot_type, metric=metric) + if use_postgres_rcon_storage(explicit_sqlite_path=db_path): + from .postgres_display_storage import get_snapshot + + return get_snapshot( + server_key=server_key, + snapshot_type=snapshot_type, + metric=metric, + window=window, + ) snapshots_root = resolve_historical_snapshot_storage_path(db_path=db_path) snapshot_path = _build_snapshot_path( snapshots_root=snapshots_root, diff --git a/backend/app/historical_snapshots.py b/backend/app/historical_snapshots.py index 0182339..8477617 100644 --- a/backend/app/historical_snapshots.py +++ b/backend/app/historical_snapshots.py @@ -6,7 +6,7 @@ import sqlite3 from datetime import datetime, timezone from pathlib import Path -from .config import get_historical_data_source_kind +from .config import get_database_url, get_historical_data_source_kind from .data_sources import SOURCE_KIND_RCON, get_rcon_historical_read_model from .historical_storage import ( ALL_SERVERS_SLUG, @@ -726,6 +726,10 @@ def _build_player_event_scope_where(*, server_key: str) -> tuple[str, list[objec def _connect(db_path: Path) -> sqlite3.Connection: + if get_database_url(): + from .postgres_display_storage import connect_postgres_compat + + return connect_postgres_compat() connection = sqlite3.connect(db_path) connection.row_factory = sqlite3.Row return connection diff --git a/backend/app/historical_storage.py b/backend/app/historical_storage.py index 379773d..f8c4674 100644 --- a/backend/app/historical_storage.py +++ b/backend/app/historical_storage.py @@ -12,6 +12,7 @@ from .config import ( get_historical_weekly_fallback_max_weekday, get_historical_weekly_fallback_min_matches, get_storage_path, + use_postgres_rcon_storage, ) from .historical_models import HistoricalServerDefinition from .monthly_mvp import build_monthly_mvp_rankings @@ -739,6 +740,10 @@ def list_recent_historical_matches( db_path: Path | None = None, ) -> list[dict[str, object]]: """Return recent persisted matches grouped for the historical API layer.""" + if use_postgres_rcon_storage(explicit_sqlite_path=db_path): + from .postgres_display_storage import list_recent_scoreboard_matches + + return list_recent_scoreboard_matches(server_slug=server_slug, limit=limit) resolved_path = initialize_historical_storage(db_path=db_path) where_clause = "" params: list[object] = [] @@ -821,6 +826,13 @@ def get_historical_match_detail( normalized_match_id = _stringify(match_id) if not normalized_server_slug or not normalized_match_id: return None + if use_postgres_rcon_storage(explicit_sqlite_path=db_path): + from .postgres_display_storage import get_scoreboard_match_detail + + return get_scoreboard_match_detail( + server_slug=normalized_server_slug, + match_id=normalized_match_id, + ) resolved_path = initialize_historical_storage(db_path=db_path) with _connect(resolved_path) as connection: row = connection.execute( @@ -939,6 +951,10 @@ def list_historical_server_summaries( db_path: Path | None = None, ) -> list[dict[str, object]]: """Return aggregate historical metrics per server.""" + if use_postgres_rcon_storage(explicit_sqlite_path=db_path): + from .postgres_display_storage import list_scoreboard_server_summaries + + return list_scoreboard_server_summaries(server_slug=server_slug) resolved_path = initialize_historical_storage(db_path=db_path) if _is_all_servers_selector(server_slug): return [_build_all_servers_summary(db_path=resolved_path)] @@ -1247,6 +1263,15 @@ def list_weekly_leaderboard( db_path: Path | None = None, ) -> dict[str, object]: """Return ranked weekly leaderboard totals from persisted historical match stats.""" + if use_postgres_rcon_storage(explicit_sqlite_path=db_path): + from .postgres_display_storage import list_scoreboard_leaderboard + + return list_scoreboard_leaderboard( + timeframe="weekly", + metric=metric, + server_id=server_id, + limit=limit, + ) resolved_path = initialize_historical_storage(db_path=db_path) aggregate_all_servers = _is_all_servers_selector(server_id) current_time = datetime.now(timezone.utc) @@ -1432,6 +1457,15 @@ def list_monthly_leaderboard( db_path: Path | None = None, ) -> dict[str, object]: """Return ranked monthly leaderboard totals from persisted historical match stats.""" + if use_postgres_rcon_storage(explicit_sqlite_path=db_path): + from .postgres_display_storage import list_scoreboard_leaderboard + + return list_scoreboard_leaderboard( + timeframe="monthly", + metric=metric, + server_id=server_id, + limit=limit, + ) resolved_path = initialize_historical_storage(db_path=db_path) aggregate_all_servers = _is_all_servers_selector(server_id) current_time = datetime.now(timezone.utc) diff --git a/backend/app/main.py b/backend/app/main.py index 18e3265..249b3bc 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -3,6 +3,7 @@ from __future__ import annotations import json +from datetime import date, datetime from http import HTTPStatus from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer @@ -38,7 +39,7 @@ class HealthHandler(BaseHTTPRequestHandler): return def _write_json(self, status: HTTPStatus, payload: dict[str, object]) -> None: - body = json.dumps(payload).encode("utf-8") + body = json.dumps(payload, default=_json_default).encode("utf-8") self.send_response(status) self._send_default_headers(content_length=len(body)) self.end_headers() @@ -61,6 +62,13 @@ def create_server() -> ThreadingHTTPServer: return ThreadingHTTPServer((host, port), HealthHandler) +def _json_default(value: object) -> str: + """Serialize PostgreSQL date/time values before they can abort an HTTP response.""" + if isinstance(value, (date, datetime)): + return value.isoformat() + raise TypeError(f"Object of type {type(value).__name__} is not JSON serializable") + + def run() -> None: """Start the local bootstrap server.""" host, port = get_bind_address() diff --git a/backend/app/player_event_aggregates.py b/backend/app/player_event_aggregates.py index d95449d..8229213 100644 --- a/backend/app/player_event_aggregates.py +++ b/backend/app/player_event_aggregates.py @@ -5,7 +5,7 @@ from __future__ import annotations import sqlite3 from pathlib import Path -from .config import get_storage_path +from .config import get_database_url, get_storage_path from .player_event_storage import initialize_player_event_storage @@ -249,6 +249,10 @@ def _build_common_where( def _connect(db_path: Path) -> sqlite3.Connection: + if get_database_url(): + from .postgres_display_storage import connect_postgres_compat + + return connect_postgres_compat() connection = sqlite3.connect(db_path or get_storage_path()) connection.row_factory = sqlite3.Row return connection diff --git a/backend/app/player_event_storage.py b/backend/app/player_event_storage.py index c55f5b4..1997583 100644 --- a/backend/app/player_event_storage.py +++ b/backend/app/player_event_storage.py @@ -7,7 +7,11 @@ from collections.abc import Iterable from datetime import datetime, timedelta, timezone from pathlib import Path -from .config import get_player_event_refresh_overlap_hours, get_storage_path +from .config import ( + get_player_event_refresh_overlap_hours, + get_storage_path, + use_postgres_rcon_storage, +) from .player_event_models import PlayerEventRecord from .sqlite_utils import connect_sqlite_writer @@ -95,6 +99,10 @@ def upsert_player_events( db_path: Path | None = None, ) -> dict[str, int]: """Insert normalized events idempotently into the raw ledger.""" + if use_postgres_rcon_storage(explicit_sqlite_path=db_path): + from .postgres_display_storage import upsert_player_event_rows + + return upsert_player_event_rows(events) resolved_path = initialize_player_event_storage(db_path=db_path) inserted = 0 duplicates = 0 diff --git a/backend/app/postgres_display_storage.py b/backend/app/postgres_display_storage.py new file mode 100644 index 0000000..2b73044 --- /dev/null +++ b/backend/app/postgres_display_storage.py @@ -0,0 +1,916 @@ +"""PostgreSQL read/write storage for data displayed outside the RCON write path.""" + +from __future__ import annotations + +import json +from contextlib import contextmanager +from datetime import datetime, timedelta, timezone +from pathlib import Path +from typing import Any, Iterable, Mapping + +from .config import get_database_url, get_historical_weekly_fallback_max_weekday +from .historical_models import HistoricalSnapshotRecord +from .scoreboard_origins import resolve_trusted_scoreboard_match_url + + +ALL_SERVERS_SLUG = "all-servers" +ALL_SERVERS_DISPLAY_NAME = "Todos" +SUMMARY_SNAPSHOT_LIMIT = 6 + + +DISPLAY_SCHEMA_SQL = """ +CREATE TABLE IF NOT EXISTS game_sources ( + id BIGSERIAL PRIMARY KEY, + slug TEXT NOT NULL UNIQUE, + display_name TEXT NOT NULL, + provider_kind TEXT NOT NULL, + is_active BOOLEAN NOT NULL DEFAULT TRUE, + created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP, + updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP +); +CREATE TABLE IF NOT EXISTS servers ( + id BIGSERIAL PRIMARY KEY, + game_source_id BIGINT NOT NULL REFERENCES game_sources(id), + external_server_id TEXT, + server_name TEXT NOT NULL, + region TEXT, + first_seen_at TEXT NOT NULL, + last_seen_at TEXT NOT NULL, + created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP, + updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP, + UNIQUE(game_source_id, external_server_id) +); +CREATE TABLE IF NOT EXISTS server_snapshots ( + id BIGSERIAL PRIMARY KEY, + server_id BIGINT NOT NULL REFERENCES servers(id), + captured_at TEXT NOT NULL, + status TEXT NOT NULL, + players INTEGER, + max_players INTEGER, + current_map TEXT, + source_name TEXT NOT NULL, + snapshot_origin TEXT, + source_ref TEXT, + raw_payload_ref TEXT, + created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP, + UNIQUE(server_id, captured_at, source_name, source_ref) +); +CREATE INDEX IF NOT EXISTS idx_pg_server_snapshots_server_time +ON server_snapshots(server_id, captured_at DESC); + +CREATE TABLE IF NOT EXISTS historical_servers ( + id BIGSERIAL PRIMARY KEY, + slug TEXT NOT NULL UNIQUE, + display_name TEXT NOT NULL, + scoreboard_base_url TEXT NOT NULL UNIQUE, + server_number INTEGER, + source_kind TEXT NOT NULL, + created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP, + updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP +); +CREATE TABLE IF NOT EXISTS historical_maps ( + id BIGSERIAL PRIMARY KEY, + external_map_id TEXT UNIQUE, + map_name TEXT, + pretty_name TEXT, + game_mode TEXT, + image_name TEXT, + created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP, + updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP +); +CREATE TABLE IF NOT EXISTS historical_matches ( + id BIGSERIAL PRIMARY KEY, + historical_server_id BIGINT NOT NULL REFERENCES historical_servers(id), + external_match_id TEXT NOT NULL, + historical_map_id BIGINT REFERENCES historical_maps(id), + created_at_source TEXT, + started_at TEXT, + ended_at TEXT, + map_name TEXT, + map_pretty_name TEXT, + game_mode TEXT, + image_name TEXT, + allied_score INTEGER, + axis_score INTEGER, + last_seen_at TEXT NOT NULL, + raw_payload_ref TEXT, + created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP, + updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP, + UNIQUE(historical_server_id, external_match_id) +); +CREATE TABLE IF NOT EXISTS historical_players ( + id BIGSERIAL PRIMARY KEY, + stable_player_key TEXT NOT NULL UNIQUE, + display_name TEXT NOT NULL, + steam_id TEXT, + source_player_id TEXT, + first_seen_at TEXT NOT NULL, + last_seen_at TEXT NOT NULL, + created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP, + updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP +); +CREATE TABLE IF NOT EXISTS historical_player_match_stats ( + id BIGSERIAL PRIMARY KEY, + historical_match_id BIGINT NOT NULL REFERENCES historical_matches(id), + historical_player_id BIGINT NOT NULL REFERENCES historical_players(id), + match_player_ref TEXT, + team_side TEXT, + level INTEGER, + kills INTEGER, + deaths INTEGER, + teamkills INTEGER, + time_seconds INTEGER, + kills_per_minute DOUBLE PRECISION, + deaths_per_minute DOUBLE PRECISION, + kill_death_ratio DOUBLE PRECISION, + combat INTEGER, + offense INTEGER, + defense INTEGER, + support INTEGER, + created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP, + updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP, + UNIQUE(historical_match_id, historical_player_id) +); +CREATE INDEX IF NOT EXISTS idx_pg_historical_matches_server_end +ON historical_matches(historical_server_id, ended_at DESC, started_at DESC); +CREATE INDEX IF NOT EXISTS idx_pg_historical_player_stats_match +ON historical_player_match_stats(historical_match_id); + +CREATE TABLE IF NOT EXISTS displayed_historical_snapshots ( + server_key TEXT NOT NULL, + snapshot_type TEXT NOT NULL, + metric TEXT NOT NULL DEFAULT '', + snapshot_window TEXT NOT NULL DEFAULT '', + payload_json TEXT NOT NULL, + generated_at TEXT NOT NULL, + source_range_start TEXT, + source_range_end TEXT, + is_stale BOOLEAN NOT NULL DEFAULT FALSE, + created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP, + updated_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP, + PRIMARY KEY(server_key, snapshot_type, metric, snapshot_window) +); + +CREATE TABLE IF NOT EXISTS player_event_raw_ledger ( + id BIGSERIAL PRIMARY KEY, + event_id TEXT NOT NULL UNIQUE, + event_type TEXT NOT NULL, + occurred_at TEXT, + server_slug TEXT NOT NULL, + external_match_id TEXT NOT NULL, + source_kind TEXT NOT NULL, + source_ref TEXT, + raw_event_ref TEXT, + killer_player_key TEXT, + killer_display_name TEXT, + victim_player_key TEXT, + victim_display_name TEXT, + weapon_name TEXT, + weapon_category TEXT, + kill_category TEXT, + is_teamkill BOOLEAN NOT NULL DEFAULT FALSE, + event_value INTEGER NOT NULL DEFAULT 1, + inserted_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP +); +CREATE INDEX IF NOT EXISTS idx_pg_player_event_raw_occurred_at +ON player_event_raw_ledger(occurred_at DESC); +""" + + +def initialize_postgres_display_storage() -> None: + with connect_postgres() as connection: + connection.execute(DISPLAY_SCHEMA_SQL) + + +def connect_postgres(): + try: + import psycopg + from psycopg.rows import dict_row + except ImportError as error: # pragma: no cover - environment-specific + raise RuntimeError("psycopg is required when HLL_BACKEND_DATABASE_URL is set.") from error + database_url = get_database_url() + if not database_url: + raise RuntimeError("HLL_BACKEND_DATABASE_URL is required for displayed PostgreSQL storage.") + return psycopg.connect(database_url, row_factory=dict_row) + + +class PostgresCompatConnection: + """Small placeholder shim for SQLite-shaped displayed read queries.""" + + def __init__(self, connection: Any): + self.connection = connection + + def execute(self, sql: str, params: Iterable[object] | None = None): + return self.connection.execute(sql.replace("?", "%s"), tuple(params or ())) + + +@contextmanager +def connect_postgres_compat(): + initialize_postgres_display_storage() + with connect_postgres() as connection: + yield PostgresCompatConnection(connection) + + +def persist_snapshot_record(snapshot: Mapping[str, object]) -> HistoricalSnapshotRecord: + initialize_postgres_display_storage() + generated_at = _iso(snapshot.get("generated_at")) or _utc_now_iso() + metric = str(snapshot.get("metric") or "") + window = str(snapshot.get("window") or "") + payload = snapshot.get("payload") + payload_json = json.dumps(payload, ensure_ascii=True, separators=(",", ":")) + with connect_postgres() as connection: + connection.execute( + """ + INSERT INTO displayed_historical_snapshots ( + server_key, snapshot_type, metric, snapshot_window, payload_json, generated_at, + source_range_start, source_range_end, is_stale + ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s) + ON CONFLICT(server_key, snapshot_type, metric, snapshot_window) DO UPDATE SET + payload_json = EXCLUDED.payload_json, + generated_at = EXCLUDED.generated_at, + source_range_start = EXCLUDED.source_range_start, + source_range_end = EXCLUDED.source_range_end, + is_stale = EXCLUDED.is_stale, + updated_at = CURRENT_TIMESTAMP + """, + ( + str(snapshot["server_key"]), + str(snapshot["snapshot_type"]), + metric, + window, + payload_json, + generated_at, + _iso(snapshot.get("source_range_start")), + _iso(snapshot.get("source_range_end")), + bool(snapshot.get("is_stale", False)), + ), + ) + return HistoricalSnapshotRecord( + server_key=str(snapshot["server_key"]), + snapshot_type=str(snapshot["snapshot_type"]), + metric=metric or None, + window=window or None, + payload_json=payload_json, + generated_at=_parse_datetime(generated_at) or datetime.now(timezone.utc), + source_range_start=_parse_datetime(_iso(snapshot.get("source_range_start"))), + source_range_end=_parse_datetime(_iso(snapshot.get("source_range_end"))), + is_stale=bool(snapshot.get("is_stale", False)), + ) + + +def get_snapshot( + *, + server_key: str, + snapshot_type: str, + metric: str | None, + window: str | None, +) -> dict[str, object] | None: + initialize_postgres_display_storage() + with connect_postgres() as connection: + row = connection.execute( + """ + SELECT * + FROM displayed_historical_snapshots + WHERE server_key = %s AND snapshot_type = %s AND metric = %s AND snapshot_window = %s + """, + (server_key, snapshot_type, metric or "", window or ""), + ).fetchone() + if not row: + return None + return { + "server_key": row["server_key"], + "snapshot_type": row["snapshot_type"], + "metric": row["metric"] or None, + "window": row["snapshot_window"] or None, + "generated_at": row["generated_at"], + "source_range_start": row["source_range_start"], + "source_range_end": row["source_range_end"], + "is_stale": bool(row["is_stale"]), + "payload": json.loads(row["payload_json"]), + } + + +def list_latest_server_snapshots() -> list[dict[str, object]]: + initialize_postgres_display_storage() + with connect_postgres() as connection: + rows = connection.execute( + """ + SELECT s.id AS server_id, s.external_server_id, s.server_name, s.region, + g.slug AS context, snap.source_name, snap.snapshot_origin, + snap.source_ref, snap.captured_at, snap.status, snap.players, + snap.max_players, snap.current_map + FROM servers AS s + JOIN game_sources AS g ON g.id = s.game_source_id + JOIN server_snapshots AS snap ON snap.server_id = s.id + JOIN ( + SELECT server_id, MAX(captured_at) AS captured_at + FROM server_snapshots GROUP BY server_id + ) AS latest ON latest.server_id = snap.server_id + AND latest.captured_at = snap.captured_at + ORDER BY s.server_name ASC + """ + ).fetchall() + return [_attach_server_history(connection, dict(row)) for row in rows] + + +def persist_server_snapshots( + snapshots: Iterable[Mapping[str, object]], + *, + source_name: str, + captured_at: str, + game_source: Mapping[str, str], +) -> dict[str, object]: + initialize_postgres_display_storage() + persisted = 0 + with connect_postgres() as connection: + source = connection.execute( + """ + INSERT INTO game_sources (slug, display_name, provider_kind, is_active) + VALUES (%s, %s, %s, TRUE) + ON CONFLICT(slug) DO UPDATE SET + display_name = EXCLUDED.display_name, + provider_kind = EXCLUDED.provider_kind, + is_active = TRUE, + updated_at = CURRENT_TIMESTAMP + RETURNING id + """, + (game_source["slug"], game_source["display_name"], game_source["provider_kind"]), + ).fetchone() + for snapshot in snapshots: + external_server_id = str(snapshot.get("external_server_id") or "").strip() + if not external_server_id: + external_server_id = _fallback_external_id(snapshot.get("server_name")) + server = connection.execute( + """ + INSERT INTO servers ( + game_source_id, external_server_id, server_name, region, + first_seen_at, last_seen_at + ) VALUES (%s, %s, %s, %s, %s, %s) + ON CONFLICT(game_source_id, external_server_id) DO UPDATE SET + server_name = EXCLUDED.server_name, + region = EXCLUDED.region, + last_seen_at = EXCLUDED.last_seen_at, + updated_at = CURRENT_TIMESTAMP + RETURNING id + """, + ( + source["id"], + external_server_id, + str(snapshot.get("server_name") or "Unknown server"), + snapshot.get("region"), + captured_at, + captured_at, + ), + ).fetchone() + connection.execute( + """ + INSERT INTO server_snapshots ( + server_id, captured_at, status, players, max_players, current_map, + source_name, snapshot_origin, source_ref, raw_payload_ref + ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, NULL) + ON CONFLICT(server_id, captured_at, source_name, source_ref) DO UPDATE SET + status = EXCLUDED.status, + players = EXCLUDED.players, + max_players = EXCLUDED.max_players, + current_map = EXCLUDED.current_map, + snapshot_origin = EXCLUDED.snapshot_origin + """, + ( + server["id"], + captured_at, + snapshot.get("status") or "unknown", + snapshot.get("players"), + snapshot.get("max_players"), + snapshot.get("current_map"), + snapshot.get("source_name") or source_name, + snapshot.get("snapshot_origin"), + snapshot.get("source_ref") or snapshot.get("source_name") or source_name, + ), + ) + persisted += 1 + return { + "db_path": "postgresql", + "captured_at": captured_at, + "persisted_snapshots": persisted, + "game_source_slug": game_source["slug"], + } + + +def upsert_player_event_rows(events: Iterable[object]) -> dict[str, int]: + initialize_postgres_display_storage() + inserted = 0 + duplicates = 0 + with connect_postgres() as connection: + for event in events: + row = connection.execute( + """ + INSERT INTO player_event_raw_ledger ( + event_id, event_type, occurred_at, server_slug, external_match_id, + source_kind, source_ref, raw_event_ref, killer_player_key, + killer_display_name, victim_player_key, victim_display_name, + weapon_name, weapon_category, kill_category, is_teamkill, event_value + ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s) + ON CONFLICT(event_id) DO NOTHING + RETURNING id + """, + ( + event.event_id, + event.event_type, + event.occurred_at, + event.server_slug, + event.external_match_id, + event.source_kind, + event.source_ref, + event.raw_event_ref, + event.killer_player_key, + event.killer_display_name, + event.victim_player_key, + event.victim_display_name, + event.weapon_name, + event.weapon_category, + event.kill_category, + bool(event.is_teamkill), + max(1, int(event.event_value)), + ), + ).fetchone() + inserted += int(bool(row)) + duplicates += int(not row) + return {"events_inserted": inserted, "duplicate_events": duplicates} + + +def list_server_snapshot_history(*, server_id: str | None = None, limit: int) -> list[dict[str, object]]: + initialize_postgres_display_storage() + where = "" + params: list[object] = [] + if server_id: + if server_id.strip().isdigit(): + where = "WHERE s.id = %s" + params.append(int(server_id)) + else: + where = "WHERE s.external_server_id = %s" + params.append(server_id.strip()) + with connect_postgres() as connection: + rows = connection.execute( + f""" + SELECT s.id AS server_id, s.external_server_id, s.server_name, s.region, + g.slug AS context, snap.source_name, snap.snapshot_origin, + snap.source_ref, snap.captured_at, snap.status, snap.players, + snap.max_players, snap.current_map + FROM server_snapshots AS snap + JOIN servers AS s ON s.id = snap.server_id + JOIN game_sources AS g ON g.id = s.game_source_id + {where} + ORDER BY snap.captured_at DESC, s.server_name ASC + LIMIT %s + """, + (*params, limit), + ).fetchall() + return [dict(row) for row in rows] + + +def list_recent_scoreboard_matches(*, server_slug: str | None, limit: int) -> list[dict[str, object]]: + initialize_postgres_display_storage() + where = "" + params: list[object] = [] + if server_slug and server_slug != ALL_SERVERS_SLUG: + where = "WHERE hs.slug = %s" + params.append(server_slug) + with connect_postgres() as connection: + rows = connection.execute( + f""" + SELECT hs.slug AS server_slug, hs.display_name AS server_name, + hm.external_match_id, hm.started_at, hm.ended_at, + hm.map_pretty_name, hm.map_name, hm.allied_score, hm.axis_score, + hm.raw_payload_ref, COUNT(stats.id) AS player_count + FROM historical_matches AS hm + JOIN historical_servers AS hs ON hs.id = hm.historical_server_id + LEFT JOIN historical_player_match_stats AS stats ON stats.historical_match_id = hm.id + {where} + GROUP BY hm.id, hs.slug, hs.display_name + ORDER BY COALESCE(hm.ended_at, hm.started_at) DESC + LIMIT %s + """, + (*params, limit), + ).fetchall() + return [_recent_match_row(row) for row in rows] + + +def get_scoreboard_match_detail(*, server_slug: str, match_id: str) -> dict[str, object] | None: + initialize_postgres_display_storage() + with connect_postgres() as connection: + row = connection.execute( + """ + SELECT hm.id AS match_pk, hs.slug AS server_slug, hs.display_name AS server_name, + hm.external_match_id, hm.started_at, hm.ended_at, hm.map_pretty_name, + hm.map_name, hm.allied_score, hm.axis_score, hm.raw_payload_ref, + COUNT(stats.id) AS player_count, + SUM(COALESCE(stats.time_seconds, 0)) AS total_time_seconds + FROM historical_matches AS hm + JOIN historical_servers AS hs ON hs.id = hm.historical_server_id + LEFT JOIN historical_player_match_stats AS stats ON stats.historical_match_id = hm.id + WHERE hs.slug = %s AND hm.external_match_id = %s + GROUP BY hm.id, hs.slug, hs.display_name + LIMIT 1 + """, + (server_slug, match_id), + ).fetchone() + if not row: + return None + players = connection.execute( + """ + SELECT hp.display_name, hp.stable_player_key, stats.team_side, stats.level, + stats.kills, stats.deaths, stats.teamkills, stats.combat, stats.offense, + stats.defense, stats.support, stats.time_seconds + FROM historical_player_match_stats AS stats + JOIN historical_players AS hp ON hp.id = stats.historical_player_id + WHERE stats.historical_match_id = %s + ORDER BY COALESCE(stats.kills, 0) DESC, hp.display_name ASC + """, + (row["match_pk"],), + ).fetchall() + started_at = row["started_at"] + ended_at = row["ended_at"] + return { + "server": {"slug": row["server_slug"], "name": row["server_name"]}, + "match_id": row["external_match_id"], + "started_at": started_at, + "ended_at": ended_at, + "closed_at": ended_at or started_at, + "duration_seconds": _duration_seconds(started_at, ended_at), + "map": {"name": row["map_name"], "pretty_name": row["map_pretty_name"] or row["map_name"]}, + "result": _match_result(row["allied_score"], row["axis_score"]), + "player_count": int(row["player_count"] or 0), + "total_time_seconds": _int(row["total_time_seconds"]), + "players": [ + { + "name": player["display_name"], + "stable_player_key": player["stable_player_key"], + "team_side": player["team_side"], + **{ + key: _int(player[key]) + for key in ( + "level", "kills", "deaths", "teamkills", "combat", + "offense", "defense", "support", "time_seconds", + ) + }, + } + for player in players + ], + "capture_basis": "public-scoreboard-match", + "match_url": resolve_trusted_scoreboard_match_url(row["raw_payload_ref"], row["server_slug"]), + } + + +def list_scoreboard_server_summaries(*, server_slug: str | None) -> list[dict[str, object]]: + initialize_postgres_display_storage() + if server_slug == ALL_SERVERS_SLUG: + rows = list_scoreboard_server_summaries(server_slug=None) + return [_all_server_summary(rows)] + where = "WHERE hs.slug = %s" if server_slug else "" + params = (server_slug,) if server_slug else () + with connect_postgres() as connection: + rows = connection.execute( + f""" + SELECT hs.slug AS server_slug, hs.display_name AS server_name, + COUNT(DISTINCT hm.id) AS matches_count, + COUNT(DISTINCT hp.id) AS unique_players, + COALESCE(SUM(stats.kills), 0) AS total_kills, + COUNT(DISTINCT COALESCE(hm.map_pretty_name, hm.map_name)) AS map_count, + MIN(COALESCE(hm.ended_at, hm.started_at, hm.created_at_source)) AS first_match_at, + MAX(COALESCE(hm.ended_at, hm.started_at, hm.created_at_source)) AS last_match_at + FROM historical_servers AS hs + LEFT JOIN historical_matches AS hm ON hm.historical_server_id = hs.id + LEFT JOIN historical_player_match_stats AS stats ON stats.historical_match_id = hm.id + LEFT JOIN historical_players AS hp ON hp.id = stats.historical_player_id + {where} + GROUP BY hs.id + ORDER BY hs.server_number ASC, hs.slug ASC + """, + params, + ).fetchall() + map_rows = connection.execute( + f""" + SELECT hs.slug AS server_slug, + COALESCE(hm.map_pretty_name, hm.map_name, 'Mapa no disponible') AS map_name, + COUNT(*) AS matches_count + FROM historical_matches AS hm + JOIN historical_servers AS hs ON hs.id = hm.historical_server_id + {where} + GROUP BY hs.slug, COALESCE(hm.map_pretty_name, hm.map_name, 'Mapa no disponible') + ORDER BY hs.slug ASC, matches_count DESC, map_name ASC + """, + params, + ).fetchall() + maps: dict[str, list[dict[str, object]]] = {} + for row in map_rows: + maps.setdefault(str(row["server_slug"]), []) + if len(maps[str(row["server_slug"])]) < 3: + maps[str(row["server_slug"])].append( + {"map_name": row["map_name"], "matches_count": int(row["matches_count"] or 0)} + ) + return [_summary_row(row, maps.get(str(row["server_slug"]), [])) for row in rows] + + +def list_scoreboard_leaderboard( + *, timeframe: str, metric: str, server_id: str | None, limit: int +) -> dict[str, object]: + current = datetime.now(timezone.utc) + if timeframe == "monthly": + current_start = current.replace(day=1, hour=0, minute=0, second=0, microsecond=0) + previous_start = (current_start - timedelta(days=1)).replace( + day=1, hour=0, minute=0, second=0, microsecond=0 + ) + label = ("current-month", "Mes actual", "previous-closed-month-fallback", "Mes cerrado anterior") + else: + current_midnight = current.replace(hour=0, minute=0, second=0, microsecond=0) + current_start = current_midnight - timedelta(days=current_midnight.weekday()) + previous_start = current_start - timedelta(days=7) + label = ("current-week", "Semana actual", "previous-closed-week-fallback", "Semana cerrada anterior") + current_count = _count_scoreboard_matches(server_id, current_start, current) + previous_count = _count_scoreboard_matches(server_id, previous_start, current_start) + fallback = current_count <= 0 and previous_count > 0 + start, end = (previous_start, current_start) if fallback else (current_start, current) + rows = _leaderboard_rows(server_id=server_id, metric=metric, start=start, end=end, limit=limit) + window_days = max(1, int(((end - start).total_seconds() + 86399) // 86400)) + result = { + "metric": metric, + "window_start": _iso(start), + "window_end": _iso(end), + "window_days": window_days, + "window_kind": label[2] if fallback else label[0], + "window_label": label[3] if fallback else label[1], + "uses_fallback": fallback, + "selection_reason": ( + "no-current-month-matches" if fallback and timeframe == "monthly" + else "insufficient-current-week-sample" if fallback + else label[0] + ), + "items": rows, + } + if timeframe == "monthly": + result.update( + { + "timeframe": "monthly", + "current_month_start": _iso(current_start), + "current_month_closed_matches": current_count, + "previous_month_closed_matches": previous_count, + "sufficient_sample": { + "minimum_closed_matches": 1, + "current_month_closed_matches": current_count, + "current_month_has_sufficient_sample": current_count > 0, + "is_early_month": current.day <= 3, + }, + } + ) + else: + result.update( + { + "current_week_start": _iso(current_start), + "current_week_closed_matches": current_count, + "previous_week_closed_matches": previous_count, + "sufficient_sample": { + "minimum_closed_matches": 1, + "current_week_closed_matches": current_count, + "current_week_has_sufficient_sample": current_count > 0, + "is_early_week": current.weekday() <= get_historical_weekly_fallback_max_weekday(), + "fallback_max_weekday": get_historical_weekly_fallback_max_weekday(), + }, + } + ) + return result + + +def table_counts() -> dict[str, int]: + initialize_postgres_display_storage() + tables = ( + "historical_matches", + "historical_player_match_stats", + "displayed_historical_snapshots", + "player_event_raw_ledger", + "server_snapshots", + ) + with connect_postgres() as connection: + return { + table: int(connection.execute(f"SELECT COUNT(*) AS count FROM {table}").fetchone()["count"] or 0) + for table in tables + } + + +def _leaderboard_rows( + *, server_id: str | None, metric: str, start: datetime, end: datetime, limit: int +) -> list[dict[str, object]]: + metric_sql = { + "kills": "COALESCE(SUM(stats.kills), 0)", + "deaths": "COALESCE(SUM(stats.deaths), 0)", + "support": "COALESCE(SUM(stats.support), 0)", + "matches_over_100_kills": ( + "COALESCE(SUM(CASE WHEN COALESCE(stats.kills, 0) >= 100 THEN 1 ELSE 0 END), 0)" + ), + }[metric] + aggregate = server_id == ALL_SERVERS_SLUG + where, server_params = _server_where(server_id) + server_slug = f"'{ALL_SERVERS_SLUG}'" if aggregate else "hs.slug" + server_name = f"'{ALL_SERVERS_DISPLAY_NAME}'" if aggregate else "hs.display_name" + partition = f"'{ALL_SERVERS_SLUG}'" if aggregate else "hs.slug" + group_by = "hp.id" if aggregate else "hs.slug, hs.display_name, hp.id" + with connect_postgres() as connection: + rows = connection.execute( + f""" + WITH ranked AS ( + SELECT {server_slug} AS server_slug, {server_name} AS server_name, + hp.stable_player_key, hp.display_name AS player_name, hp.steam_id, + COUNT(DISTINCT hm.id) AS matches_count, {metric_sql} AS metric_value, + ROW_NUMBER() OVER ( + PARTITION BY {partition} + ORDER BY {metric_sql} DESC, COUNT(DISTINCT hm.id) ASC, hp.display_name ASC + ) AS ranking_position + FROM historical_player_match_stats AS stats + JOIN historical_matches AS hm ON hm.id = stats.historical_match_id + JOIN historical_servers AS hs ON hs.id = hm.historical_server_id + JOIN historical_players AS hp ON hp.id = stats.historical_player_id + WHERE hm.ended_at IS NOT NULL AND hm.ended_at >= %s AND hm.ended_at < %s {where} + GROUP BY {group_by} + ) + SELECT * FROM ranked WHERE ranking_position <= %s + ORDER BY server_slug ASC, ranking_position ASC + """, + (_iso(start), _iso(end), *server_params, limit), + ).fetchall() + return [ + { + "server": {"slug": row["server_slug"], "name": row["server_name"]}, + "time_range": {"start": _iso(start), "end": _iso(end), "window_days": max(1, (end - start).days or 1)}, + "player": { + "stable_player_key": row["stable_player_key"], + "name": row["player_name"], + "steam_id": row["steam_id"], + }, + "metric": metric, + "ranking_position": int(row["ranking_position"]), + "metric_value": int(row["metric_value"] or 0), + "matches_considered": int(row["matches_count"] or 0), + } + for row in rows + ] + + +def _count_scoreboard_matches(server_id: str | None, start: datetime, end: datetime) -> int: + where, server_params = _server_where(server_id) + with connect_postgres() as connection: + row = connection.execute( + f""" + SELECT COUNT(DISTINCT hm.id) AS count + FROM historical_matches AS hm + JOIN historical_servers AS hs ON hs.id = hm.historical_server_id + JOIN historical_player_match_stats AS stats ON stats.historical_match_id = hm.id + WHERE hm.ended_at IS NOT NULL AND hm.ended_at >= %s AND hm.ended_at < %s {where} + """, + (_iso(start), _iso(end), *server_params), + ).fetchone() + return int(row["count"] or 0) + + +def _server_where(server_id: str | None) -> tuple[str, tuple[object, ...]]: + if not server_id or server_id == ALL_SERVERS_SLUG: + return "", () + return "AND (hs.slug = %s OR CAST(hs.server_number AS TEXT) = %s)", (server_id, server_id) + + +def _recent_match_row(row: Mapping[str, object]) -> dict[str, object]: + return { + "server": {"slug": row["server_slug"], "name": row["server_name"]}, + "match_id": row["external_match_id"], + "started_at": row["started_at"], + "ended_at": row["ended_at"], + "closed_at": row["ended_at"] or row["started_at"], + "map": {"name": row["map_name"], "pretty_name": row["map_pretty_name"] or row["map_name"]}, + "result": _match_result(row["allied_score"], row["axis_score"]), + "player_count": int(row["player_count"] or 0), + "match_url": resolve_trusted_scoreboard_match_url(row["raw_payload_ref"], row["server_slug"]), + } + + +def _summary_row(row: Mapping[str, object], top_maps: list[dict[str, object]]) -> dict[str, object]: + first = row["first_match_at"] + last = row["last_match_at"] + matches = int(row["matches_count"] or 0) + return { + "server": {"slug": row["server_slug"], "name": row["server_name"]}, + "matches_count": matches, + "imported_matches_count": matches, + "unique_players": int(row["unique_players"] or 0), + "total_kills": int(row["total_kills"] or 0), + "map_count": int(row["map_count"] or 0), + "top_maps": top_maps, + "coverage": { + "basis": "postgres-migrated-public-scoreboard", + "status": "available" if matches else "empty", + "imported_matches_count": matches, + "discovered_total_matches": None, + "first_match_at": first, + "last_match_at": last, + "coverage_days": _coverage_days(first, last), + }, + "backfill": {}, + "time_range": {"start": first, "end": last}, + } + + +def _all_server_summary(items: list[dict[str, object]]) -> dict[str, object]: + starts = [item["time_range"]["start"] for item in items if item["time_range"]["start"]] + ends = [item["time_range"]["end"] for item in items if item["time_range"]["end"]] + return { + "server": {"slug": ALL_SERVERS_SLUG, "name": ALL_SERVERS_DISPLAY_NAME}, + "matches_count": sum(int(item["matches_count"]) for item in items), + "imported_matches_count": sum(int(item["imported_matches_count"]) for item in items), + "unique_players": None, + "total_kills": sum(int(item["total_kills"]) for item in items), + "map_count": None, + "top_maps": [], + "coverage": {"basis": "postgres-migrated-public-scoreboard", "status": "available" if items else "empty"}, + "backfill": {}, + "time_range": {"start": min(starts) if starts else None, "end": max(ends) if ends else None}, + } + + +def _attach_server_history(connection: Any, item: dict[str, object]) -> dict[str, object]: + rows = connection.execute( + """ + SELECT captured_at, status, players FROM server_snapshots + WHERE server_id = %s ORDER BY captured_at DESC LIMIT %s + """, + (item["server_id"], SUMMARY_SNAPSHOT_LIMIT), + ).fetchall() + players = [int(row["players"]) for row in rows if row["players"] is not None] + online = [row for row in rows if row["status"] == "online"] + item["history_summary"] = { + "window_size": SUMMARY_SNAPSHOT_LIMIT, + "recent_capture_count": len(rows), + "recent_online_count": len(online), + "recent_average_players": round(sum(players) / len(players), 1) if players else None, + "recent_peak_players": max(players, default=None), + "last_seen_online_at": online[0]["captured_at"] if online else None, + "minutes_since_last_capture": _minutes_since(rows[0]["captured_at"]) if rows else None, + } + return item + + +def _match_result(allied: object, axis: object) -> dict[str, object]: + allied_int, axis_int = _int(allied), _int(axis) + winner = None + if allied_int is not None and axis_int is not None: + winner = "allied" if allied_int > axis_int else "axis" if axis_int > allied_int else "draw" + return {"allied_score": allied_int, "axis_score": axis_int, "winner": winner} + + +def _duration_seconds(start: object, end: object) -> int | None: + start_point, end_point = _parse_datetime(_iso(start)), _parse_datetime(_iso(end)) + return max(0, int((end_point - start_point).total_seconds())) if start_point and end_point else None + + +def _coverage_days(start: object, end: object) -> int | None: + seconds = _duration_seconds(start, end) + return max(1, int((seconds + 86399) // 86400)) if seconds is not None else None + + +def _minutes_since(value: object) -> int | None: + point = _parse_datetime(_iso(value)) + return max(0, int((datetime.now(timezone.utc) - point).total_seconds() // 60)) if point else None + + +def _int(value: object) -> int | None: + try: + return None if value is None else int(value) + except (TypeError, ValueError): + return None + + +def _fallback_external_id(value: object) -> str: + normalized = "".join( + character.lower() if character.isalnum() else "-" + for character in str(value or "unknown-server") + ) + compact = "-".join(part for part in normalized.split("-") if part) + return compact or "unknown-server" + + +def _iso(value: object) -> str | None: + if isinstance(value, datetime): + point = value if value.tzinfo else value.replace(tzinfo=timezone.utc) + return point.astimezone(timezone.utc).isoformat().replace("+00:00", "Z") + text = str(value or "").strip() + return text or None + + +def _parse_datetime(value: str | None) -> datetime | None: + if not value: + return None + try: + point = datetime.fromisoformat(value.replace("Z", "+00:00")) + except ValueError: + return None + return point.astimezone(timezone.utc) if point.tzinfo else point.replace(tzinfo=timezone.utc) + + +def _utc_now_iso() -> str: + return datetime.now(timezone.utc).isoformat().replace("+00:00", "Z") diff --git a/backend/app/postgres_rcon_storage.py b/backend/app/postgres_rcon_storage.py index 220a84a..75f307c 100644 --- a/backend/app/postgres_rcon_storage.py +++ b/backend/app/postgres_rcon_storage.py @@ -669,7 +669,6 @@ def upsert_scoreboard_candidates( def count_migrated_tables() -> dict[str, int]: - initialize_postgres_rcon_storage() table_names = ( "rcon_admin_log_events", "rcon_player_profile_snapshots", @@ -693,13 +692,11 @@ def count_migrated_tables() -> dict[str, int]: def _fetchall(sql: str, params: Iterable[object] = ()) -> list[dict[str, object]]: - initialize_postgres_rcon_storage() with connect_postgres() as connection: return [dict(row) for row in connection.execute(sql, tuple(params)).fetchall()] def _fetchone(sql: str, params: Iterable[object] = ()) -> dict[str, object] | None: - initialize_postgres_rcon_storage() with connect_postgres() as connection: row = connection.execute(sql, tuple(params)).fetchone() return dict(row) if row else None diff --git a/backend/app/rcon_admin_log_materialization.py b/backend/app/rcon_admin_log_materialization.py index 92a76b4..879983c 100644 --- a/backend/app/rcon_admin_log_materialization.py +++ b/backend/app/rcon_admin_log_materialization.py @@ -262,6 +262,16 @@ def get_materialized_rcon_match_detail( """, (match_key, server_key, server_key), ).fetchone() + if match is None and match_key.startswith(f"{server_key}:"): + match = connection.execute( + """ + SELECT * + FROM rcon_materialized_matches + WHERE match_key = ? + LIMIT 1 + """, + (match_key,), + ).fetchone() if match is None: return None stat_rows = connection.execute( diff --git a/backend/app/sqlite_to_postgres_migration.py b/backend/app/sqlite_to_postgres_migration.py new file mode 100644 index 0000000..a28d102 --- /dev/null +++ b/backend/app/sqlite_to_postgres_migration.py @@ -0,0 +1,368 @@ +"""Idempotent phase-2 migration from displayed SQLite/files into PostgreSQL.""" + +from __future__ import annotations + +import json +import sqlite3 +from collections import defaultdict +from contextlib import closing +from pathlib import Path +from typing import Any + +from .config import get_storage_path +from .postgres_display_storage import ( + connect_postgres as connect_display_postgres, + initialize_postgres_display_storage, + persist_snapshot_record, +) +from .postgres_rcon_storage import initialize_postgres_rcon_storage + + +RCON_TABLES = ( + "rcon_historical_targets", + "rcon_historical_capture_runs", + "rcon_historical_samples", + "rcon_historical_checkpoints", + "rcon_historical_competitive_windows", + "rcon_admin_log_events", + "rcon_player_profile_snapshots", + "rcon_materialized_matches", + "rcon_match_player_stats", + "rcon_scoreboard_match_candidates", +) +DISPLAY_TABLES = ( + "game_sources", + "servers", + "server_snapshots", + "historical_servers", + "historical_maps", + "historical_matches", + "historical_players", + "historical_player_match_stats", + "player_event_raw_ledger", +) +SKIP_SLUG = "comunidad-hispana-03" + + +def migrate_sqlite_to_postgres() -> dict[str, object]: + """Copy displayed legacy data to PostgreSQL without deleting legacy sources.""" + initialize_postgres_rcon_storage() + initialize_postgres_display_storage() + summary: dict[str, object] = { + "status": "ok", + "source_paths": [], + "migrated_tables": [], + "migrated_domains": [], + "rows_read": {}, + "rows_inserted": {}, + "rows_updated": {}, + "rows_skipped": {}, + "errors": [], + } + table_totals: dict[str, dict[str, int]] = defaultdict( + lambda: {"read": 0, "inserted": 0, "updated": 0, "skipped": 0} + ) + for db_path in _discover_sqlite_paths(): + summary["source_paths"].append(str(db_path)) + try: + _migrate_sqlite_path(db_path, table_totals) + except Exception as error: # noqa: BLE001 - report all source failures + summary["errors"].append({"source_path": str(db_path), "error": str(error)}) + + snapshots_root = get_storage_path().parent / "snapshots" + if snapshots_root.exists(): + summary["source_paths"].append(str(snapshots_root)) + _migrate_snapshot_files(snapshots_root, table_totals, summary["errors"]) + _sync_sequences() + summary["migrated_tables"] = sorted(table_totals) + summary["migrated_domains"] = [ + "rcon-admin-log-events", + "rcon-player-profile-snapshots", + "rcon-historical-capture-samples-and-windows", + "rcon-materialized-matches", + "rcon-materialized-player-stats", + "rcon-safe-scoreboard-candidates", + "public-scoreboard-historical-matches-and-player-stats", + "weekly-and-monthly-scoreboard-rankings", + "displayed-historical-snapshots", + "live-server-summary-cache", + "player-event-ledger", + ] + for table_name, totals in sorted(table_totals.items()): + summary["rows_read"][table_name] = totals["read"] + summary["rows_inserted"][table_name] = totals["inserted"] + summary["rows_updated"][table_name] = totals["updated"] + summary["rows_skipped"][table_name] = totals["skipped"] + summary["status"] = "ok" if not summary["errors"] else "completed-with-errors" + return summary + + +def _migrate_sqlite_path(db_path: Path, totals: dict[str, dict[str, int]]) -> None: + with closing(sqlite3.connect(db_path)) as sqlite_connection: + sqlite_connection.row_factory = sqlite3.Row + available_tables = { + row["name"] + for row in sqlite_connection.execute( + "SELECT name FROM sqlite_master WHERE type = 'table'" + ).fetchall() + } + tables = [table for table in (*RCON_TABLES, *DISPLAY_TABLES) if table in available_tables] + with connect_display_postgres() as postgres_connection: + postgres_columns = { + table: _postgres_columns(postgres_connection, table) + for table in tables + } + historical_server_ids = _legacy_server03_ids(sqlite_connection) + historical_match_ids = _legacy_match_ids(sqlite_connection, historical_server_ids) + legacy_rcon_target_ids = _legacy_rcon_target03_ids(sqlite_connection) + for table_name in tables: + _copy_table( + sqlite_connection, + postgres_connection, + table_name=table_name, + postgres_columns=postgres_columns[table_name], + totals=totals[table_name], + historical_server_ids=historical_server_ids, + historical_match_ids=historical_match_ids, + legacy_rcon_target_ids=legacy_rcon_target_ids, + ) + + +def _copy_table( + sqlite_connection: sqlite3.Connection, + postgres_connection: Any, + *, + table_name: str, + postgres_columns: list[str], + totals: dict[str, int], + historical_server_ids: set[int], + historical_match_ids: set[int], + legacy_rcon_target_ids: set[int], +) -> None: + sqlite_columns = [ + str(row["name"]) + for row in sqlite_connection.execute(f"PRAGMA table_info({table_name})").fetchall() + ] + columns = [column for column in sqlite_columns if column in postgres_columns] + if not columns: + return + rows = sqlite_connection.execute( + f"SELECT {', '.join(columns)} FROM {table_name}" + ).fetchall() + placeholders = ", ".join(["%s"] * len(columns)) + sql = ( + f"INSERT INTO {table_name} ({', '.join(columns)}) " + f"VALUES ({placeholders}) ON CONFLICT DO NOTHING" + ) + values: list[tuple[object, ...]] = [] + for row in rows: + totals["read"] += 1 + row_dict = dict(row) + if _skip_row( + table_name, + row_dict, + historical_server_ids=historical_server_ids, + historical_match_ids=historical_match_ids, + legacy_rcon_target_ids=legacy_rcon_target_ids, + ): + totals["skipped"] += 1 + continue + values.append(tuple(_postgres_value(column, row_dict[column]) for column in columns)) + with postgres_connection.cursor() as cursor: + for start in range(0, len(values), 1000): + batch = values[start : start + 1000] + cursor.executemany(sql, batch) + inserted = max(0, int(cursor.rowcount or 0)) + totals["inserted"] += inserted + totals["skipped"] += len(batch) - inserted + + +def _migrate_snapshot_files( + snapshots_root: Path, + totals: dict[str, dict[str, int]], + errors: list[object], +) -> None: + snapshot_totals = totals["displayed_historical_snapshots"] + for snapshot_path in sorted(snapshots_root.glob("*/*.json")): + snapshot_totals["read"] += 1 + try: + document = json.loads(snapshot_path.read_text(encoding="utf-8")) + if str(document.get("server_key") or "") == SKIP_SLUG: + snapshot_totals["skipped"] += 1 + continue + before = _snapshot_exists(document) + persist_snapshot_record(document) + snapshot_totals["updated" if before else "inserted"] += 1 + except Exception as error: # noqa: BLE001 - keep migrating neighboring snapshots + snapshot_totals["skipped"] += 1 + errors.append({"source_path": str(snapshot_path), "error": str(error)}) + + +def _snapshot_exists(document: dict[str, object]) -> bool: + with connect_display_postgres() as connection: + row = connection.execute( + """ + SELECT 1 FROM displayed_historical_snapshots + WHERE server_key = %s AND snapshot_type = %s AND metric = %s AND snapshot_window = %s + """, + ( + str(document.get("server_key") or ""), + str(document.get("snapshot_type") or ""), + str(document.get("metric") or ""), + str(document.get("window") or ""), + ), + ).fetchone() + return bool(row) + + +def _skip_row( + table_name: str, + row: dict[str, object], + *, + historical_server_ids: set[int], + historical_match_ids: set[int], + legacy_rcon_target_ids: set[int], +) -> bool: + if row.get("server_slug") == SKIP_SLUG or row.get("slug") == SKIP_SLUG: + return True + if row.get("external_server_id") == SKIP_SLUG or row.get("target_key") == SKIP_SLUG: + return True + if table_name == "historical_matches" and row.get("historical_server_id") in historical_server_ids: + return True + if ( + table_name == "historical_player_match_stats" + and row.get("historical_match_id") in historical_match_ids + ): + return True + if table_name == "rcon_historical_samples" and row.get("target_id") in legacy_rcon_target_ids: + return True + if table_name == "rcon_historical_checkpoints" and row.get("target_id") in legacy_rcon_target_ids: + return True + if table_name == "rcon_historical_competitive_windows" and row.get("target_id") in legacy_rcon_target_ids: + return True + return False + + +def _legacy_server03_ids(connection: sqlite3.Connection) -> set[int]: + if not _has_table(connection, "historical_servers"): + return set() + return { + int(row["id"]) + for row in connection.execute( + "SELECT id FROM historical_servers WHERE slug = ?", + (SKIP_SLUG,), + ).fetchall() + } + + +def _legacy_rcon_target03_ids(connection: sqlite3.Connection) -> set[int]: + if not _has_table(connection, "rcon_historical_targets"): + return set() + return { + int(row["id"]) + for row in connection.execute( + """ + SELECT id FROM rcon_historical_targets + WHERE external_server_id = ? OR target_key = ? + """, + (SKIP_SLUG, SKIP_SLUG), + ).fetchall() + } + + +def _legacy_match_ids(connection: sqlite3.Connection, historical_server_ids: set[int]) -> set[int]: + if not historical_server_ids or not _has_table(connection, "historical_matches"): + return set() + placeholders = ", ".join(["?"] * len(historical_server_ids)) + return { + int(row["id"]) + for row in connection.execute( + f"SELECT id FROM historical_matches WHERE historical_server_id IN ({placeholders})", + tuple(sorted(historical_server_ids)), + ).fetchall() + } + + +def _postgres_columns(connection: Any, table_name: str) -> list[str]: + rows = connection.execute( + """ + SELECT column_name + FROM information_schema.columns + WHERE table_schema = 'public' AND table_name = %s + ORDER BY ordinal_position + """, + (table_name,), + ).fetchall() + return [str(row["column_name"]) for row in rows] + + +def _sync_sequences() -> None: + tables = ( + "game_sources", + "servers", + "server_snapshots", + "historical_servers", + "historical_maps", + "historical_matches", + "historical_players", + "historical_player_match_stats", + "player_event_raw_ledger", + "rcon_historical_targets", + "rcon_historical_capture_runs", + "rcon_historical_samples", + "rcon_historical_competitive_windows", + "rcon_admin_log_events", + "rcon_player_profile_snapshots", + "rcon_materialized_matches", + "rcon_match_player_stats", + "rcon_scoreboard_match_candidates", + ) + with connect_display_postgres() as connection: + for table_name in tables: + connection.execute( + f""" + SELECT setval( + pg_get_serial_sequence(%s, 'id'), + GREATEST(COALESCE((SELECT MAX(id) FROM {table_name}), 1), 1), + TRUE + ) + """, + (table_name,), + ) + + +def _discover_sqlite_paths() -> list[Path]: + configured = get_storage_path() + candidates = {configured} + if configured.parent.exists(): + candidates.update(configured.parent.glob("*.sqlite*")) + return sorted( + path + for path in candidates + if path.exists() + and path.is_file() + and not str(path).endswith(("-shm", "-wal")) + ) + + +def _has_table(connection: sqlite3.Connection, table_name: str) -> bool: + return bool( + connection.execute( + "SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = ?", + (table_name,), + ).fetchone() + ) + + +def _postgres_value(column: str, value: object) -> object: + if column in {"is_active", "is_teamkill"}: + return bool(value) + return value + + +def main() -> None: + print(json.dumps(migrate_sqlite_to_postgres(), ensure_ascii=False, indent=2)) + + +if __name__ == "__main__": + main() diff --git a/backend/app/storage.py b/backend/app/storage.py index e99dd4e..e64c4ad 100644 --- a/backend/app/storage.py +++ b/backend/app/storage.py @@ -7,7 +7,7 @@ from datetime import datetime, timezone from pathlib import Path from typing import Iterable, Mapping -from .config import get_storage_path +from .config import get_storage_path, use_postgres_rcon_storage from .sqlite_utils import connect_sqlite_readonly, connect_sqlite_writer @@ -90,10 +90,19 @@ def persist_snapshot_batch( db_path: Path | None = None, ) -> dict[str, object]: """Persist a batch of normalized snapshots into local SQLite storage.""" - resolved_path = initialize_storage(db_path=db_path) source_definition = dict(DEFAULT_GAME_SOURCE) if game_source is not None: source_definition.update(game_source) + if use_postgres_rcon_storage(explicit_sqlite_path=db_path): + from .postgres_display_storage import persist_server_snapshots + + return persist_server_snapshots( + snapshots, + source_name=source_name, + captured_at=captured_at, + game_source=source_definition, + ) + resolved_path = initialize_storage(db_path=db_path) persisted = 0 with _connect(resolved_path) as connection: @@ -145,6 +154,10 @@ def persist_snapshot_batch( def list_latest_snapshots(*, db_path: Path | None = None) -> list[dict[str, object]]: """Return the latest persisted snapshot for each known server.""" + if use_postgres_rcon_storage(explicit_sqlite_path=db_path): + from .postgres_display_storage import list_latest_server_snapshots + + return list_latest_server_snapshots() resolved_path = resolve_storage_path(db_path=db_path) if not resolved_path.exists(): return [] @@ -190,6 +203,10 @@ def list_snapshot_history( limit: int = 20, ) -> list[dict[str, object]]: """Return recent persisted snapshots across all servers.""" + if use_postgres_rcon_storage(explicit_sqlite_path=db_path): + from .postgres_display_storage import list_server_snapshot_history + + return list_server_snapshot_history(limit=limit) resolved_path = resolve_storage_path(db_path=db_path) if not resolved_path.exists(): return [] @@ -230,6 +247,10 @@ def list_server_history( limit: int = 20, ) -> list[dict[str, object]]: """Return recent history for one server by numeric id or external id.""" + if use_postgres_rcon_storage(explicit_sqlite_path=db_path): + from .postgres_display_storage import list_server_snapshot_history + + return list_server_snapshot_history(server_id=server_id, limit=limit) resolved_path = resolve_storage_path(db_path=db_path) if not resolved_path.exists(): return [] diff --git a/backend/app/storage_diagnostics.py b/backend/app/storage_diagnostics.py index fc001ab..6309e98 100644 --- a/backend/app/storage_diagnostics.py +++ b/backend/app/storage_diagnostics.py @@ -1,4 +1,4 @@ -"""Report active phase-1 RCON storage backend and migrated table counts.""" +"""Report active PostgreSQL/displayed storage backend and migration parity counts.""" from __future__ import annotations @@ -12,7 +12,7 @@ from .rcon_admin_log_storage import initialize_rcon_admin_log_storage from .sqlite_utils import connect_sqlite_readonly -MIGRATED_TABLES = ( +MIGRATED_RCON_TABLES = ( "rcon_admin_log_events", "rcon_player_profile_snapshots", "rcon_materialized_matches", @@ -28,11 +28,14 @@ def build_storage_diagnostics() -> dict[str, object]: """Return one JSON-safe diagnostic payload for the migrated domains.""" if use_postgres_rcon_storage(): from .postgres_rcon_storage import count_migrated_tables + from .postgres_display_storage import table_counts - counts = count_migrated_tables() + rcon_counts = count_migrated_tables() + displayed_counts = table_counts() backend = "postgresql" else: - counts = _count_sqlite_tables() + rcon_counts = _count_sqlite_tables() + displayed_counts = {} backend = "sqlite-fallback" materialization = summarize_rcon_materialization_status() return { @@ -46,22 +49,63 @@ def build_storage_diagnostics() -> dict[str, object]: "rcon-materialized-matches", "rcon-materialized-player-stats", "rcon-safe-scoreboard-candidates", + "public-scoreboard-historical-matches-and-player-stats", + "weekly-rankings", + "monthly-rankings", + "displayed-historical-snapshots", + "server-summary-and-live-server-cache", + "player-event-ledger", ], - "table_counts": counts, + "table_counts": { + **rcon_counts, + **displayed_counts, + "admin_log_events": rcon_counts.get("rcon_admin_log_events", 0), + "materialized_matches": rcon_counts.get("rcon_materialized_matches", 0), + "player_stats": rcon_counts.get("rcon_match_player_stats", 0), + "public_scoreboard_historical_matches": displayed_counts.get( + "historical_matches", 0 + ), + "weekly_rankings_source_stats": displayed_counts.get( + "historical_player_match_stats", 0 + ), + "monthly_rankings_source_stats": displayed_counts.get( + "historical_player_match_stats", 0 + ), + "server_summary_cache": displayed_counts.get("displayed_historical_snapshots", 0), + "player_event_ledger": displayed_counts.get("player_event_raw_ledger", 0), + "scoreboard_candidates": rcon_counts.get("rcon_scoreboard_match_candidates", 0), + }, "latest_materialized_matches": materialization["latest_materialized_matches"], "latest_admin_log_match_end_events": materialization[ "latest_admin_log_match_end_events" ], "match_end_status": materialization["match_end_status"], - "sqlite_remaining": [ - "live server snapshots and /api/servers cache", - "public-scoreboard historical_* tables and scoreboard fallback data", - "historical snapshots, player-event ledger and Elo/MMR tables", + "remaining_sqlite_or_file_backed_domains": [ + { + "domain": "public-scoreboard ingestion run and backfill checkpoints", + "displayed_in_frontend": False, + "reason": "operational import bookkeeping is not read by visible pages", + "planned_phase": "phase-3-or-when-scoreboard-import-runs-on-postgresql", + }, + { + "domain": "Elo/MMR tables", + "displayed_in_frontend": False, + "reason": "Elo/MMR remains paused and hidden from visible pages", + "planned_phase": "phase-3", + }, ], - "scoreboard_correlation": ( - "PostgreSQL safe candidates are preferred when present; phase-1 still falls " - "back to trusted persisted historical_* scoreboard rows on SQLite." - ), + "sqlite_remaining": [ + "public-scoreboard ingestion run and backfill checkpoints", + "paused Elo/MMR tables", + ], + "scoreboard_correlation": "PostgreSQL safe candidates and migrated trusted historical match URLs are used.", + "migration_parity_summary": { + "available": backend == "postgresql", + "source_command": "python -m app.sqlite_to_postgres_migration", + "displayed_historical_storage": ( + "postgresql" if backend == "postgresql" else "sqlite-or-file-fallback" + ), + }, } @@ -69,7 +113,7 @@ def _count_sqlite_tables() -> dict[str, int]: resolved_path = initialize_rcon_admin_log_storage() counts: dict[str, int] = {} with closing(connect_sqlite_readonly(resolved_path)) as connection: - for table_name in MIGRATED_TABLES: + for table_name in MIGRATED_RCON_TABLES: try: row = connection.execute( f"SELECT COUNT(*) AS count FROM {table_name}" @@ -82,7 +126,7 @@ def _count_sqlite_tables() -> dict[str, int]: def main() -> None: - print(json.dumps(build_storage_diagnostics(), ensure_ascii=False, indent=2)) + print(json.dumps(build_storage_diagnostics(), ensure_ascii=False, indent=2, default=str)) if __name__ == "__main__": diff --git a/backend/tests/test_json_serialization.py b/backend/tests/test_json_serialization.py new file mode 100644 index 0000000..be237e3 --- /dev/null +++ b/backend/tests/test_json_serialization.py @@ -0,0 +1,27 @@ +"""Regression coverage for API JSON encoding of PostgreSQL value types.""" + +from __future__ import annotations + +import json +import unittest +from datetime import date, datetime, timezone + +from app.main import _json_default + + +class JsonSerializationTests(unittest.TestCase): + def test_json_default_serializes_postgres_datetime_and_date_values(self) -> None: + payload = { + "started_at": datetime(2026, 5, 21, 10, 11, 12, tzinfo=timezone.utc), + "day": date(2026, 5, 21), + } + + encoded = json.loads(json.dumps(payload, default=_json_default)) + + self.assertEqual( + encoded, + { + "started_at": "2026-05-21T10:11:12+00:00", + "day": "2026-05-21", + }, + ) diff --git a/backend/tests/test_rcon_materialization_pipeline.py b/backend/tests/test_rcon_materialization_pipeline.py index 9ec1143..4f1c04d 100644 --- a/backend/tests/test_rcon_materialization_pipeline.py +++ b/backend/tests/test_rcon_materialization_pipeline.py @@ -250,6 +250,29 @@ class RconMaterializationPipelineTests(unittest.TestCase): self.assertNotEqual(payload["data"]["selected_source"], "public-scoreboard") gc.collect() + def test_recent_materialized_detail_id_resolves_through_detail_read_model(self) -> None: + with tempfile.TemporaryDirectory() as tmpdir: + db_path = Path(tmpdir) / "historical.sqlite3" + previous_storage_path = os.environ.get("HLL_BACKEND_STORAGE_PATH") + os.environ["HLL_BACKEND_STORAGE_PATH"] = str(db_path) + try: + _persist_admin_log_fixture(db_path) + materialize_rcon_admin_log(db_path=db_path) + recent = list_rcon_historical_recent_activity( + server_key="comunidad-hispana-01", + limit=1, + )[0] + detail = get_rcon_historical_match_detail( + server_key="comunidad-hispana-01", + match_id=str(recent["internal_detail_match_id"]), + ) + finally: + _restore_env("HLL_BACKEND_STORAGE_PATH", previous_storage_path) + + self.assertIsNotNone(detail) + self.assertEqual(detail["match_id"], recent["internal_detail_match_id"]) + gc.collect() + def test_public_scoreboard_fallback_used_only_without_rcon_activity(self) -> None: with tempfile.TemporaryDirectory() as tmpdir: db_path = Path(tmpdir) / "historical.sqlite3" diff --git a/docs/decisions.md b/docs/decisions.md index cff44e4..293f7d6 100644 --- a/docs/decisions.md +++ b/docs/decisions.md @@ -205,3 +205,38 @@ La correlacion de URL publica en detalle usa primero candidatos PostgreSQL confiables cuando existan y puede seguir leyendo filas `historical_*` persistidas en SQLite durante la transicion. El diagnostico operativo se expone con `python -m app.storage_diagnostics`. + +## Decision 018: PostgreSQL phase 2 for displayed historical data + +PostgreSQL pasa a ser la fuente de lectura para los datos historicos visibles: + +- fallback publico `historical_*` de partidas, detalle y rankings +- snapshots historicos precalculados que consume `historico.html` +- cache live de servidores que consume `/api/servers` +- ledger player-event usado para reconstruir snapshots visibles +- tablas RCON de AdminLog, perfiles, ventanas, partidas materializadas, + estadisticas y candidatos seguros ya migradas en phase 1 + +La migracion se ejecuta de forma idempotente con: + +```powershell +cd backend +python -m app.sqlite_to_postgres_migration +python -m app.storage_diagnostics +``` + +El comando conserva IDs y `external_match_id` del scoreboard publico, claves +`match_key` materializadas y URLs seguras existentes. Copia SQLite y los JSON +historicos de `backend/data/snapshots` como fuentes legacy; no los vuelve a +usar como read model visible cuando `HLL_BACKEND_DATABASE_URL` esta definido. +Las filas legacy de `comunidad-hispana-03` se omiten en el read model visible +de esta migracion para no reactivar ese target. + +Permanecen fuera de phase 2: + +- checkpoints y runs operativos del import publico que no aparecen en frontend +- Elo/MMR pausado y oculto en la UI actual + +`app.storage_diagnostics` muestra conteos PostgreSQL, ultimas partidas +materializadas, ultimos `match_end`, dominios restantes y un resumen de paridad +para verificar la migracion antes de retirar fuentes legacy.