Export app, process and log telemetry when OTEL_ENABLED is set

Adds tavolo.active_users and tavolo.ongoing_games gauges, process/CPython metrics via opentelemetry-instrumentation-system-metrics and OTLP log export. The game store tracks ongoing matches (Redis sorted set, in-memory set) and the session stores expose live counts for the gauges; a poller refreshes them since metric callbacks cannot await the stores.
2026-09-28 03:38:59 +08:00
parent 5be4926885
commit 98d4485375
15 changed files with 872 additions and 31 deletions
@@ -11,6 +11,10 @@ The deployable tavolo application: the composition root wiring
  `ScoponeEngine` with its configured timeouts, and exposes
  `tavolo.app:app` for granian.
 - `config.py` — environment-driven frozen `Settings`.
+- `telemetry.py` — the optional OpenTelemetry wiring beyond kaya-otel:
+  meter/logger providers, the `tavolo.active_users`/`tavolo.ongoing_games`
+  gauges (polled from the stores), process/CPython metrics via
+  `opentelemetry-instrumentation-system-metrics` and OTLP log export.
 - `aerich_config.py` — Tortoise ORM configuration consumed by the aerich
  CLI (migrations live in `server/migrations/`; aerich itself is
  configured via the `[tool.aerich]` section of `server/pyproject.toml`).
@@ -23,6 +27,7 @@ The deployable tavolo application: the composition root wiring
 .venv/bin/python -m mypy --namespace-packages --explicit-package-bases \
  src/tavolo/app.py \
  src/tavolo/config.py \
+  src/tavolo/telemetry.py \
  src/tavolo/static.py \
  src/tavolo/aerich_config.py \
  src/tavolo/logging_config.py
@@ -35,6 +35,7 @@ dev = [
 ]
 otel = [
    "kaya-otel>=0.0.4",
+    "opentelemetry-instrumentation-system-metrics>=0.66b0",
 ]

 [tool.setuptools.packages.find]
@@ -31,14 +31,13 @@ from __future__ import annotations

 from importlib.metadata import version as _pkg_version
 from logging import getLogger
-from typing import Optional
+from typing import Any, Dict, Optional

 from kaya.core import KayaApp, KayaMixin
 from kaya.cors import CorsMixin
 from kaya.oidc import OIDCConfig, OIDCMixin
 from kaya.openapi import OpenAPIMixin
-from kaya.session import InMemorySessionStore, SessionMixin, SessionStore
-from kaya.session.redis import RedisSessionStore
+from kaya.session import SessionMixin
 from redis.asyncio import Redis
 from tavolo.platform import GameRegistry, Platform, PlatformMixin
 from tavolo.platform.deadlines import DeadlineScheduler, DeadlineSchedulerMixin
@@ -48,6 +47,13 @@ from tavolo.scopone import ScoponeEngine

 from .config import Settings, settings
 from .logging_config import configure_logging
+from .telemetry import (
+    CountableSessionStore,
+    CountingInMemorySessionStore,
+    CountingRedisSessionStore,
+    otel_headers,
+    telemetry_from_settings,
+)

 configure_logging(settings.logging_config)
 log = getLogger(__name__)
@@ -73,13 +79,20 @@ def cors_mixin_from_settings(settings: Settings) -> Optional[CorsMixin]:
    )


-def otel_mixin_from_settings(settings: Settings) -> Optional[KayaMixin]:
+def otel_mixin_from_settings(
+    settings: Settings, meter_provider: Optional[Any] = None
+) -> Optional[KayaMixin]:
    """Build a :class:`~kaya.otel.OTelMixin` from the OTEL_* settings.

    Returns ``None`` — telemetry disabled — unless ``OTEL_ENABLED`` is
    truthy. kaya-otel is an optional dependency (the ``otel`` extra), so it
    is imported lazily here: default installs and the test suite never need
    the OpenTelemetry packages.
+
+    ``meter_provider`` is the provider built by
+    :func:`tavolo.telemetry.telemetry_from_settings`; injecting it lets the
+    custom gauges share the same OTLP pipeline (kaya-otel then leaves its
+    shutdown to the telemetry mixin).
    """
    if not settings.otel_enabled:
        return None
