Add player event V2 pipeline foundation

ai/architecture-index.md

@@ -72,5 +72,6 @@ Community website repository with a static landing in the current phase and a pl
 - The persisted historical domain model for CRCON matches, players and ingestion runs is documented in `docs/historical-domain-model.md`.
 - The V1 monthly MVP scoring proposal for persisted historical player metrics is documented in `docs/monthly-mvp-ranking-scoring-design.md`.
 - The audited boundary between direct live RCON and future event-driven RCON metrics is documented in `docs/rcon-data-capability-audit.md`.
+- The first V2 player-event foundation now lives in dedicated `player_event_*` backend modules and starts from CRCON match-detail summaries, not from live RCON.
 - Frontend data consumption should remain progressive, endpoint by endpoint, with static fallbacks preserved during migration.
 - The frontend integration strategy is documented in `docs/frontend-data-consumption-plan.md`.

ai/tasks/done/TASK-077-player-kill-event-source-adapter.md

@@ -0,0 +1,77 @@
+# TASK-077-player-kill-event-source-adapter
+
+## Goal
+Implementar una primera capa/adaptador de fuente de eventos centrada en eventos de kill/death para preparar la V2 del sistema de métricas de jugador.
+
+## Context
+Las auditorías previas concluyen que el cliente RCON live actual no basta por sí solo para obtener directamente todos los agregados avanzados visibles en ecosistemas tipo CRCON/HLL Records. Para construir métricas como:
+- killer -> victim
+- most_killed
+- death_by
+- kills por arma
+- teamkills por evento
+
+hace falta una fuente de eventos o una adaptación técnica que permita modelar esos datos en bruto antes de agregarlos.
+
+La V2 debe empezar por una primera capa mínima centrada en eventos de kill/death, sin abarcar todavía todo el resto de acciones tácticas.
+
+## Steps
+1. Revisar las auditorías y documentos de diseño V2:
+   - docs/rcon-data-capability-audit.md
+   - docs/crcon-advanced-metrics-origin-audit.md
+   - docs/player-event-pipeline-v2-design.md
+2. Identificar la fuente o punto técnico más viable dentro del proyecto actual para empezar a capturar eventos de kill/death.
+3. Diseñar e implementar un adaptador/fuente mínima para eventos de jugador que, como mínimo, pueda producir un formato común con campos base como:
+   - server
+   - match
+   - timestamp
+   - killer
+   - victim
+   - weapon si existe
+   - teamkill si existe
+4. Mantener esta capa desacoplada de la persistencia final.
+5. Documentar claramente:
+   - qué datos se capturan realmente en esta fase
+   - qué datos siguen fuera de alcance
+6. No implementar todavía snapshots/UI del MVP V2.
+7. No romper live RCON ni el histórico actual.
+8. Al completar la implementación:
+   - dejar el repositorio consistente
+   - hacer commit
+   - hacer push al remoto si el entorno lo permite
+
+## Files to Read First
+- AGENTS.md
+- ai/repo-context.md
+- ai/architecture-index.md
+- docs/rcon-data-capability-audit.md
+- docs/crcon-advanced-metrics-origin-audit.md
+- docs/player-event-pipeline-v2-design.md
+- backend/README.md
+- backend/app/rcon_client.py
+- backend/app/providers/rcon_provider.py
+- backend/app/data_sources.py
+
+## Expected Files to Modify
+- backend/README.md
+- opcionalmente nuevos módulos, por ejemplo:
+  - backend/app/player_event_source.py
+  - backend/app/providers/player_event_source_provider.py
+  - backend/app/event_sources/*.py
+- opcionalmente ai/architecture-index.md si conviene enlazar la nueva pieza
+
+## Constraints
+- No implementar todavía agregados finales ni UI.
+- No romper el proveedor RCON live actual.
+- No hacer cambios destructivos.
+- Mantener el trabajo centrado en una fuente mínima de eventos kill/death.
+
+## Validation
+- Existe una capa/adaptador clara para eventos de kill/death.
+- Produce un formato mínimo reutilizable por la persistencia posterior.
+- Queda documentado qué cubre y qué no cubre todavía.
+- Los cambios quedan committeados y se hace push si el entorno lo permite.
+
+## Change Budget
+- Preferir menos de 6 archivos modificados o creados.
+- Preferir menos de 240 líneas cambiadas.

ai/tasks/done/TASK-078-player-event-raw-ledger-storage.md

@@ -0,0 +1,65 @@
+# TASK-078-player-event-raw-ledger-storage
+
+## Goal
+Crear la persistencia raw mínima para eventos de jugador V2, permitiendo almacenar de forma fiable eventos de kill/death con claves estables, deduplicación y base para agregados posteriores.
+
+## Context
+Una vez exista una fuente/adaptador de eventos, hace falta una capa de persistencia en bruto. Esta capa debe ser lo bastante simple para operar ya, pero lo bastante sólida para soportar luego agregados como:
+- most_killed
+- death_by
+- kills por arma
+- teamkills por jugador
+- killer/victim mensual
+
+## Steps
+1. Revisar el diseño V2 del pipeline de eventos.
+2. Diseñar una tabla o conjunto mínimo de tablas para el ledger raw de eventos de jugador.
+3. Incluir como base, si la fuente lo permite:
+   - server key
+   - match reference
+   - event timestamp
+   - killer identity
+   - victim identity
+   - weapon
+   - teamkill flag
+   - event source reference o clave de deduplicación
+4. Implementar inicialización segura del storage de eventos.
+5. Implementar inserción idempotente o deduplicada.
+6. Mantener separada esta capa del histórico actual `historical_*`.
+7. Documentar la estructura y su propósito.
+8. No implementar todavía agregados finales ni UI.
+9. Al completar la implementación:
+   - dejar el repositorio consistente
+   - hacer commit
+   - hacer push al remoto si el entorno lo permite
+
+## Files to Read First
+- AGENTS.md
+- docs/player-event-pipeline-v2-design.md
+- backend/README.md
+- backend/app/historical_models.py
+- backend/app/historical_storage.py
+- cualquier módulo nuevo de la task anterior relacionado con eventos
+
+## Expected Files to Modify
+- backend/README.md
+- opcionalmente nuevos módulos, por ejemplo:
+  - backend/app/player_event_storage.py
+  - backend/app/player_event_models.py
+- opcionalmente docs/decisions.md si conviene fijar una decisión de persistencia
+
+## Constraints
+- No romper el histórico actual.
+- No mezclar esta persistencia con snapshots ni UI.
+- No hacer cambios destructivos.
+- Mantener el trabajo centrado en el ledger raw de eventos.
+
+## Validation
+- Existe una persistencia raw mínima para eventos de jugador.
+- Soporta deduplicación o idempotencia razonable.
+- Queda separada del histórico actual.
+- Los cambios quedan committeados y se hace push si el entorno lo permite.
+
+## Change Budget
+- Preferir menos de 6 archivos modificados o creados.
+- Preferir menos de 240 líneas cambiadas.

ai/tasks/done/TASK-079-player-event-ingestion-worker.md

@@ -0,0 +1,57 @@
+# TASK-079-player-event-ingestion-worker
+
+## Goal
+Añadir un worker/proceso incremental para ingerir eventos de jugador V2 de forma segura, reanudable y compatible con la operación actual del proyecto.
+
+## Context
+Con una fuente de eventos y un ledger raw, hace falta un proceso de ingestión incremental que capture eventos sin depender del request path del usuario y sin romper el pipeline actual de histórico y snapshots.
+
+## Steps
+1. Revisar la nueva capa de fuente de eventos y la persistencia raw.
+2. Diseñar un worker incremental específico para eventos de jugador.
+3. Añadir soporte para:
+   - ejecución manual
+   - reintentos básicos
+   - checkpoint o mecanismo de reanudación si aplica
+   - deduplicación segura
+4. Mantenerlo desacoplado del `historical-runner` actual si eso reduce riesgo operacional.
+5. Documentar cómo se ejecuta y cuál es su alcance.
+6. No implementar todavía snapshots finales ni UI del MVP V2.
+7. No romper el stack Docker actual ni el flujo histórico existente.
+8. Al completar la implementación:
+   - dejar el repositorio consistente
+   - hacer commit
+   - hacer push al remoto si el entorno lo permite
+
+## Files to Read First
+- AGENTS.md
+- docs/player-event-pipeline-v2-design.md
+- backend/README.md
+- backend/app/historical_runner.py
+- backend/app/config.py
+- módulos de eventos añadidos en tasks previas
+
+## Expected Files to Modify
+- backend/app/config.py
+- backend/README.md
+- opcionalmente docker-compose.yml si hace falta dejar preparado el worker
+- opcionalmente nuevos módulos, por ejemplo:
+  - backend/app/player_event_worker.py
+  - backend/app/player_event_ingestion.py
+
+## Constraints
+- No romper el runner histórico actual.
+- No meter todavía lógica de scoring V2.
+- No tocar producto visible.
+- No hacer cambios destructivos.
+- Mantener el trabajo centrado en ingestión incremental de eventos.
+
+## Validation
+- Existe un worker o flujo incremental para eventos de jugador.
+- Puede ejecutarse de forma segura y reanudable.
+- No rompe el resto del sistema.
+- Los cambios quedan committeados y se hace push si el entorno lo permite.
+
+## Change Budget
+- Preferir menos de 6 archivos modificados o creados.
+- Preferir menos de 240 líneas cambiadas.

ai/tasks/done/TASK-080-player-event-derived-aggregates.md

@@ -0,0 +1,65 @@
+# TASK-080-player-event-derived-aggregates
+
+## Goal
+Construir los primeros agregados derivados V2 a partir del ledger raw de eventos de jugador, centrados en duelos y armas, para preparar una futura V2 del MVP.
+
+## Context
+Con eventos raw ya capturados, el siguiente paso es derivar métricas útiles y reutilizables. Las primeras más valiosas para la V2 del MVP son:
+- most_killed
+- death_by
+- killer/victim
+- kills por arma
+- teamkills por jugador
+
+## Steps
+1. Revisar la estructura del ledger raw y el diseño V2.
+2. Diseñar e implementar agregados básicos por:
+   - partida
+   - mes
+   - servidor
+   - all-servers cuando aplique
+3. Incluir al menos:
+   - most_killed
+   - death_by
+   - net duel summaries
+   - kills por arma
+   - teamkills derivados por jugador
+4. Mantener los agregados separados de la UI y del MVP V1.
+5. Documentar qué métricas derivadas quedan ya disponibles para una futura V2 del score.
+6. No implementar todavía la fórmula final del MVP V2 ni su UI.
+7. No hacer cambios destructivos.
+8. Al completar la implementación:
+   - dejar el repositorio consistente
+   - hacer commit
+   - hacer push al remoto si el entorno lo permite
+
+## Files to Read First
+- AGENTS.md
+- docs/player-event-pipeline-v2-design.md
+- docs/crcon-advanced-metrics-origin-audit.md
+- backend/README.md
+- módulos de eventos raw ya implementados
+- backend/app/historical_storage.py
+- docs/monthly-mvp-ranking-scoring-design.md
+
+## Expected Files to Modify
+- backend/README.md
+- opcionalmente nuevos módulos, por ejemplo:
+  - backend/app/player_event_aggregates.py
+  - backend/app/player_event_queries.py
+- opcionalmente docs/decisions.md si hace falta fijar alguna decisión de agregación
+
+## Constraints
+- No tocar todavía la UI del MVP V2.
+- No romper el MVP V1.
+- No hacer cambios destructivos.
+- Mantener el trabajo centrado en agregados base de duelos y armas.
+
+## Validation
+- Existen agregados derivados básicos reutilizables para V2.
+- Queda clara su disponibilidad por servidor/mes/all-servers cuando aplique.
+- Los cambios quedan committeados y se hace push si el entorno lo permite.
+
+## Change Budget
+- Preferir menos de 6 archivos modificados o creados.
+- Preferir menos de 240 líneas cambiadas.

backend/README.md

@@ -998,6 +998,91 @@ sin tocar endpoints ni payloads.
 
 Esta separacion mantiene el backend simple y deja una base clara para futuras tasks sin introducir integraciones reales todavia.
 
+## Fuente y ledger de eventos de jugador V2
+
+La repo incluye ahora una primera base V2 separada del historico `historical_*`
+para preparar metricas avanzadas de duelos, armas y teamkills sin tocar todavia
+la UI ni el scoring final.
+
+Fuente minima elegida en esta fase:
+
+- detalle de partida `GET /api/get_map_scoreboard?map_id={id}` del scoreboard CRCON
+
+Importante:
+
+- esta fuente no es un feed raw por kill
+- el adaptador actual normaliza solo senales parciales ya visibles en el
+  resumen de partida:
+  - `most_killed`
+  - `death_by`
+  - `weapons`
+  - `death_by_weapons`
+  - `teamkills`
+- `occurred_at` usa el timestamp de cierre o inicio de la partida, no el
+  instante exacto del kill
+- el ledger raw es append-only y deduplica por `event_id`
+- la persistencia queda separada de `historical_matches` y
+  `historical_player_match_stats` aunque comparte el mismo SQLite de desarrollo
+
+Contrato minimo normalizado por evento:
+
+- `event_id`
+- `event_type`
+- `occurred_at`
+- `server_slug`
+- `external_match_id`
+- `source_kind`
+- `source_ref`
+- `killer_player_key`
+- `victim_player_key`
+- `weapon_name`
+- `kill_category`
+- `is_teamkill`
+- `event_value`
+
+Tablas nuevas:
+
+- `player_event_raw_ledger`
+- `player_event_ingestion_runs`
+- `player_event_backfill_progress`
+
+Comandos manuales desde `backend/`:
+
+```powershell
+python -m app.player_event_worker refresh
+python -m app.player_event_worker refresh --server comunidad-hispana-01 --max-pages 1
+python -m app.player_event_worker loop --interval 1800
+```
+
+Variables opcionales del worker:
+
+- `HLL_PLAYER_EVENT_REFRESH_INTERVAL_SECONDS`
+- `HLL_PLAYER_EVENT_REFRESH_MAX_RETRIES`
+- `HLL_PLAYER_EVENT_REFRESH_RETRY_DELAY_SECONDS`
+
+Politica operativa minima:
+
+- el worker corre fuera del request path HTTP
+- reusa la capa historica `public-scoreboard` solo como fuente de detalle
+- persiste checkpoints por servidor y pagina
+- la reejecucion es segura porque el ledger usa insercion idempotente por
+  `event_id`
+
+Agregados V2 ya disponibles desde codigo:
+
+- `list_most_killed()`
+- `list_death_by()`
+- `list_net_duel_summaries()`
+- `list_weapon_kills()`
+- `list_teamkill_summaries()`
+
+Limitaciones actuales de esta fase:
+
+- no existe todavia un ledger raw por kill individual
+- los agregados de duelos y armas son parciales, porque dependen del mejor
+  resumen disponible por jugador en CRCON y no de todos los encounters del match
+- la V2 no expone aun endpoints HTTP ni snapshots propios
+
 ## Alcance
 
 Esta fase no implementa:

backend/app/config.py

@@ -25,6 +25,9 @@ DEFAULT_HISTORICAL_REFRESH_RETRY_DELAY_SECONDS = 30
 DEFAULT_HISTORICAL_FULL_SNAPSHOT_EVERY_RUNS = 4
 DEFAULT_HISTORICAL_WEEKLY_FALLBACK_MIN_MATCHES = 3
 DEFAULT_HISTORICAL_WEEKLY_FALLBACK_MAX_WEEKDAY = 2
+DEFAULT_PLAYER_EVENT_REFRESH_INTERVAL_SECONDS = 1800
+DEFAULT_PLAYER_EVENT_REFRESH_MAX_RETRIES = 2
+DEFAULT_PLAYER_EVENT_REFRESH_RETRY_DELAY_SECONDS = 30
 DEFAULT_ALLOWED_ORIGINS = (
     "null",
     "http://127.0.0.1:5500",
@@ -269,6 +272,44 @@ def get_historical_weekly_fallback_max_weekday() -> int:
     return max_weekday
 
 
+def get_player_event_refresh_interval_seconds() -> int:
+    """Return the default interval used by the player event refresh loop."""
+    configured_value = os.getenv(
+        "HLL_PLAYER_EVENT_REFRESH_INTERVAL_SECONDS",
+        str(DEFAULT_PLAYER_EVENT_REFRESH_INTERVAL_SECONDS),
+    )
+    interval_seconds = int(configured_value)
+    if interval_seconds <= 0:
+        raise ValueError("HLL_PLAYER_EVENT_REFRESH_INTERVAL_SECONDS must be positive.")
+    return interval_seconds
+
+
+def get_player_event_refresh_max_retries() -> int:
+    """Return the retry count used by the player event refresh loop."""
+    configured_value = os.getenv(
+        "HLL_PLAYER_EVENT_REFRESH_MAX_RETRIES",
+        str(DEFAULT_PLAYER_EVENT_REFRESH_MAX_RETRIES),
+    )
+    max_retries = int(configured_value)
+    if max_retries < 0:
+        raise ValueError("HLL_PLAYER_EVENT_REFRESH_MAX_RETRIES must be zero or positive.")
+    return max_retries
+
+
+def get_player_event_refresh_retry_delay_seconds() -> int:
+    """Return the wait time between player event refresh retries."""
+    configured_value = os.getenv(
+        "HLL_PLAYER_EVENT_REFRESH_RETRY_DELAY_SECONDS",
+        str(DEFAULT_PLAYER_EVENT_REFRESH_RETRY_DELAY_SECONDS),
+    )
+    retry_delay_seconds = int(configured_value)
+    if retry_delay_seconds < 0:
+        raise ValueError(
+            "HLL_PLAYER_EVENT_REFRESH_RETRY_DELAY_SECONDS must be zero or positive."
+        )
+    return retry_delay_seconds
+
+
 def get_a2s_targets_payload() -> str | None:
     """Return the optional JSON payload that overrides local A2S targets."""
     raw_payload = os.getenv(DEFAULT_A2S_TARGETS_ENV_VAR)

backend/app/player_event_aggregates.py

@@ -0,0 +1,254 @@
+"""Derived duel and weapon aggregates computed from the raw player event ledger."""
+
+from __future__ import annotations
+
+import sqlite3
+from pathlib import Path
+
+from .config import get_storage_path
+from .player_event_storage import initialize_player_event_storage
+
+
+def list_most_killed(
+    *,
+    server_slug: str | None = None,
+    month: str | None = None,
+    external_match_id: str | None = None,
+    limit: int = 10,
+    db_path: Path | None = None,
+) -> list[dict[str, object]]:
+    """Return strongest killer -> victim summaries from the raw ledger."""
+    return _query_pair_summary(
+        event_type="player_kill_summary",
+        server_slug=server_slug,
+        month=month,
+        external_match_id=external_match_id,
+        limit=limit,
+        db_path=db_path,
+    )
+
+
+def list_death_by(
+    *,
+    server_slug: str | None = None,
+    month: str | None = None,
+    external_match_id: str | None = None,
+    limit: int = 10,
+    db_path: Path | None = None,
+) -> list[dict[str, object]]:
+    """Return strongest killer -> victim summaries from the victim perspective."""
+    return _query_pair_summary(
+        event_type="player_death_summary",
+        server_slug=server_slug,
+        month=month,
+        external_match_id=external_match_id,
+        limit=limit,
+        db_path=db_path,
+    )
+
+
+def list_net_duel_summaries(
+    *,
+    server_slug: str | None = None,
+    month: str | None = None,
+    external_match_id: str | None = None,
+    limit: int = 10,
+    db_path: Path | None = None,
+) -> list[dict[str, object]]:
+    """Return partial net duel summaries using the strongest encounter signals available."""
+    resolved_path = initialize_player_event_storage(db_path=db_path)
+    where_sql, params = _build_common_where(
+        event_type="player_kill_summary",
+        server_slug=server_slug,
+        month=month,
+        external_match_id=external_match_id,
+    )
+    with _connect(resolved_path) as connection:
+        rows = connection.execute(
+            f"""
+            WITH duel_pairs AS (
+                SELECT
+                    CASE
+                        WHEN COALESCE(killer_player_key, '') <= COALESCE(victim_player_key, '')
+                        THEN killer_player_key
+                        ELSE victim_player_key
+                    END AS player_a_key,
+                    CASE
+                        WHEN COALESCE(killer_player_key, '') <= COALESCE(victim_player_key, '')
+                        THEN killer_display_name
+                        ELSE victim_display_name
+                    END AS player_a_name,
+                    CASE
+                        WHEN COALESCE(killer_player_key, '') <= COALESCE(victim_player_key, '')
+                        THEN victim_player_key
+                        ELSE killer_player_key
+                    END AS player_b_key,
+                    CASE
+                        WHEN COALESCE(killer_player_key, '') <= COALESCE(victim_player_key, '')
+                        THEN victim_display_name
+                        ELSE killer_display_name
+                    END AS player_b_name,
+                    CASE
+                        WHEN COALESCE(killer_player_key, '') <= COALESCE(victim_player_key, '')
+                        THEN event_value
+                        ELSE -event_value
+                    END AS net_value,
+                    event_value
+                FROM player_event_raw_ledger
+                WHERE {where_sql}
+                    AND killer_player_key IS NOT NULL
+                    AND victim_player_key IS NOT NULL
+            )
+            SELECT
+                player_a_key,
+                player_a_name,
+                player_b_key,
+                player_b_name,
+                COALESCE(SUM(event_value), 0) AS total_encounters,
+                COALESCE(SUM(net_value), 0) AS net_duel_value
+            FROM duel_pairs
+            GROUP BY player_a_key, player_a_name, player_b_key, player_b_name
+            ORDER BY ABS(net_duel_value) DESC, total_encounters DESC, player_a_name ASC, player_b_name ASC
+            LIMIT ?
+            """,
+            [*params, limit],
+        ).fetchall()
+    return [dict(row) for row in rows]
+
+
+def list_weapon_kills(
+    *,
+    server_slug: str | None = None,
+    month: str | None = None,
+    external_match_id: str | None = None,
+    limit: int = 10,
+    db_path: Path | None = None,
+) -> list[dict[str, object]]:
+    """Return partial weapon summaries derived from top kill events."""
+    resolved_path = initialize_player_event_storage(db_path=db_path)
+    where_sql, params = _build_common_where(
+        event_type="player_kill_summary",
+        server_slug=server_slug,
+        month=month,
+        external_match_id=external_match_id,
+    )
+    with _connect(resolved_path) as connection:
+        rows = connection.execute(
+            f"""
+            SELECT
+                killer_player_key,
+                killer_display_name,
+                COALESCE(weapon_name, 'unknown') AS weapon_name,
+                COALESCE(SUM(event_value), 0) AS total_kills
+            FROM player_event_raw_ledger
+            WHERE {where_sql}
+                AND killer_player_key IS NOT NULL
+            GROUP BY killer_player_key, killer_display_name, COALESCE(weapon_name, 'unknown')
+            ORDER BY total_kills DESC, killer_display_name ASC, weapon_name ASC
+            LIMIT ?
+            """,
+            [*params, limit],
+        ).fetchall()
+    return [dict(row) for row in rows]
+
+
+def list_teamkill_summaries(
+    *,
+    server_slug: str | None = None,
+    month: str | None = None,
+    external_match_id: str | None = None,
+    limit: int = 10,
+    db_path: Path | None = None,
+) -> list[dict[str, object]]:
+    """Return derived teamkill totals per player from the raw ledger."""
+    resolved_path = initialize_player_event_storage(db_path=db_path)
+    where_sql, params = _build_common_where(
+        event_type="player_teamkill_summary",
+        server_slug=server_slug,
+        month=month,
+        external_match_id=external_match_id,
+    )
+    with _connect(resolved_path) as connection:
+        rows = connection.execute(
+            f"""
+            SELECT
+                killer_player_key,
+                killer_display_name,
+                COALESCE(SUM(event_value), 0) AS total_teamkills
+            FROM player_event_raw_ledger
+            WHERE {where_sql}
+                AND killer_player_key IS NOT NULL
+            GROUP BY killer_player_key, killer_display_name
+            ORDER BY total_teamkills DESC, killer_display_name ASC
+            LIMIT ?
+            """,
+            [*params, limit],
+        ).fetchall()
+    return [dict(row) for row in rows]
+
+
+def _query_pair_summary(
+    *,
+    event_type: str,
+    server_slug: str | None,
+    month: str | None,
+    external_match_id: str | None,
+    limit: int,
+    db_path: Path | None,
+) -> list[dict[str, object]]:
+    resolved_path = initialize_player_event_storage(db_path=db_path)
+    where_sql, params = _build_common_where(
+        event_type=event_type,
+        server_slug=server_slug,
+        month=month,
+        external_match_id=external_match_id,
+    )
+    with _connect(resolved_path) as connection:
+        rows = connection.execute(
+            f"""
+            SELECT
+                killer_player_key,
+                killer_display_name,
+                victim_player_key,
+                victim_display_name,
+                COALESCE(SUM(event_value), 0) AS total_kills
+            FROM player_event_raw_ledger
+            WHERE {where_sql}
+                AND killer_player_key IS NOT NULL
+                AND victim_player_key IS NOT NULL
+            GROUP BY killer_player_key, killer_display_name, victim_player_key, victim_display_name
+            ORDER BY total_kills DESC, killer_display_name ASC, victim_display_name ASC
+            LIMIT ?
+            """,
+            [*params, limit],
+        ).fetchall()
+    return [dict(row) for row in rows]
+
+
+def _build_common_where(
+    *,
+    event_type: str,
+    server_slug: str | None,
+    month: str | None,
+    external_match_id: str | None,
+) -> tuple[str, list[object]]:
+    clauses = ["event_type = ?"]
+    params: list[object] = [event_type]
+
+    if server_slug and server_slug != "all-servers":
+        clauses.append("server_slug = ?")
+        params.append(server_slug.strip())
+    if month:
+        clauses.append("substr(COALESCE(occurred_at, ''), 1, 7) = ?")
+        params.append(month.strip())
+    if external_match_id:
+        clauses.append("external_match_id = ?")
+        params.append(external_match_id.strip())
+
+    return " AND ".join(clauses), params
+
+
+def _connect(db_path: Path) -> sqlite3.Connection:
+    connection = sqlite3.connect(db_path or get_storage_path())
+    connection.row_factory = sqlite3.Row
+    return connection

backend/app/player_event_models.py

@@ -0,0 +1,32 @@
+"""Normalized player event models for the V2 event pipeline foundation."""
+
+from __future__ import annotations
+
+from dataclasses import asdict, dataclass
+
+
+@dataclass(frozen=True, slots=True)
+class PlayerEventRecord:
+    """Minimal normalized player event contract reused across source and storage."""
+
+    event_id: str
+    event_type: str
+    occurred_at: str | None
+    server_slug: str
+    external_match_id: str
+    source_kind: str
+    source_ref: str | None
+    raw_event_ref: str | None
+    killer_player_key: str | None
+    killer_display_name: str | None
+    victim_player_key: str | None
+    victim_display_name: str | None
+    weapon_name: str | None
+    weapon_category: str | None
+    kill_category: str | None
+    is_teamkill: bool
+    event_value: int = 1
+
+    def to_dict(self) -> dict[str, object]:
+        """Return the event as a plain dictionary."""
+        return asdict(self)

backend/app/player_event_source.py

@@ -0,0 +1,62 @@
+"""Player event source selection and contracts for the V2 pipeline."""
+
+from __future__ import annotations
+
+from typing import Protocol
+
+from .config import get_historical_data_source_kind
+from .player_event_models import PlayerEventRecord
+from .providers.player_event_source_provider import PublicScoreboardPlayerEventSource
+
+
+class PlayerEventSource(Protocol):
+    """Contract for adapters that normalize player event signals."""
+
+    source_kind: str
+
+    def extract_match_events(
+        self,
+        *,
+        server_slug: str,
+        match_payload: dict[str, object],
+        source_ref: str | None = None,
+    ) -> list[PlayerEventRecord]:
+        """Normalize one match payload into reusable player event records."""
+
+    def describe_scope(self) -> dict[str, object]:
+        """Describe what the adapter can and cannot capture today."""
+
+
+class RconPlayerEventSource:
+    """Placeholder adapter for a future raw RCON/log feed."""
+
+    source_kind = "rcon-events"
+
+    def extract_match_events(
+        self,
+        *,
+        server_slug: str,
+        match_payload: dict[str, object],
+        source_ref: str | None = None,
+    ) -> list[PlayerEventRecord]:
+        raise RuntimeError("Raw RCON player event extraction is not implemented yet.")
+
+    def describe_scope(self) -> dict[str, object]:
+        return {
+            "source_kind": self.source_kind,
+            "supports_raw_kill_events": False,
+            "captures": [],
+            "limitations": [
+                "No raw RCON event or log feed is integrated in this repository yet.",
+            ],
+        }
+
+
+def get_player_event_source() -> PlayerEventSource:
+    """Select the event adapter that best matches the configured historical source."""
+    source_kind = get_historical_data_source_kind()
+    if source_kind == "public-scoreboard":
+        return PublicScoreboardPlayerEventSource()
+    if source_kind == "rcon":
+        return RconPlayerEventSource()
+    raise ValueError(f"Unsupported player event source: {source_kind}")

backend/app/player_event_storage.py

@@ -0,0 +1,413 @@
+"""Raw storage and run tracking for the V2 player event pipeline."""
+
+from __future__ import annotations
+
+import sqlite3
+from collections.abc import Iterable
+from datetime import datetime, timezone
+from pathlib import Path
+
+from .config import get_storage_path
+from .player_event_models import PlayerEventRecord
+
+
+def initialize_player_event_storage(*, db_path: Path | None = None) -> Path:
+    """Create the append-only player event ledger and its worker metadata tables."""
+    resolved_path = db_path or get_storage_path()
+    resolved_path.parent.mkdir(parents=True, exist_ok=True)
+
+    with _connect(resolved_path) as connection:
+        connection.executescript(
+            """
+            CREATE TABLE IF NOT EXISTS player_event_raw_ledger (
+                id INTEGER PRIMARY KEY AUTOINCREMENT,
+                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 INTEGER NOT NULL DEFAULT 0,
+                event_value INTEGER NOT NULL DEFAULT 1,
+                inserted_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
+            );
+
+            CREATE TABLE IF NOT EXISTS player_event_ingestion_runs (
+                id INTEGER PRIMARY KEY AUTOINCREMENT,
+                mode TEXT NOT NULL,
+                status TEXT NOT NULL,
+                target_server_slug TEXT,
+                started_at TEXT NOT NULL,
+                completed_at TEXT,
+                pages_processed INTEGER NOT NULL DEFAULT 0,
+                matches_seen INTEGER NOT NULL DEFAULT 0,
+                matches_fetched INTEGER NOT NULL DEFAULT 0,
+                events_inserted INTEGER NOT NULL DEFAULT 0,
+                duplicate_events INTEGER NOT NULL DEFAULT 0,
+                notes TEXT,
+                created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
+            );
+
+            CREATE TABLE IF NOT EXISTS player_event_backfill_progress (
+                server_slug TEXT NOT NULL,
+                mode TEXT NOT NULL,
+                next_page INTEGER NOT NULL DEFAULT 1,
+                last_completed_page INTEGER,
+                cutoff_occurred_at TEXT,
+                discovered_total_matches INTEGER,
+                archive_exhausted INTEGER NOT NULL DEFAULT 0,
+                last_run_id INTEGER,
+                last_run_status TEXT,
+                last_run_started_at TEXT,
+                last_run_completed_at TEXT,
+                last_error TEXT,
+                updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
+                PRIMARY KEY (server_slug, mode)
+            );
+
+            CREATE INDEX IF NOT EXISTS idx_player_event_raw_server_match
+            ON player_event_raw_ledger(server_slug, external_match_id);
+
+            CREATE INDEX IF NOT EXISTS idx_player_event_raw_occurred_at
+            ON player_event_raw_ledger(occurred_at DESC);
+
+            CREATE INDEX IF NOT EXISTS idx_player_event_raw_killer_victim
+            ON player_event_raw_ledger(killer_player_key, victim_player_key);
+            """
+        )
+
+    return resolved_path
+
+
+def upsert_player_events(
+    events: Iterable[PlayerEventRecord],
+    *,
+    db_path: Path | None = None,
+) -> dict[str, int]:
+    """Insert normalized events idempotently into the raw ledger."""
+    resolved_path = initialize_player_event_storage(db_path=db_path)
+    inserted = 0
+    duplicates = 0
+    with _connect(resolved_path) as connection:
+        for event in events:
+            cursor = connection.execute(
+                """
+                INSERT OR IGNORE 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 (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+                """,
+                (
+                    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,
+                    1 if event.is_teamkill else 0,
+                    max(1, int(event.event_value)),
+                ),
+            )
+            if int(cursor.rowcount or 0) > 0:
+                inserted += 1
+            else:
+                duplicates += 1
+    return {
+        "events_inserted": inserted,
+        "duplicate_events": duplicates,
+    }
+
+
+def start_player_event_ingestion_run(
+    *,
+    mode: str,
+    target_server_slug: str | None = None,
+    db_path: Path | None = None,
+) -> int:
+    """Persist one player event ingestion attempt."""
+    resolved_path = initialize_player_event_storage(db_path=db_path)
+    with _connect(resolved_path) as connection:
+        cursor = connection.execute(
+            """
+            INSERT INTO player_event_ingestion_runs (
+                mode,
+                status,
+                target_server_slug,
+                started_at
+            ) VALUES (?, 'running', ?, ?)
+            """,
+            (mode, target_server_slug, _utc_now_iso()),
+        )
+        return int(cursor.lastrowid)
+
+
+def finalize_player_event_ingestion_run(
+    run_id: int,
+    *,
+    status: str,
+    pages_processed: int,
+    matches_seen: int,
+    matches_fetched: int,
+    events_inserted: int,
+    duplicate_events: int,
+    notes: str | None = None,
+    db_path: Path | None = None,
+) -> None:
+    """Update one player event ingestion attempt with final counters."""
+    resolved_path = initialize_player_event_storage(db_path=db_path)
+    with _connect(resolved_path) as connection:
+        connection.execute(
+            """
+            UPDATE player_event_ingestion_runs
+            SET status = ?,
+                completed_at = ?,
+                pages_processed = ?,
+                matches_seen = ?,
+                matches_fetched = ?,
+                events_inserted = ?,
+                duplicate_events = ?,
+                notes = ?
+            WHERE id = ?
+            """,
+            (
+                status,
+                _utc_now_iso(),
+                pages_processed,
+                matches_seen,
+                matches_fetched,
+                events_inserted,
+                duplicate_events,
+                notes,
+                run_id,
+            ),
+        )
+
+
+def mark_player_event_progress_started(
+    *,
+    server_slug: str,
+    mode: str,
+    run_id: int,
+    cutoff_occurred_at: str | None,
+    db_path: Path | None = None,
+) -> None:
+    """Persist the start state for one server ingestion attempt."""
+    resolved_path = initialize_player_event_storage(db_path=db_path)
+    with _connect(resolved_path) as connection:
+        connection.execute(
+            """
+            INSERT INTO player_event_backfill_progress (
+                server_slug,
+                mode,
+                next_page,
+                cutoff_occurred_at,
+                archive_exhausted,
+                last_run_id,
+                last_run_status,
+                last_run_started_at,
+                last_run_completed_at,
+                last_error
+            ) VALUES (?, ?, 1, ?, 0, ?, 'running', ?, NULL, NULL)
+            ON CONFLICT(server_slug, mode) DO UPDATE SET
+                cutoff_occurred_at = excluded.cutoff_occurred_at,
+                last_run_id = excluded.last_run_id,
+                last_run_status = excluded.last_run_status,
+                last_run_started_at = excluded.last_run_started_at,
+                last_run_completed_at = NULL,
+                last_error = NULL,
+                updated_at = CURRENT_TIMESTAMP
+            """,
+            (server_slug, mode, cutoff_occurred_at, run_id, _utc_now_iso()),
+        )
+
+
+def mark_player_event_progress_page_completed(
+    *,
+    server_slug: str,
+    mode: str,
+    page_number: int,
+    discovered_total_matches: int | None,
+    run_id: int,
+    db_path: Path | None = None,
+) -> None:
+    """Advance the resume checkpoint after one page completes successfully."""
+    resolved_path = initialize_player_event_storage(db_path=db_path)
+    with _connect(resolved_path) as connection:
+        connection.execute(
+            """
+            INSERT INTO player_event_backfill_progress (
+                server_slug,
+                mode,
+                next_page,
+                last_completed_page,
+                discovered_total_matches,
+                archive_exhausted,
+                last_run_id,
+                last_run_status,
+                last_run_started_at,
+                last_run_completed_at,
+                last_error
+            ) VALUES (?, ?, ?, ?, ?, 0, ?, 'running', ?, NULL, NULL)
+            ON CONFLICT(server_slug, mode) DO UPDATE SET
+                next_page = excluded.next_page,
+                last_completed_page = excluded.last_completed_page,
+                discovered_total_matches = COALESCE(
+                    excluded.discovered_total_matches,
+                    player_event_backfill_progress.discovered_total_matches
+                ),
+                archive_exhausted = 0,
+                last_run_id = excluded.last_run_id,
+                last_run_status = excluded.last_run_status,
+                last_run_started_at = excluded.last_run_started_at,
+                last_run_completed_at = NULL,
+                last_error = NULL,
+                updated_at = CURRENT_TIMESTAMP
+            """,
+            (
+                server_slug,
+                mode,
+                page_number + 1,
+                page_number,
+                discovered_total_matches,
+                run_id,
+                _utc_now_iso(),
+            ),
+        )
+
+
+def finalize_player_event_progress(
+    *,
+    server_slug: str,
+    mode: str,
+    run_id: int,
+    status: str,
+    archive_exhausted: bool = False,
+    error_message: str | None = None,
+    db_path: Path | None = None,
+) -> None:
+    """Persist the final state of one server event ingestion attempt."""
+    resolved_path = initialize_player_event_storage(db_path=db_path)
+    with _connect(resolved_path) as connection:
+        connection.execute(
+            """
+            INSERT INTO player_event_backfill_progress (
+                server_slug,
+                mode,
+                next_page,
+                archive_exhausted,
+                last_run_id,
+                last_run_status,
+                last_run_started_at,
+                last_run_completed_at,
+                last_error
+            ) VALUES (?, ?, 1, ?, ?, ?, ?, ?, ?)
+            ON CONFLICT(server_slug, mode) DO UPDATE SET
+                archive_exhausted = CASE
+                    WHEN excluded.last_run_status = 'success' AND excluded.archive_exhausted = 1
+                    THEN 1
+                    ELSE player_event_backfill_progress.archive_exhausted
+                END,
+                last_run_id = excluded.last_run_id,
+                last_run_status = excluded.last_run_status,
+                last_run_started_at = COALESCE(
+                    player_event_backfill_progress.last_run_started_at,
+                    excluded.last_run_started_at
+                ),
+                last_run_completed_at = excluded.last_run_completed_at,
+                last_error = excluded.last_error,
+                updated_at = CURRENT_TIMESTAMP
+            """,
+            (
+                server_slug,
+                mode,
+                1 if archive_exhausted else 0,
+                run_id,
+                status,
+                _utc_now_iso(),
+                _utc_now_iso(),
+                error_message,
+            ),
+        )
+
+
+def get_player_event_resume_page(
+    server_slug: str,
+    *,
+    mode: str = "bootstrap",
+    db_path: Path | None = None,
+) -> int:
+    """Return the saved resume page for a bootstrap-like event backfill."""
+    resolved_path = initialize_player_event_storage(db_path=db_path)
+    with _connect(resolved_path) as connection:
+        row = connection.execute(
+            """
+            SELECT next_page
+            FROM player_event_backfill_progress
+            WHERE server_slug = ? AND mode = ?
+            """,
+            (server_slug, mode),
+        ).fetchone()
+    return max(1, int(row["next_page"])) if row and row["next_page"] else 1
+
+
+def get_player_event_refresh_cutoff_for_server(
+    server_slug: str,
+    *,
+    db_path: Path | None = None,
+) -> str | None:
+    """Return the latest occurred_at already persisted for one server."""
+    resolved_path = initialize_player_event_storage(db_path=db_path)
+    with _connect(resolved_path) as connection:
+        row = connection.execute(
+            """
+            SELECT MAX(occurred_at) AS latest_occurred_at
+            FROM player_event_raw_ledger
+            WHERE server_slug = ?
+            """,
+            (server_slug,),
+        ).fetchone()
+    return str(row["latest_occurred_at"]) if row and row["latest_occurred_at"] else None
+
+
+def _connect(db_path: Path) -> sqlite3.Connection:
+    connection = sqlite3.connect(db_path)
+    connection.row_factory = sqlite3.Row
+    connection.execute("PRAGMA foreign_keys = ON")
+    return connection
+
+
+def _utc_now_iso() -> str:
+    return datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")

backend/app/player_event_worker.py

@@ -0,0 +1,462 @@
+"""Incremental worker for the V2 player event ingestion pipeline."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import time
+from dataclasses import dataclass
+from typing import Iterable
+
+from .config import (
+    get_historical_crcon_detail_workers,
+    get_historical_crcon_page_size,
+    get_player_event_refresh_interval_seconds,
+    get_player_event_refresh_max_retries,
+    get_player_event_refresh_retry_delay_seconds,
+)
+from .data_sources import get_historical_data_source
+from .historical_storage import list_historical_servers
+from .player_event_source import get_player_event_source
+from .player_event_storage import (
+    finalize_player_event_ingestion_run,
+    finalize_player_event_progress,
+    get_player_event_refresh_cutoff_for_server,
+    get_player_event_resume_page,
+    initialize_player_event_storage,
+    mark_player_event_progress_page_completed,
+    mark_player_event_progress_started,
+    start_player_event_ingestion_run,
+    upsert_player_events,
+)
+
+
+@dataclass(slots=True)
+class PlayerEventIngestionStats:
+    pages_processed: int = 0
+    matches_seen: int = 0
+    matches_fetched: int = 0
+    events_inserted: int = 0
+    duplicate_events: int = 0
+
+    def apply(self, delta: dict[str, int]) -> None:
+        self.events_inserted += int(delta.get("events_inserted", 0))
+        self.duplicate_events += int(delta.get("duplicate_events", 0))
+
+
+def run_player_event_refresh(
+    *,
+    server_slug: str | None = None,
+    max_pages: int | None = None,
+    page_size: int | None = None,
+    start_page: int | None = None,
+    detail_workers: int | None = None,
+) -> dict[str, object]:
+    """Refresh recent player event summaries from the configured historical source."""
+    initialize_player_event_storage()
+    data_source = get_historical_data_source()
+    event_source = get_player_event_source()
+    resolved_page_size = page_size or get_historical_crcon_page_size()
+    resolved_detail_workers = detail_workers or get_historical_crcon_detail_workers()
+    selected_servers = _select_servers(server_slug)
+    processed_servers: list[dict[str, object]] = []
+    active_runs: dict[str, int] = {}
+
+    try:
+        for server in selected_servers:
+            current_server_slug = str(server["slug"])
+            run_id = start_player_event_ingestion_run(
+                mode="refresh",
+                target_server_slug=current_server_slug,
+            )
+            active_runs[current_server_slug] = run_id
+            cutoff = get_player_event_refresh_cutoff_for_server(current_server_slug)
+            mark_player_event_progress_started(
+                server_slug=current_server_slug,
+                mode="refresh",
+                run_id=run_id,
+                cutoff_occurred_at=cutoff,
+            )
+            server_stats = _ingest_server(
+                server=server,
+                run_id=run_id,
+                data_source=data_source,
+                event_source=event_source,
+                page_size=resolved_page_size,
+                max_pages=max_pages,
+                start_page=_resolve_start_page(
+                    server_slug=current_server_slug,
+                    start_page=start_page,
+                ),
+                detail_workers=resolved_detail_workers,
+                cutoff=cutoff,
+            )
+            finalize_player_event_ingestion_run(
+                run_id,
+                status="success",
+                pages_processed=server_stats["pages_processed"],
+                matches_seen=server_stats["matches_seen"],
+                matches_fetched=server_stats["matches_fetched"],
+                events_inserted=server_stats["events_inserted"],
+                duplicate_events=server_stats["duplicate_events"],
+                notes=f"source={data_source.source_kind};adapter={event_source.source_kind}",
+            )
+            finalize_player_event_progress(
+                server_slug=current_server_slug,
+                mode="refresh",
+                run_id=run_id,
+                status="success",
+                archive_exhausted=bool(server_stats["archive_exhausted"]),
+            )
+            processed_servers.append(server_stats)
+            active_runs.pop(current_server_slug, None)
+    except Exception as exc:
+        for active_server_slug, run_id in active_runs.items():
+            finalize_player_event_ingestion_run(
+                run_id,
+                status="failed",
+                pages_processed=0,
+                matches_seen=0,
+                matches_fetched=0,
+                events_inserted=0,
+                duplicate_events=0,
+                notes=str(exc),
+            )
+            finalize_player_event_progress(
+                server_slug=active_server_slug,
+                mode="refresh",
+                run_id=run_id,
+                status="failed",
+                error_message=str(exc),
+            )
+        raise
+
+    return {
+        "status": "ok",
+        "mode": "refresh",
+        "source_provider": data_source.source_kind,
+        "event_adapter": event_source.source_kind,
+        "page_size": resolved_page_size,
+        "detail_workers": resolved_detail_workers,
+        "scope": event_source.describe_scope(),
+        "servers": processed_servers,
+    }
+
+
+def run_periodic_player_event_refresh(
+    *,
+    interval_seconds: int,
+    max_retries: int,
+    retry_delay_seconds: int,
+    server_slug: str | None = None,
+    max_pages: int | None = None,
+    page_size: int | None = None,
+    detail_workers: int | None = None,
+    max_runs: int | None = None,
+) -> None:
+    """Run the refresh worker repeatedly with bounded retries."""
+    completed_runs = 0
+    print(
+        json.dumps(
+            {
+                "event": "player-event-refresh-loop-started",
+                "interval_seconds": interval_seconds,
+                "max_retries": max_retries,
+                "retry_delay_seconds": retry_delay_seconds,
+                "server_scope": [server_slug] if server_slug else [server["slug"] for server in list_historical_servers()],
+            },
+            indent=2,
+        )
+    )
+    print("Press Ctrl+C to stop.")
+
+    try:
+        while max_runs is None or completed_runs < max_runs:
+            completed_runs += 1
+            payload = _run_refresh_with_retries(
+                max_retries=max_retries,
+                retry_delay_seconds=retry_delay_seconds,
+                server_slug=server_slug,
+                max_pages=max_pages,
+                page_size=page_size,
+                detail_workers=detail_workers,
+            )
+            print(json.dumps({"run": completed_runs, **payload}, indent=2))
+            if max_runs is not None and completed_runs >= max_runs:
+                break
+            time.sleep(interval_seconds)
+    except KeyboardInterrupt:
+        print("\nPlayer event refresh loop stopped by user.")
+
+
+def _run_refresh_with_retries(
+    *,
+    max_retries: int,
+    retry_delay_seconds: int,
+    server_slug: str | None,
+    max_pages: int | None,
+    page_size: int | None,
+    detail_workers: int | None,
+) -> dict[str, object]:
+    attempt = 0
+    while True:
+        attempt += 1
+        try:
+            return {
+                "status": "ok",
+                "attempts_used": attempt,
+                "refresh_result": run_player_event_refresh(
+                    server_slug=server_slug,
+                    max_pages=max_pages,
+                    page_size=page_size,
+                    detail_workers=detail_workers,
+                ),
+            }
+        except Exception as exc:
+            if attempt > max_retries:
+                return {
+                    "status": "error",
+                    "attempts_used": attempt,
+                    "error": str(exc),
+                }
+            if retry_delay_seconds > 0:
+                time.sleep(retry_delay_seconds)
+
+
+def _ingest_server(
+    *,
+    server: dict[str, object],
+    run_id: int,
+    data_source: object,
+    event_source: object,
+    page_size: int,
+    max_pages: int | None,
+    start_page: int,
+    detail_workers: int,
+    cutoff: str | None,
+) -> dict[str, object]:
+    page_limit = max_pages or 1000000
+    local_stats = PlayerEventIngestionStats()
+    discovered_total_matches: int | None = None
+    archive_exhausted = False
+
+    for page_number in range(start_page, start_page + page_limit):
+        payload = data_source.fetch_match_page(
+            base_url=str(server["scoreboard_base_url"]),
+            page=page_number,
+            limit=page_size,
+        )
+        if discovered_total_matches is None:
+            discovered_total_matches = _coerce_int(payload.get("total"))
+        page_matches = _coerce_match_list(payload.get("maps"))
+        if not page_matches:
+            archive_exhausted = True
+            break
+
+        local_stats.pages_processed += 1
+        stop_after_page = False
+        match_ids_to_fetch: list[str] = []
+
+        for match_summary in page_matches:
+            local_stats.matches_seen += 1
+            reference_timestamp = _pick_match_timestamp(match_summary)
+            if cutoff and reference_timestamp and reference_timestamp < cutoff:
+                stop_after_page = True
+                continue
+
+            match_id = _stringify(match_summary.get("id"))
+            if match_id:
+                match_ids_to_fetch.append(match_id)
+
+        detail_payloads = data_source.fetch_match_details(
+            base_url=str(server["scoreboard_base_url"]),
+            match_ids=match_ids_to_fetch,
+            max_workers=detail_workers,
+        )
+        local_stats.matches_fetched += len(detail_payloads)
+        for detail_payload in detail_payloads:
+            match_id = _stringify(detail_payload.get("id")) or "unknown"
+            source_ref = (
+                f"{server['scoreboard_base_url']}/api/get_map_scoreboard?map_id={match_id}"
+            )
+            normalized_events = event_source.extract_match_events(
+                server_slug=str(server["slug"]),
+                match_payload=detail_payload,
+                source_ref=source_ref,
+            )
+            local_stats.apply(upsert_player_events(normalized_events))
+
+        mark_player_event_progress_page_completed(
+            server_slug=str(server["slug"]),
+            mode="refresh",
+            page_number=page_number,
+            discovered_total_matches=discovered_total_matches,
+            run_id=run_id,
+        )
+
+        if stop_after_page:
+            break
+
+    return {
+        "server_slug": server["slug"],
+        "source_provider": data_source.source_kind,
+        "event_adapter": event_source.source_kind,
+        "pages_processed": local_stats.pages_processed,
+        "matches_seen": local_stats.matches_seen,
+        "matches_fetched": local_stats.matches_fetched,
+        "events_inserted": local_stats.events_inserted,
+        "duplicate_events": local_stats.duplicate_events,
+        "cutoff": cutoff,
+        "archive_exhausted": archive_exhausted,
+        "discovered_total_matches": discovered_total_matches,
+    }
+
+
+def _resolve_start_page(*, server_slug: str, start_page: int | None) -> int:
+    if start_page is not None:
+        return max(1, start_page)
+    return get_player_event_resume_page(server_slug, mode="refresh")
+
+
+def _select_servers(server_slug: str | None) -> list[dict[str, object]]:
+    servers = list_historical_servers()
+    if server_slug is None:
+        return servers
+    normalized = server_slug.strip()
+    selected = [server for server in servers if server["slug"] == normalized]
+    if not selected:
+        raise ValueError(f"Unknown historical server slug: {server_slug}")
+    return selected
+
+
+def _coerce_match_list(payload: object) -> list[dict[str, object]]:
+    if not isinstance(payload, list):
+        return []
+    return [item for item in payload if isinstance(item, dict)]
+
+
+def _pick_match_timestamp(match_payload: dict[str, object]) -> str | None:
+    for key in ("end", "start", "creation_time"):
+        value = match_payload.get(key)
+        if isinstance(value, str) and value.strip():
+            return value.strip()
+    return None
+
+
+def _stringify(value: object) -> str | None:
+    if value is None:
+        return None
+    text = str(value).strip()
+    return text or None
+
+
+def _coerce_int(value: object) -> int | None:
+    if value in (None, ""):
+        return None
+    try:
+        return int(value)
+    except (TypeError, ValueError):
+        return None
+
+
+def build_arg_parser() -> argparse.ArgumentParser:
+    """Create the CLI parser for manual or periodic player event ingestion."""
+    parser = argparse.ArgumentParser(
+        description="Player event refresh worker for HLL Vietnam.",
+    )
+    parser.add_argument(
+        "mode",
+        choices=("refresh", "loop"),
+        help="refresh runs once; loop keeps the worker running periodically",
+    )
+    parser.add_argument(
+        "--server",
+        dest="server_slug",
+        help="optional historical server slug",
+    )
+    parser.add_argument(
+        "--max-pages",
+        type=int,
+        help="optional page cap for local validation",
+    )
+    parser.add_argument(
+        "--page-size",
+        type=int,
+        help="override CRCON page size",
+    )
+    parser.add_argument(
+        "--start-page",
+        type=int,
+        help="override the saved resume page",
+    )
+    parser.add_argument(
+        "--detail-workers",
+        type=int,
+        help="parallel worker count for per-match detail requests",
+    )
+    parser.add_argument(
+        "--interval",
+        type=int,
+        default=get_player_event_refresh_interval_seconds(),
+        help="seconds to wait between loop runs",
+    )
+    parser.add_argument(
+        "--retries",
+        type=int,
+        default=get_player_event_refresh_max_retries(),
+        help="retry attempts after a failed refresh",
+    )
+    parser.add_argument(
+        "--retry-delay",
+        type=int,
+        default=get_player_event_refresh_retry_delay_seconds(),
+        help="seconds to wait between failed attempts",
+    )
+    parser.add_argument(
+        "--max-runs",
+        type=int,
+        help="optional safety cap for loop mode",
+    )
+    return parser
+
+
+def main(argv: Iterable[str] | None = None) -> int:
+    """Run the player event worker CLI."""
+    parser = build_arg_parser()
+    args = parser.parse_args(list(argv) if argv is not None else None)
+
+    if args.mode == "refresh":
+        result = run_player_event_refresh(
+            server_slug=args.server_slug,
+            max_pages=args.max_pages,
+            page_size=args.page_size,
+            start_page=args.start_page,
+            detail_workers=args.detail_workers,
+        )
+        print(json.dumps(result, indent=2))
+        return 0
+
+    if args.interval <= 0:
+        raise ValueError("--interval must be a positive integer.")
+    if args.retries < 0:
+        raise ValueError("--retries must be zero or positive.")
+    if args.retry_delay < 0:
+        raise ValueError("--retry-delay must be zero or positive.")
+    if args.max_runs is not None and args.max_runs <= 0:
+        raise ValueError("--max-runs must be positive when provided.")
+
+    run_periodic_player_event_refresh(
+        interval_seconds=args.interval,
+        max_retries=args.retries,
+        retry_delay_seconds=args.retry_delay,
+        server_slug=args.server_slug,
+        max_pages=args.max_pages,
+        page_size=args.page_size,
+        detail_workers=args.detail_workers,
+        max_runs=args.max_runs,
+    )
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

backend/app/providers/player_event_source_provider.py

@@ -0,0 +1,328 @@
+"""Player event adapter backed by public CRCON scoreboard match details."""
+
+from __future__ import annotations
+
+import hashlib
+from collections.abc import Mapping
+from dataclasses import dataclass
+
+from ..player_event_models import PlayerEventRecord
+
+
+@dataclass(frozen=True, slots=True)
+class _PlayerIdentity:
+    stable_player_key: str
+    display_name: str | None
+
+
+@dataclass(frozen=True, slots=True)
+class PublicScoreboardPlayerEventSource:
+    """Normalize partial duel and weapon signals from CRCON match detail payloads."""
+
+    source_kind: str = "public-scoreboard-match-summary"
+
+    def extract_match_events(
+        self,
+        *,
+        server_slug: str,
+        match_payload: dict[str, object],
+        source_ref: str | None = None,
+    ) -> list[PlayerEventRecord]:
+        match_id = _stringify(match_payload.get("id"))
+        if not match_id:
+            return []
+
+        occurred_at = _pick_match_timestamp(match_payload)
+        player_rows = _coerce_player_rows(match_payload.get("player_stats"))
+        if not player_rows:
+            return []
+
+        identity_index = _build_identity_index(player_rows)
+        events: list[PlayerEventRecord] = []
+
+        for player_row in player_rows:
+            actor = _build_player_identity(player_row)
+            if actor is None:
+                continue
+
+            top_victim_name, top_victim_count = _extract_named_count(player_row.get("most_killed"))
+            top_weapon_name, _ = _extract_named_count(player_row.get("weapons"))
+            top_kill_type_name, _ = _extract_named_count(player_row.get("kills_by_type"))
+            victim = _find_identity_by_name(identity_index, top_victim_name)
+            if victim is not None and top_victim_count > 0:
+                events.append(
+                    _build_event(
+                        event_type="player_kill_summary",
+                        occurred_at=occurred_at,
+                        server_slug=server_slug,
+                        match_id=match_id,
+                        source_kind=self.source_kind,
+                        source_ref=source_ref,
+                        raw_event_ref=f"match:{match_id}:player:{actor.stable_player_key}:most-killed",
+                        killer=actor,
+                        victim=victim,
+                        weapon_name=top_weapon_name,
+                        kill_category=top_kill_type_name,
+                        is_teamkill=False,
+                        event_value=top_victim_count,
+                    )
+                )
+
+            top_killer_name, top_killer_count = _extract_named_count(player_row.get("death_by"))
+            death_weapon_name, _ = _extract_named_count(player_row.get("death_by_weapons"))
+            killer = _find_identity_by_name(identity_index, top_killer_name)
+            if killer is not None and top_killer_count > 0:
+                events.append(
+                    _build_event(
+                        event_type="player_death_summary",
+                        occurred_at=occurred_at,
+                        server_slug=server_slug,
+                        match_id=match_id,
+                        source_kind=self.source_kind,
+                        source_ref=source_ref,
+                        raw_event_ref=f"match:{match_id}:player:{actor.stable_player_key}:death-by",
+                        killer=killer,
+                        victim=actor,
+                        weapon_name=death_weapon_name,
+                        kill_category=None,
+                        is_teamkill=False,
+                        event_value=top_killer_count,
+                    )
+                )
+
+            teamkills = _coerce_int(player_row.get("teamkills")) or 0
+            if teamkills > 0:
+                events.append(
+                    _build_event(
+                        event_type="player_teamkill_summary",
+                        occurred_at=occurred_at,
+                        server_slug=server_slug,
+                        match_id=match_id,
+                        source_kind=self.source_kind,
+                        source_ref=source_ref,
+                        raw_event_ref=f"match:{match_id}:player:{actor.stable_player_key}:teamkills",
+                        killer=actor,
+                        victim=None,
+                        weapon_name=top_weapon_name,
+                        kill_category=top_kill_type_name,
+                        is_teamkill=True,
+                        event_value=teamkills,
+                    )
+                )
+
+        return events
+
+    def describe_scope(self) -> dict[str, object]:
+        return {
+            "source_kind": self.source_kind,
+            "supports_raw_kill_events": False,
+            "captures": [
+                "Top victim per player from most_killed",
+                "Top killer per player from death_by",
+                "Top weapon hints from weapons and death_by_weapons",
+                "Aggregated teamkills per player and match",
+            ],
+            "limitations": [
+                "The current source is match-summary data, not a true per-kill event feed.",
+                "occurred_at uses the match end/start timestamp, not the exact kill timestamp.",
+                "Only the strongest encounter and weapon signals available in the CRCON detail payload are normalized.",
+                "Full killer->victim ledgers, complete weapon breakdowns, and exact per-event teamkills still require a dedicated raw event/log source.",
+            ],
+        }
+
+
+def _build_identity_index(player_rows: list[dict[str, object]]) -> dict[str, _PlayerIdentity]:
+    identity_index: dict[str, _PlayerIdentity] = {}
+    for player_row in player_rows:
+        identity = _build_player_identity(player_row)
+        if identity is None or not identity.display_name:
+            continue
+        identity_index[_normalize_name(identity.display_name)] = identity
+    return identity_index
+
+
+def _build_player_identity(player_row: dict[str, object]) -> _PlayerIdentity | None:
+    display_name = _stringify(player_row.get("player")) or _stringify(player_row.get("name"))
+    source_player_id = _stringify(player_row.get("player_id")) or _stringify(player_row.get("id"))
+    steam_id = _extract_steam_id(player_row.get("steaminfo"))
+    stable_player_key = _build_stable_player_key(steam_id=steam_id, source_player_id=source_player_id)
+    if stable_player_key is None:
+        return None
+    return _PlayerIdentity(
+        stable_player_key=stable_player_key,
+        display_name=display_name or stable_player_key,
+    )
+
+
+def _find_identity_by_name(
+    identity_index: dict[str, _PlayerIdentity],
+    player_name: str | None,
+) -> _PlayerIdentity | None:
+    if not player_name:
+        return None
+    return identity_index.get(_normalize_name(player_name))
+
+
+def _build_event(
+    *,
+    event_type: str,
+    occurred_at: str | None,
+    server_slug: str,
+    match_id: str,
+    source_kind: str,
+    source_ref: str | None,
+    raw_event_ref: str,
+    killer: _PlayerIdentity | None,
+    victim: _PlayerIdentity | None,
+    weapon_name: str | None,
+    kill_category: str | None,
+    is_teamkill: bool,
+    event_value: int,
+) -> PlayerEventRecord:
+    event_id = _build_event_id(
+        event_type=event_type,
+        occurred_at=occurred_at,
+        server_slug=server_slug,
+        match_id=match_id,
+        killer_player_key=killer.stable_player_key if killer else None,
+        victim_player_key=victim.stable_player_key if victim else None,
+        weapon_name=weapon_name,
+        is_teamkill=is_teamkill,
+        event_value=event_value,
+    )
+    return PlayerEventRecord(
+        event_id=event_id,
+        event_type=event_type,
+        occurred_at=occurred_at,
+        server_slug=server_slug,
+        external_match_id=match_id,
+        source_kind=source_kind,
+        source_ref=source_ref,
+        raw_event_ref=raw_event_ref,
+        killer_player_key=killer.stable_player_key if killer else None,
+        killer_display_name=killer.display_name if killer else None,
+        victim_player_key=victim.stable_player_key if victim else None,
+        victim_display_name=victim.display_name if victim else None,
+        weapon_name=weapon_name,
+        weapon_category=None,
+        kill_category=kill_category,
+        is_teamkill=is_teamkill,
+        event_value=max(1, event_value),
+    )
+
+
+def _build_event_id(
+    *,
+    event_type: str,
+    occurred_at: str | None,
+    server_slug: str,
+    match_id: str,
+    killer_player_key: str | None,
+    victim_player_key: str | None,
+    weapon_name: str | None,
+    is_teamkill: bool,
+    event_value: int,
+) -> str:
+    raw_key = "|".join(
+        [
+            event_type,
+            occurred_at or "",
+            server_slug,
+            match_id,
+            killer_player_key or "",
+            victim_player_key or "",
+            weapon_name or "",
+            "1" if is_teamkill else "0",
+            str(event_value),
+        ]
+    )
+    return hashlib.sha1(raw_key.encode("utf-8")).hexdigest()
+
+
+def _pick_match_timestamp(match_payload: Mapping[str, object]) -> str | None:
+    for key in ("end", "start", "creation_time"):
+        value = _stringify(match_payload.get(key))
+        if value:
+            return value
+    return None
+
+
+def _extract_named_count(value: object) -> tuple[str | None, int]:
+    if isinstance(value, str):
+        return _stringify(value), 1
+    if isinstance(value, Mapping):
+        nested_name = None
+        nested_player = value.get("player")
+        if isinstance(nested_player, Mapping):
+            nested_name = _stringify(nested_player.get("name")) or _stringify(nested_player.get("player"))
+        name = (
+            _stringify(value.get("name"))
+            or _stringify(value.get("player"))
+            or _stringify(value.get("victim"))
+            or _stringify(value.get("killer"))
+            or nested_name
+        )
+        count = (
+            _coerce_int(value.get("count"))
+            or _coerce_int(value.get("kills"))
+            or _coerce_int(value.get("deaths"))
+            or _coerce_int(value.get("value"))
+            or _coerce_int(value.get("total"))
+            or 1
+        )
+        return name, max(1, count)
+    if isinstance(value, list):
+        for item in value:
+            name, count = _extract_named_count(item)
+            if name:
+                return name, count
+    return None, 0
+
+
+def _extract_steam_id(value: object) -> str | None:
+    if isinstance(value, Mapping):
+        profile = value.get("profile")
+        if isinstance(profile, Mapping):
+            steam_id = _stringify(profile.get("steamid"))
+            if steam_id:
+                return steam_id
+        return _stringify(value.get("id"))
+    return None
+
+
+def _build_stable_player_key(
+    *,
+    steam_id: str | None,
+    source_player_id: str | None,
+) -> str | None:
+    if steam_id:
+        return f"steam:{steam_id}"
+    if source_player_id:
+        return f"crcon-player:{source_player_id}"
+    return None
+
+
+def _coerce_player_rows(value: object) -> list[dict[str, object]]:
+    if not isinstance(value, list):
+        return []
+    return [item for item in value if isinstance(item, dict)]
+
+
+def _normalize_name(value: str) -> str:
+    return value.strip().casefold()
+
+
+def _stringify(value: object) -> str | None:
+    if value is None:
+        return None
+    text = str(value).strip()
+    return text or None
+
+
+def _coerce_int(value: object) -> int | None:
+    if value in (None, ""):
+        return None
+    try:
+        return int(value)
+    except (TypeError, ValueError):
+        return None