Merge remote-tracking branch 'origin/codex/migrate-all-displayed-historical-data-to-postgres'
This commit is contained in:
135
ai/tasks/done/TASK-POSTGRES-HISTORICAL-PHASE-2.md
Normal file
135
ai/tasks/done/TASK-POSTGRES-HISTORICAL-PHASE-2.md
Normal file
@@ -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.
|
||||||
@@ -1579,6 +1579,41 @@ Estado real a fecha de esta fase:
|
|||||||
RCON-backed primario y usar `public-scoreboard` solo como suplemento/fallback
|
RCON-backed primario y usar `public-scoreboard` solo como suplemento/fallback
|
||||||
para estadisticas por jugador sin paridad RCON
|
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
|
## Elo/MMR Monthly Ranking
|
||||||
|
|
||||||
Se añade una primera base operativa inspirada en el documento
|
Se añade una primera base operativa inspirada en el documento
|
||||||
|
|||||||
@@ -6,7 +6,7 @@ import json
|
|||||||
from datetime import datetime, timezone
|
from datetime import datetime, timezone
|
||||||
from pathlib import Path
|
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_models import HistoricalSnapshotRecord
|
||||||
from .historical_snapshots import validate_snapshot_identity
|
from .historical_snapshots import validate_snapshot_identity
|
||||||
|
|
||||||
@@ -46,6 +46,22 @@ def persist_historical_snapshot(
|
|||||||
raise ValueError("server_key is required for historical snapshots.")
|
raise ValueError("server_key is required for historical snapshots.")
|
||||||
|
|
||||||
validate_snapshot_identity(snapshot_type=snapshot_type, metric=metric)
|
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)
|
snapshots_root = initialize_historical_snapshot_storage(db_path=db_path)
|
||||||
generated_at_value = _as_utc(generated_at or datetime.now(timezone.utc))
|
generated_at_value = _as_utc(generated_at or datetime.now(timezone.utc))
|
||||||
payload_json = json.dumps(payload, ensure_ascii=True)
|
payload_json = json.dumps(payload, ensure_ascii=True)
|
||||||
@@ -148,6 +164,15 @@ def get_historical_snapshot(
|
|||||||
) -> dict[str, object] | None:
|
) -> dict[str, object] | None:
|
||||||
"""Return one persisted snapshot and decoded payload, if present."""
|
"""Return one persisted snapshot and decoded payload, if present."""
|
||||||
validate_snapshot_identity(snapshot_type=snapshot_type, metric=metric)
|
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)
|
snapshots_root = resolve_historical_snapshot_storage_path(db_path=db_path)
|
||||||
snapshot_path = _build_snapshot_path(
|
snapshot_path = _build_snapshot_path(
|
||||||
snapshots_root=snapshots_root,
|
snapshots_root=snapshots_root,
|
||||||
|
|||||||
@@ -6,7 +6,7 @@ import sqlite3
|
|||||||
from datetime import datetime, timezone
|
from datetime import datetime, timezone
|
||||||
from pathlib import Path
|
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 .data_sources import SOURCE_KIND_RCON, get_rcon_historical_read_model
|
||||||
from .historical_storage import (
|
from .historical_storage import (
|
||||||
ALL_SERVERS_SLUG,
|
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:
|
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 = sqlite3.connect(db_path)
|
||||||
connection.row_factory = sqlite3.Row
|
connection.row_factory = sqlite3.Row
|
||||||
return connection
|
return connection
|
||||||
|
|||||||
@@ -12,6 +12,7 @@ from .config import (
|
|||||||
get_historical_weekly_fallback_max_weekday,
|
get_historical_weekly_fallback_max_weekday,
|
||||||
get_historical_weekly_fallback_min_matches,
|
get_historical_weekly_fallback_min_matches,
|
||||||
get_storage_path,
|
get_storage_path,
|
||||||
|
use_postgres_rcon_storage,
|
||||||
)
|
)
|
||||||
from .historical_models import HistoricalServerDefinition
|
from .historical_models import HistoricalServerDefinition
|
||||||
from .monthly_mvp import build_monthly_mvp_rankings
|
from .monthly_mvp import build_monthly_mvp_rankings
|
||||||
@@ -739,6 +740,10 @@ def list_recent_historical_matches(
|
|||||||
db_path: Path | None = None,
|
db_path: Path | None = None,
|
||||||
) -> list[dict[str, object]]:
|
) -> list[dict[str, object]]:
|
||||||
"""Return recent persisted matches grouped for the historical API layer."""
|
"""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)
|
resolved_path = initialize_historical_storage(db_path=db_path)
|
||||||
where_clause = ""
|
where_clause = ""
|
||||||
params: list[object] = []
|
params: list[object] = []
|
||||||
@@ -821,6 +826,13 @@ def get_historical_match_detail(
|
|||||||
normalized_match_id = _stringify(match_id)
|
normalized_match_id = _stringify(match_id)
|
||||||
if not normalized_server_slug or not normalized_match_id:
|
if not normalized_server_slug or not normalized_match_id:
|
||||||
return None
|
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)
|
resolved_path = initialize_historical_storage(db_path=db_path)
|
||||||
with _connect(resolved_path) as connection:
|
with _connect(resolved_path) as connection:
|
||||||
row = connection.execute(
|
row = connection.execute(
|
||||||
@@ -939,6 +951,10 @@ def list_historical_server_summaries(
|
|||||||
db_path: Path | None = None,
|
db_path: Path | None = None,
|
||||||
) -> list[dict[str, object]]:
|
) -> list[dict[str, object]]:
|
||||||
"""Return aggregate historical metrics per server."""
|
"""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)
|
resolved_path = initialize_historical_storage(db_path=db_path)
|
||||||
if _is_all_servers_selector(server_slug):
|
if _is_all_servers_selector(server_slug):
|
||||||
return [_build_all_servers_summary(db_path=resolved_path)]
|
return [_build_all_servers_summary(db_path=resolved_path)]
|
||||||
@@ -1247,6 +1263,15 @@ def list_weekly_leaderboard(
|
|||||||
db_path: Path | None = None,
|
db_path: Path | None = None,
|
||||||
) -> dict[str, object]:
|
) -> dict[str, object]:
|
||||||
"""Return ranked weekly leaderboard totals from persisted historical match stats."""
|
"""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)
|
resolved_path = initialize_historical_storage(db_path=db_path)
|
||||||
aggregate_all_servers = _is_all_servers_selector(server_id)
|
aggregate_all_servers = _is_all_servers_selector(server_id)
|
||||||
current_time = datetime.now(timezone.utc)
|
current_time = datetime.now(timezone.utc)
|
||||||
@@ -1432,6 +1457,15 @@ def list_monthly_leaderboard(
|
|||||||
db_path: Path | None = None,
|
db_path: Path | None = None,
|
||||||
) -> dict[str, object]:
|
) -> dict[str, object]:
|
||||||
"""Return ranked monthly leaderboard totals from persisted historical match stats."""
|
"""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)
|
resolved_path = initialize_historical_storage(db_path=db_path)
|
||||||
aggregate_all_servers = _is_all_servers_selector(server_id)
|
aggregate_all_servers = _is_all_servers_selector(server_id)
|
||||||
current_time = datetime.now(timezone.utc)
|
current_time = datetime.now(timezone.utc)
|
||||||
|
|||||||
@@ -3,6 +3,7 @@
|
|||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
import json
|
import json
|
||||||
|
from datetime import date, datetime
|
||||||
from http import HTTPStatus
|
from http import HTTPStatus
|
||||||
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
||||||
|
|
||||||
@@ -38,7 +39,7 @@ class HealthHandler(BaseHTTPRequestHandler):
|
|||||||
return
|
return
|
||||||
|
|
||||||
def _write_json(self, status: HTTPStatus, payload: dict[str, object]) -> None:
|
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_response(status)
|
||||||
self._send_default_headers(content_length=len(body))
|
self._send_default_headers(content_length=len(body))
|
||||||
self.end_headers()
|
self.end_headers()
|
||||||
@@ -61,6 +62,13 @@ def create_server() -> ThreadingHTTPServer:
|
|||||||
return ThreadingHTTPServer((host, port), HealthHandler)
|
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:
|
def run() -> None:
|
||||||
"""Start the local bootstrap server."""
|
"""Start the local bootstrap server."""
|
||||||
host, port = get_bind_address()
|
host, port = get_bind_address()
|
||||||
|
|||||||
@@ -5,7 +5,7 @@ from __future__ import annotations
|
|||||||
import sqlite3
|
import sqlite3
|
||||||
from pathlib import Path
|
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
|
from .player_event_storage import initialize_player_event_storage
|
||||||
|
|
||||||
|
|
||||||
@@ -249,6 +249,10 @@ def _build_common_where(
|
|||||||
|
|
||||||
|
|
||||||
def _connect(db_path: Path) -> sqlite3.Connection:
|
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 = sqlite3.connect(db_path or get_storage_path())
|
||||||
connection.row_factory = sqlite3.Row
|
connection.row_factory = sqlite3.Row
|
||||||
return connection
|
return connection
|
||||||
|
|||||||
@@ -7,7 +7,11 @@ from collections.abc import Iterable
|
|||||||
from datetime import datetime, timedelta, timezone
|
from datetime import datetime, timedelta, timezone
|
||||||
from pathlib import Path
|
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 .player_event_models import PlayerEventRecord
|
||||||
from .sqlite_utils import connect_sqlite_writer
|
from .sqlite_utils import connect_sqlite_writer
|
||||||
|
|
||||||
@@ -95,6 +99,10 @@ def upsert_player_events(
|
|||||||
db_path: Path | None = None,
|
db_path: Path | None = None,
|
||||||
) -> dict[str, int]:
|
) -> dict[str, int]:
|
||||||
"""Insert normalized events idempotently into the raw ledger."""
|
"""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)
|
resolved_path = initialize_player_event_storage(db_path=db_path)
|
||||||
inserted = 0
|
inserted = 0
|
||||||
duplicates = 0
|
duplicates = 0
|
||||||
|
|||||||
916
backend/app/postgres_display_storage.py
Normal file
916
backend/app/postgres_display_storage.py
Normal file
@@ -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")
|
||||||
@@ -669,7 +669,6 @@ def upsert_scoreboard_candidates(
|
|||||||
|
|
||||||
|
|
||||||
def count_migrated_tables() -> dict[str, int]:
|
def count_migrated_tables() -> dict[str, int]:
|
||||||
initialize_postgres_rcon_storage()
|
|
||||||
table_names = (
|
table_names = (
|
||||||
"rcon_admin_log_events",
|
"rcon_admin_log_events",
|
||||||
"rcon_player_profile_snapshots",
|
"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]]:
|
def _fetchall(sql: str, params: Iterable[object] = ()) -> list[dict[str, object]]:
|
||||||
initialize_postgres_rcon_storage()
|
|
||||||
with connect_postgres() as connection:
|
with connect_postgres() as connection:
|
||||||
return [dict(row) for row in connection.execute(sql, tuple(params)).fetchall()]
|
return [dict(row) for row in connection.execute(sql, tuple(params)).fetchall()]
|
||||||
|
|
||||||
|
|
||||||
def _fetchone(sql: str, params: Iterable[object] = ()) -> dict[str, object] | None:
|
def _fetchone(sql: str, params: Iterable[object] = ()) -> dict[str, object] | None:
|
||||||
initialize_postgres_rcon_storage()
|
|
||||||
with connect_postgres() as connection:
|
with connect_postgres() as connection:
|
||||||
row = connection.execute(sql, tuple(params)).fetchone()
|
row = connection.execute(sql, tuple(params)).fetchone()
|
||||||
return dict(row) if row else None
|
return dict(row) if row else None
|
||||||
|
|||||||
@@ -262,6 +262,16 @@ def get_materialized_rcon_match_detail(
|
|||||||
""",
|
""",
|
||||||
(match_key, server_key, server_key),
|
(match_key, server_key, server_key),
|
||||||
).fetchone()
|
).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:
|
if match is None:
|
||||||
return None
|
return None
|
||||||
stat_rows = connection.execute(
|
stat_rows = connection.execute(
|
||||||
|
|||||||
368
backend/app/sqlite_to_postgres_migration.py
Normal file
368
backend/app/sqlite_to_postgres_migration.py
Normal file
@@ -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()
|
||||||
@@ -7,7 +7,7 @@ from datetime import datetime, timezone
|
|||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
from typing import Iterable, Mapping
|
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
|
from .sqlite_utils import connect_sqlite_readonly, connect_sqlite_writer
|
||||||
|
|
||||||
|
|
||||||
@@ -90,10 +90,19 @@ def persist_snapshot_batch(
|
|||||||
db_path: Path | None = None,
|
db_path: Path | None = None,
|
||||||
) -> dict[str, object]:
|
) -> dict[str, object]:
|
||||||
"""Persist a batch of normalized snapshots into local SQLite storage."""
|
"""Persist a batch of normalized snapshots into local SQLite storage."""
|
||||||
resolved_path = initialize_storage(db_path=db_path)
|
|
||||||
source_definition = dict(DEFAULT_GAME_SOURCE)
|
source_definition = dict(DEFAULT_GAME_SOURCE)
|
||||||
if game_source is not None:
|
if game_source is not None:
|
||||||
source_definition.update(game_source)
|
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
|
persisted = 0
|
||||||
with _connect(resolved_path) as connection:
|
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]]:
|
def list_latest_snapshots(*, db_path: Path | None = None) -> list[dict[str, object]]:
|
||||||
"""Return the latest persisted snapshot for each known server."""
|
"""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)
|
resolved_path = resolve_storage_path(db_path=db_path)
|
||||||
if not resolved_path.exists():
|
if not resolved_path.exists():
|
||||||
return []
|
return []
|
||||||
@@ -190,6 +203,10 @@ def list_snapshot_history(
|
|||||||
limit: int = 20,
|
limit: int = 20,
|
||||||
) -> list[dict[str, object]]:
|
) -> list[dict[str, object]]:
|
||||||
"""Return recent persisted snapshots across all servers."""
|
"""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)
|
resolved_path = resolve_storage_path(db_path=db_path)
|
||||||
if not resolved_path.exists():
|
if not resolved_path.exists():
|
||||||
return []
|
return []
|
||||||
@@ -230,6 +247,10 @@ def list_server_history(
|
|||||||
limit: int = 20,
|
limit: int = 20,
|
||||||
) -> list[dict[str, object]]:
|
) -> list[dict[str, object]]:
|
||||||
"""Return recent history for one server by numeric id or external id."""
|
"""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)
|
resolved_path = resolve_storage_path(db_path=db_path)
|
||||||
if not resolved_path.exists():
|
if not resolved_path.exists():
|
||||||
return []
|
return []
|
||||||
|
|||||||
@@ -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
|
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
|
from .sqlite_utils import connect_sqlite_readonly
|
||||||
|
|
||||||
|
|
||||||
MIGRATED_TABLES = (
|
MIGRATED_RCON_TABLES = (
|
||||||
"rcon_admin_log_events",
|
"rcon_admin_log_events",
|
||||||
"rcon_player_profile_snapshots",
|
"rcon_player_profile_snapshots",
|
||||||
"rcon_materialized_matches",
|
"rcon_materialized_matches",
|
||||||
@@ -28,11 +28,14 @@ def build_storage_diagnostics() -> dict[str, object]:
|
|||||||
"""Return one JSON-safe diagnostic payload for the migrated domains."""
|
"""Return one JSON-safe diagnostic payload for the migrated domains."""
|
||||||
if use_postgres_rcon_storage():
|
if use_postgres_rcon_storage():
|
||||||
from .postgres_rcon_storage import count_migrated_tables
|
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"
|
backend = "postgresql"
|
||||||
else:
|
else:
|
||||||
counts = _count_sqlite_tables()
|
rcon_counts = _count_sqlite_tables()
|
||||||
|
displayed_counts = {}
|
||||||
backend = "sqlite-fallback"
|
backend = "sqlite-fallback"
|
||||||
materialization = summarize_rcon_materialization_status()
|
materialization = summarize_rcon_materialization_status()
|
||||||
return {
|
return {
|
||||||
@@ -46,22 +49,63 @@ def build_storage_diagnostics() -> dict[str, object]:
|
|||||||
"rcon-materialized-matches",
|
"rcon-materialized-matches",
|
||||||
"rcon-materialized-player-stats",
|
"rcon-materialized-player-stats",
|
||||||
"rcon-safe-scoreboard-candidates",
|
"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_materialized_matches": materialization["latest_materialized_matches"],
|
||||||
"latest_admin_log_match_end_events": materialization[
|
"latest_admin_log_match_end_events": materialization[
|
||||||
"latest_admin_log_match_end_events"
|
"latest_admin_log_match_end_events"
|
||||||
],
|
],
|
||||||
"match_end_status": materialization["match_end_status"],
|
"match_end_status": materialization["match_end_status"],
|
||||||
"sqlite_remaining": [
|
"remaining_sqlite_or_file_backed_domains": [
|
||||||
"live server snapshots and /api/servers cache",
|
{
|
||||||
"public-scoreboard historical_* tables and scoreboard fallback data",
|
"domain": "public-scoreboard ingestion run and backfill checkpoints",
|
||||||
"historical snapshots, player-event ledger and Elo/MMR tables",
|
"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": (
|
"sqlite_remaining": [
|
||||||
"PostgreSQL safe candidates are preferred when present; phase-1 still falls "
|
"public-scoreboard ingestion run and backfill checkpoints",
|
||||||
"back to trusted persisted historical_* scoreboard rows on SQLite."
|
"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()
|
resolved_path = initialize_rcon_admin_log_storage()
|
||||||
counts: dict[str, int] = {}
|
counts: dict[str, int] = {}
|
||||||
with closing(connect_sqlite_readonly(resolved_path)) as connection:
|
with closing(connect_sqlite_readonly(resolved_path)) as connection:
|
||||||
for table_name in MIGRATED_TABLES:
|
for table_name in MIGRATED_RCON_TABLES:
|
||||||
try:
|
try:
|
||||||
row = connection.execute(
|
row = connection.execute(
|
||||||
f"SELECT COUNT(*) AS count FROM {table_name}"
|
f"SELECT COUNT(*) AS count FROM {table_name}"
|
||||||
@@ -82,7 +126,7 @@ def _count_sqlite_tables() -> dict[str, int]:
|
|||||||
|
|
||||||
|
|
||||||
def main() -> None:
|
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__":
|
if __name__ == "__main__":
|
||||||
|
|||||||
27
backend/tests/test_json_serialization.py
Normal file
27
backend/tests/test_json_serialization.py
Normal file
@@ -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",
|
||||||
|
},
|
||||||
|
)
|
||||||
@@ -250,6 +250,29 @@ class RconMaterializationPipelineTests(unittest.TestCase):
|
|||||||
self.assertNotEqual(payload["data"]["selected_source"], "public-scoreboard")
|
self.assertNotEqual(payload["data"]["selected_source"], "public-scoreboard")
|
||||||
gc.collect()
|
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:
|
def test_public_scoreboard_fallback_used_only_without_rcon_activity(self) -> None:
|
||||||
with tempfile.TemporaryDirectory() as tmpdir:
|
with tempfile.TemporaryDirectory() as tmpdir:
|
||||||
db_path = Path(tmpdir) / "historical.sqlite3"
|
db_path = Path(tmpdir) / "historical.sqlite3"
|
||||||
|
|||||||
@@ -205,3 +205,38 @@ La correlacion de URL publica en detalle usa primero candidatos PostgreSQL
|
|||||||
confiables cuando existan y puede seguir leyendo filas `historical_*`
|
confiables cuando existan y puede seguir leyendo filas `historical_*`
|
||||||
persistidas en SQLite durante la transicion. El diagnostico operativo se expone
|
persistidas en SQLite durante la transicion. El diagnostico operativo se expone
|
||||||
con `python -m app.storage_diagnostics`.
|
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.
|
||||||
|
|||||||
Reference in New Issue
Block a user