@@ -90,17 +103,15 @@ def otel_mixin_from_settings(settings: Settings) -> Optional[KayaMixin]:
            "OTEL_ENABLED is set but kaya-otel is not installed; "
            "install tavolo-app with the 'otel' extra"
        ) from exc
-    headers = dict(
-        pair.split("=", 1)
-        for pair in (settings.otel_exporter_headers or ())
-        if "=" in pair
-    )
-    return OTelMixin(
-        service_name=settings.otel_service_name,
-        endpoint=settings.otel_exporter_endpoint,
-        headers=headers or None,
-        excluded_paths=settings.otel_excluded_paths,
-    )
+    kwargs: Dict[str, Any] = {
+        "service_name": settings.otel_service_name,
+        "endpoint": settings.otel_exporter_endpoint,
+        "headers": otel_headers(settings),
+        "excluded_paths": settings.otel_excluded_paths,
+    }
+    if meter_provider is not None:
+        kwargs["meter_provider"] = meter_provider
+    return OTelMixin(**kwargs)


 registry = GameRegistry()
@@ -117,12 +128,12 @@ log.debug(
    settings.turn_timeout_seconds,
 )

-session_store: SessionStore
+session_store: CountableSessionStore
 game_store: GameStore
 if settings.redis_url is not None:
    # Lazy client: no connection is opened until a session is actually
    # loaded/saved, so importing this module never requires a live Redis.
-    session_store = RedisSessionStore(Redis.from_url(settings.redis_url))
+    session_store = CountingRedisSessionStore(Redis.from_url(settings.redis_url))
    game_store = RedisGameStore(
        Redis.from_url(settings.redis_url, decode_responses=False),
        registry,
@@ -130,7 +141,7 @@ if settings.redis_url is not None:
    )
    log.info("using Redis stores (sessions + live games, game TTL %ds)", settings.game_ttl_seconds)
 else:
-    session_store = InMemorySessionStore()
+    session_store = CountingInMemorySessionStore()
    game_store = InMemoryGameStore(registry)
    log.info("REDIS_URL unset: using in-memory stores (sessions + live games)")

@@ -176,15 +187,22 @@ platform = Platform(

 mixins: list[KayaMixin] = [session_mixin, oidc_mixin, tortoise_mixin, openapi_mixin,
                           PlatformMixin(platform), DeadlineSchedulerMixin(scheduler)]
-otel_mixin = otel_mixin_from_settings(settings)
-if otel_mixin is not None:
-    # First in the list: before hooks run in registration order (after hooks
+telemetry = telemetry_from_settings(settings, session_store, game_store)
+if telemetry is not None:
+    otel_mixin = otel_mixin_from_settings(
+        settings, meter_provider=telemetry.meter_provider
+    )
+    assert otel_mixin is not None  # OTEL_ENABLED was checked above
+    # kaya-otel first: before hooks run in registration order (after hooks
    # in reverse), so the span covers session loading, OIDC handling and the
    # handler itself. CORS, when enabled, is still prepended before it so
-    # preflight short-circuits stay untraced.
+    # preflight short-circuits stay untraced. The telemetry mixin carries no
+    # hooks but owns the metric/logger providers' polling and shutdown, so
+    # it must be in the list too.
+    mixins.insert(0, telemetry.mixin)
    mixins.insert(0, otel_mixin)
    log.info(
-        "OpenTelemetry enabled: service=%s endpoint=%s",
+        "OpenTelemetry enabled: service=%s endpoint=%s (traces, metrics, logs)",
        settings.otel_service_name,
        settings.otel_exporter_endpoint or "(OTLP default)",
    )
@@ -125,6 +125,10 @@ class Settings:
    # Paths excluded from tracing and metrics (exact matches). Defaults to
    # the health endpoint, which k8s probes would otherwise spam.
    otel_excluded_paths: Tuple[str, ...]
+    # How often the OTLP metric reader exports (milliseconds) and how often
+    # the telemetry gauges refresh the session/game counts (seconds).
+    otel_metric_export_interval_millis: int
+    otel_metrics_poll_interval_seconds: int

    @staticmethod
    def from_env() -> "Settings":
@@ -171,14 +175,20 @@ class Settings:
            cors_allow_credentials=_env_bool("CORS_ALLOW_CREDENTIALS"),
            cors_expose_headers=_env_list("CORS_EXPOSE_HEADERS"),
            cors_max_age=int(_env("CORS_MAX_AGE", "600")),
-            # OpenTelemetry is opt-in: set OTEL_ENABLED=1 to export traces
-            # and metrics via OTLP/HTTP (requires the ``otel`` extra).
+            # OpenTelemetry is opt-in: set OTEL_ENABLED=1 to export traces,
+            # metrics and logs via OTLP/HTTP (requires the ``otel`` extra).
            otel_enabled=_env_bool("OTEL_ENABLED"),
            otel_service_name=_env("OTEL_SERVICE_NAME", "tavolo"),
            otel_exporter_endpoint=os.environ.get("OTEL_EXPORTER_OTLP_ENDPOINT") or None,
            # Comma-separated key=value pairs, e.g. "Authorization=Bearer x".
            otel_exporter_headers=_env_list("OTEL_EXPORTER_OTLP_HEADERS"),
            otel_excluded_paths=_env_list("OTEL_EXCLUDED_PATHS") or ("/api/health",),
+            otel_metric_export_interval_millis=int(
+                _env("OTEL_METRIC_EXPORT_INTERVAL_MILLIS", "60000")
+            ),
+            otel_metrics_poll_interval_seconds=int(
+                _env("OTEL_METRICS_POLL_INTERVAL_SECONDS", "15")
+            ),
        )


@@ -0,0 +1,300 @@
+"""OpenTelemetry metrics and log export beyond kaya-otel.
+
+kaya-otel instruments HTTP requests and websockets with traces and HTTP
+metrics; this module adds what it does not cover, all behind the optional
+``otel`` extra (OpenTelemetry imports are lazy, so default installs and
+the test suite never need them):
+
+* application gauges: ``tavolo.active_users`` (sessions held in the
+  session store) and ``tavolo.ongoing_games`` (live matches that have not
+  finished yet);
+* process metrics through ``opentelemetry-instrumentation-system-metrics``:
+  CPU time/utilization, memory usage and CPython garbage-collection
+  counters;
+* OTLP log export through a stdlib :class:`logging.Handler` attached to
+  the root logger, mirroring what the console sees.
+
+The store counts are asynchronous while OpenTelemetry instrument
+callbacks are synchronous, so :class:`TelemetryMixin` polls the stores on
+a background task and its callbacks report the cached values.
+"""
+from __future__ import annotations
+
+import asyncio
+import logging
+from dataclasses import dataclass
+from logging import getLogger
+from typing import TYPE_CHECKING, Any, Dict, Optional, Union
+
+from kaya.core import KayaApp, KayaMixin
+from kaya.session import InMemorySessionStore, SessionStore
+from kaya.session.redis import RedisSessionStore
+from redis.asyncio import Redis
+from tavolo.platform.store import GameStore
+
+from .config import Settings
+
+if TYPE_CHECKING:
+    from opentelemetry.sdk._logs import LoggerProvider
+    from opentelemetry.sdk.metrics import MeterProvider
+
+log = getLogger(__name__)
+
+# Session keys live under kaya's default prefix. Keeping it explicit here
+# makes the Redis count scan exact and preserves existing sessions.
+SESSION_KEY_PREFIX = "kaya:session:"
+
+# The interpreter metrics to export. The instrumentation package can also
+# emit host-level ``system.*`` metrics; those are deliberately left off:
+# the application cares about its own process, and per-CPU host metrics
+# add cardinality without context.
+#
+# ``process.cpu.utilization`` ignores its configuration value upstream
+# (the callback always emits one aggregate observation), so its ``None``
+# is only for readability.
+PROCESS_METRICS_CONFIG: Dict[str, Optional[list]] = {
+    "process.cpu.time": ["user", "system"],
+    "process.cpu.utilization": None,
+    "process.memory.usage": None,
+    "process.memory.virtual": None,
+    "cpython.gc.collections": None,
+    "cpython.gc.collected_objects": None,
+    "cpython.gc.uncollectable_objects": None,
+}
+
+
+def otel_headers(settings: Settings) -> Optional[Dict[str, str]]:
+    """Parse ``OTEL_EXPORTER_OTLP_HEADERS`` into a mapping (or ``None``)."""
+    headers = dict(
+        pair.split("=", 1)
+        for pair in (settings.otel_exporter_headers or ())
+        if "=" in pair
+    )
+    return headers or None
+
+
+def _signal_endpoint(base: Optional[str], signal: str) -> Optional[str]:
+    """``{base}/v1/<signal>``, or ``None`` for the exporter's own default."""
+    return f"{base}/{signal}" if base else None
+
+
+class _NoOtelLogs(logging.Filter):
+    """Drop OpenTelemetry's own log records.
+
+    Export failures are logged by the SDK; forwarding those through the
+    same exporter would feed an endless retry loop.
+    """
+
+    def filter(self, record: logging.LogRecord) -> bool:
+        return not record.name.startswith("opentelemetry")
+
+
+class CountingInMemorySessionStore(InMemorySessionStore):
+    """In-memory session store that can report its live session count."""
+
+    async def count(self) -> int:
+        now = self._clock()
+        return sum(
+            1
+            for session_id in self._data
+            if session_id not in self._expires
+            or self._expires[session_id] > now
+        )
+
+
+class CountingRedisSessionStore(RedisSessionStore):
+    """Redis session store that can report its live session count."""
+
+    def __init__(self, client: Redis, prefix: str = SESSION_KEY_PREFIX) -> None:
+        super().__init__(client, prefix=prefix)
+
+    async def count(self) -> int:
+        total = 0
+        async for _ in self._client.scan_iter(match=f"{self._prefix}*", count=500):
+            total += 1
+        return total
+
+
+# The session stores the application builds: both expose ``count()`` on top
+# of kaya's ``SessionStore`` interface. Game stores need no alias — the
+# platform's ``GameStore`` declares ``count_ongoing()`` itself.
+CountableSessionStore = Union[CountingInMemorySessionStore, CountingRedisSessionStore]
+
+
+class TelemetryMixin(KayaMixin):
+    """Application gauges and log export on top of kaya-otel.
+
+    The meter and logger providers are built by
+    :func:`telemetry_from_settings` and injected: kaya-otel only shuts
+    down providers it created itself, so this mixin owns their shutdown.
+    """
+
+    def __init__(
+        self,
+        *,
+        meter_provider: "MeterProvider",
+        logger_provider: Optional["LoggerProvider"],
+        log_handler: Optional[logging.Handler],
+        session_store: CountableSessionStore,
+        game_store: GameStore,
+        poll_interval_seconds: float,
+    ) -> None:
+        self._meter_provider = meter_provider
+        self._logger_provider = logger_provider
+        self._log_handler = log_handler
+        self._session_store = session_store
+        self._game_store = game_store
+        self._poll_interval = poll_interval_seconds
+        self._active_users: Optional[int] = None
+        self._ongoing_games: Optional[int] = None
+        self._poll_task: Optional[asyncio.Task] = None
+        self._instrumentor: Optional[Any] = None
+
+    def apply(self, app: KayaApp) -> None:
+        from opentelemetry.instrumentation.system_metrics import (
+            SystemMetricsInstrumentor,
+        )
+        from opentelemetry.metrics import CallbackOptions, Observation
+
+        instrumentor = SystemMetricsInstrumentor(config=PROCESS_METRICS_CONFIG)
+        self._instrumentor = instrumentor
+        instrumentor.instrument(meter_provider=self._meter_provider)
+
+        meter = self._meter_provider.get_meter(__name__)
+
+        def active_users(options: CallbackOptions):
+            if self._active_users is not None:
+                yield Observation(self._active_users)
+
+        def ongoing_games(options: CallbackOptions):
+            if self._ongoing_games is not None:
+                yield Observation(self._ongoing_games)
+
+        meter.create_observable_gauge(
+            "tavolo.active_users",
+            callbacks=[active_users],
+            unit="{user}",
+            description="Sessions currently held in the session store.",
+        )
+        meter.create_observable_gauge(
+            "tavolo.ongoing_games",
+            callbacks=[ongoing_games],
+            unit="{game}",
+            description="Live games that have not finished yet.",
+        )
+
+    # -- lifecycle ----------------------------------------------------------
+
+    def setup(self, loop: asyncio.AbstractEventLoop) -> None:
+        self._poll_task = loop.create_task(self._poll())
+
+    def shutdown(self, loop: asyncio.AbstractEventLoop) -> None:
+        if self._poll_task is not None:
+            self._poll_task.cancel()
+            self._poll_task = None
+        if self._log_handler is not None:
+            logging.getLogger().removeHandler(self._log_handler)
+            self._log_handler.close()
+            self._log_handler = None
+        if self._logger_provider is not None:
+            self._logger_provider.shutdown()
+        self._meter_provider.shutdown()
+
+    async def _poll(self) -> None:
+        while True:
+            await self.refresh_counts()
+            await asyncio.sleep(self._poll_interval)
+
+    async def refresh_counts(self) -> None:
+        """Refresh the cached store counts (the poll loop calls this)."""
+        try:
+            self._active_users = await self._session_store.count()
+        except Exception:
+            log.warning("active_users refresh failed", exc_info=True)
+        try:
+            self._ongoing_games = await self._game_store.count_ongoing()
+        except Exception:
+            log.warning("ongoing_games refresh failed", exc_info=True)
+
+
+@dataclass
+class Telemetry:
+    """The providers and mixin built from the ``OTEL_*`` settings."""
+
+    meter_provider: "MeterProvider"
+    logger_provider: Optional["LoggerProvider"]
+    log_handler: Optional[logging.Handler]
+    mixin: TelemetryMixin
+
+
+def telemetry_from_settings(
+    settings: Settings,
+    session_store: CountableSessionStore,
+    game_store: GameStore,
+) -> Optional[Telemetry]:
+    """Build the metric/log providers, or ``None`` when telemetry is off.
+
+    Only called when ``OTEL_ENABLED`` is set; the OpenTelemetry imports
+    are lazy so the default install does not need the ``otel`` extra.
+    """
+    if not settings.otel_enabled:
+        return None
+    try:
+        from opentelemetry.exporter.otlp.proto.http._log_exporter import (
+            OTLPLogExporter,
+        )
+        from opentelemetry.exporter.otlp.proto.http.metric_exporter import (
+            OTLPMetricExporter,
+        )
+        from opentelemetry.sdk._logs import LoggerProvider, LoggingHandler
+        from opentelemetry.sdk._logs.export import BatchLogRecordProcessor
+        from opentelemetry.sdk.metrics import MeterProvider
+        from opentelemetry.sdk.metrics.export import PeriodicExportingMetricReader
+        from opentelemetry.sdk.resources import Resource
+    except ImportError as exc:
+        raise RuntimeError(
+            "OTEL_ENABLED is set but the OpenTelemetry dependencies are not "
+            "installed; install tavolo-app with the 'otel' extra"
+        ) from exc
+
+    endpoint = settings.otel_exporter_endpoint
+    headers = otel_headers(settings)
+    resource = Resource.create({"service.name": settings.otel_service_name})
+    meter_provider = MeterProvider(
+        resource=resource,
+        metric_readers=[
+            PeriodicExportingMetricReader(
+                OTLPMetricExporter(
+                    endpoint=_signal_endpoint(endpoint, "v1/metrics"),
+                    headers=headers,
+                ),
+                export_interval_millis=settings.otel_metric_export_interval_millis,
+            )
+        ],
+    )
+    logger_provider = LoggerProvider(resource=resource)
+    logger_provider.add_log_record_processor(
+        BatchLogRecordProcessor(
+            OTLPLogExporter(
+                endpoint=_signal_endpoint(endpoint, "v1/logs"),
+                headers=headers,
+            )
+        )
+    )
+    handler = LoggingHandler(level=logging.NOTSET, logger_provider=logger_provider)
+    handler.addFilter(_NoOtelLogs())
+    logging.getLogger().addHandler(handler)
+    mixin = TelemetryMixin(
+        meter_provider=meter_provider,
+        logger_provider=logger_provider,
+        log_handler=handler,
+        session_store=session_store,
+        game_store=game_store,
+        poll_interval_seconds=settings.otel_metrics_poll_interval_seconds,
+    )
+    return Telemetry(
+        meter_provider=meter_provider,
+        logger_provider=logger_provider,
+        log_handler=handler,
+        mixin=mixin,
+    )
@@ -38,6 +38,7 @@ from __future__ import annotations
 import asyncio
 import contextlib
 import json
+import time
 from abc import ABC, abstractmethod
 from datetime import datetime
 from logging import getLogger
@@ -54,6 +55,10 @@ GAME_KEY_PREFIX = "tavolo:game:"
 CODE_KEY_PREFIX = "tavolo:code:"
 CHANNEL_PREFIX = "tavolo:game:"
 DEADLINES_KEY = "tavolo:deadlines"
+# Sorted set of running games (score = expiry timestamp), maintained by the
+# stores so the telemetry gauge can count ongoing matches without loading
+# every session. Finished games are removed on their final save.
+ONGOING_GAMES_KEY = "tavolo:games:ongoing"

 # Sentinel pushed into in-memory subscriber queues to signal a change.
 _BUMP = b"update"
@@ -120,6 +125,14 @@ class GameStore(ABC):
    async def find_by_code(self, code: str) -> Optional[GameSession]:
        """Return the live session for a join ``code`` or ``None``."""

+    @abstractmethod
+    async def count_ongoing(self) -> int:
+        """Return the number of live games that have not finished yet.
+
+        Lobbies and matches in progress count; matches that are over but
+        still retained for the final-board view do not.
+        """
+
    @abstractmethod
    def lock(self, game_id: str) -> AsyncContextManager[None]:
        """Async context manager serializing mutations of one game."""
@@ -181,9 +194,15 @@ class RedisGameStore(GameStore):

    async def save(self, session: GameSession) -> None:
        payload = json.dumps(session_to_json(session, self._registry))
+        engine = self._registry.require(session.game_type)
        async with self._redis.pipeline(transaction=True) as pipe:
            pipe.set(f"{GAME_KEY_PREFIX}{session.id}", payload, ex=self._ttl)
            pipe.set(f"{CODE_KEY_PREFIX}{session.join_code}", session.id, ex=self._ttl)
+            if engine.is_finished(session):
+                pipe.zrem(ONGOING_GAMES_KEY, session.id)
+            else:
+                pipe.zadd(ONGOING_GAMES_KEY, {session.id: time.time() + self._ttl})
+                pipe.expire(ONGOING_GAMES_KEY, self._ttl)
            await pipe.execute()
        log.debug("redis save %s (game %s, ttl %ds)", session.id, session.game_type, self._ttl)

@@ -195,6 +214,17 @@ class RedisGameStore(GameStore):
            game_id = game_id.decode("utf-8")
        return await self.load(str(game_id))

+    async def count_ongoing(self) -> int:
+        # Prune entries whose game has expired since the last poll, then
+        # count what is left. Both commands run in one transaction so the
+        # count cannot see a half-pruned set.
+        now = time.time()
+        async with self._redis.pipeline(transaction=True) as pipe:
+            pipe.zremrangebyscore(ONGOING_GAMES_KEY, "-inf", now)
+            pipe.zcard(ONGOING_GAMES_KEY)
+            _, count = await pipe.execute()
+        return int(count)
+
    @contextlib.asynccontextmanager
    async def subscribe(self, game_id: str) -> AsyncIterator[AsyncIterator[None]]:
        pubsub = self._redis.pubsub()
@@ -243,6 +273,7 @@ class InMemoryGameStore(GameStore):
        self._registry = registry
        self._games: Dict[str, str] = {}
        self._codes: Dict[str, str] = {}
+        self._ongoing: Set[str] = set()
        self._locks: Dict[str, asyncio.Lock] = {}
        self._subscribers: Dict[str, Set[asyncio.Queue]] = {}
        self._deadlines: Dict[str, float] = {}
@@ -268,6 +299,11 @@ class InMemoryGameStore(GameStore):
    async def save(self, session: GameSession) -> None:
        self._games[session.id] = json.dumps(session_to_json(session, self._registry))
        self._codes[session.join_code] = session.id
+        engine = self._registry.require(session.game_type)
+        if engine.is_finished(session):
+            self._ongoing.discard(session.id)
+        else:
+            self._ongoing.add(session.id)

    async def find_by_code(self, code: str) -> Optional[GameSession]:
        game_id = self._codes.get(code.upper())
@@ -275,6 +311,9 @@ class InMemoryGameStore(GameStore):
            return None
        return await self.load(game_id)

+    async def count_ongoing(self) -> int:
+        return len(self._ongoing)
+
    @contextlib.asynccontextmanager
    async def subscribe(self, game_id: str) -> AsyncIterator[AsyncIterator[None]]:
        queue: asyncio.Queue = asyncio.Queue()
@@ -2,11 +2,16 @@
 from __future__ import annotations

 import asyncio
+import time
 import unittest

 from tavolo.platform import GameSession, Seat
 from tavolo.platform.registry import GameRegistry
-from tavolo.platform.store import InMemoryGameStore
+from tavolo.platform.store import (
+    ONGOING_GAMES_KEY,
+    InMemoryGameStore,
+    RedisGameStore,
+)
 from helpers import DummyEngine, async_test


@@ -27,6 +32,81 @@ def _session(game_id: str = "g1", code: str = "CODE01") -> GameSession:
    return session


+class _FakePipeline:
+    """Records commands and applies them to the fake Redis on execute."""
+
+    def __init__(self, redis: "_FakeRedis") -> None:
+        self._redis = redis
+        self._ops: list = []
+
+    async def __aenter__(self) -> "_FakePipeline":
+        return self
+
+    async def __aexit__(self, *exc: object) -> bool:
+        return False
+
+    def set(self, key: str, value: str, ex: object = None) -> None:
+        self._ops.append(("set", key))
+
+    def zadd(self, key: str, mapping: dict) -> None:
+        self._ops.append(("zadd", key, mapping))
+
+    def zrem(self, key: str, member: str) -> None:
+        self._ops.append(("zrem", key, member))
+
+    def expire(self, key: str, ttl: int) -> None:
+        self._ops.append(("expire", key))
+
+    def zremrangebyscore(self, key: str, minimum: object, maximum: object) -> None:
+        self._ops.append(("zremrangebyscore", key))
+
+    def zcard(self, key: str) -> None:
+        self._ops.append(("zcard", key))
+
+    async def execute(self) -> list:
+        results = [self._redis.apply(op) for op in self._ops]
+        self._ops = []
+        return results
+
+
+class _FakeRedis:
+    """The slice of the Redis client the game store uses."""
+
+    def __init__(self) -> None:
+        self.keys: dict = {}
+        self.zsets: dict = {}
+
+    def pipeline(self, transaction: bool = True) -> _FakePipeline:
+        return _FakePipeline(self)
+
+    def apply(self, op: tuple) -> object:
+        name, *args = op
+        if name == "set":
+            self.keys[args[0]] = "payload"
+            return True
+        if name == "zadd":
+            self.zsets.setdefault(args[0], {}).update(args[1])
+            return 1
+        if name == "zrem":
+            self.zsets.get(args[0], {}).pop(args[1], None)
+            return 1
+        if name == "expire":
+            return True
+        if name == "zremrangebyscore":
+            zset = self.zsets.get(args[0], {})
+            for member in [
+                member
+                for member, score in zset.items()
+                if score <= time.time()
+            ]:
+                del zset[member]
+            return 0
+        if name == "zcard":
+            return len(self.zsets.get(args[0], {}))
+        raise AssertionError(f"unexpected command: {name}")
+
+
+
 class InMemoryGameStoreTest(unittest.TestCase):
    @async_test
    async def test_save_load_roundtrip(self) -> None:
@@ -114,6 +194,28 @@ class InMemoryGameStoreTest(unittest.TestCase):
            ["holder-enter", "holder-exit", "contender"], order
        )

+    @async_test
+    async def test_count_ongoing_tracks_unfinished_games(self) -> None:
+        store = InMemoryGameStore(_registry())
+        self.assertEqual(0, await store.count_ongoing())
+
+        session = _session()
+        await store.save(session)
+        self.assertEqual(1, await store.count_ongoing())
+
+        other = _session(game_id="g2", code="CODE02")
+        await store.save(other)
+        self.assertEqual(2, await store.count_ongoing())
+
+        # A finished match is retained for the final board but no longer
+        # counts as ongoing.
+        session.state["finished"] = True
+        await store.save(session)
+        self.assertEqual(1, await store.count_ongoing())
+        other.state["finished"] = True
+        await store.save(other)
+        self.assertEqual(0, await store.count_ongoing())
+
    @async_test
    async def test_deadline_queue(self) -> None:
        store = InMemoryGameStore(_registry())
@@ -138,5 +240,35 @@ class InMemoryGameStoreTest(unittest.TestCase):
        self.assertEqual(["b"], await store.due_deadlines(now=100.0))


+class RedisGameStoreTest(unittest.TestCase):
+    @async_test
+    async def test_save_marks_games_ongoing_and_finished_ones_stop_counting(self) -> None:
+        redis = _FakeRedis()
+        store = RedisGameStore(redis, _registry(), ttl_seconds=3600)  # type: ignore[arg-type]
+        session = _session()
+
+        await store.save(session)
+        self.assertEqual(1, await store.count_ongoing())
+        score = redis.zsets[ONGOING_GAMES_KEY][session.id]
+        self.assertGreater(score, time.time())
+        self.assertLessEqual(score, time.time() + 3600)
+
+        session.state["finished"] = True
+        await store.save(session)
+        self.assertNotIn(session.id, redis.zsets[ONGOING_GAMES_KEY])
+        self.assertEqual(0, await store.count_ongoing())
+
+    @async_test
+    async def test_count_ongoing_prunes_expired_entries(self) -> None:
+        redis = _FakeRedis()
+        redis.zsets[ONGOING_GAMES_KEY] = {
+            "stale": time.time() - 10,
+            "live": time.time() + 100,
+        }
+        store = RedisGameStore(redis, _registry())  # type: ignore[arg-type]
+        self.assertEqual(1, await store.count_ongoing())
+        self.assertNotIn("stale", redis.zsets[ONGOING_GAMES_KEY])
+
+
 if __name__ == "__main__":
    unittest.main()