Compare commits

..
Author SHA1 Message Date
woggioni b5219062f2 Let anyone with the game URL watch as a read-only spectator
CI / Build and push docker image (push) Successful in 3m0s
Anonymous users and logged-in non-players can now open /game/<id> and
follow the match: they see the table, the score and each player's
captured cards and scope, but never a hand and never the join code.
Spectator actions are rejected with a forbidden error.

Backend (tavolo-platform / tavolo-scopone):
- /ws/games/{id} accepts anonymous and non-seated connections as
  spectators; only unknown games are refused. The state payload carries
  spectator: true and withholds join_code unless the viewer is seated.
- GET /api/games/{id} is public and returns the same personalized
  (player) or read-only (spectator) view.
- GameEngine.view_for/game_over_view take Optional[str] (None = spectator)
  and are documented to hide privileged information for non-seated
  viewers.

Frontend (Sycamore):
- Parse the spectator flag and render a spectator table: card backs plus
  captured/scope for every seat, no clickable hand, "Spectating" badge.
- A pre-start spectator sees a "game not started yet" waiting room with
  the watch link and no join code.
- "Copy watch link" buttons in the waiting room and the game HUD.
- The hand-end summary carries no acknowledge button for spectators and
  the game-over overlay hides "My matches".

Tests and docs updated.
2026-10-06 12:51:53 +08:00
woggioni adc3d340b1 Upgrade kaya to 0.0.7 2026-10-01 15:39:32 +00:00
woggioni 915a294dc8 Support multiple OIDC providers
kaya-oidc 0.0.6 can run several providers in one app (OIDCProviders),
each with its own login route and namespaced session state. Adopt it:
providers are configured in indexed, contiguous OIDC_<n>_* blocks
(0-based) with OIDC_<n>_PROVIDER naming the provider in its login URL
and session keys. The old unindexed OIDC_ISSUER/OIDC_CLIENT_ID/... names
are no longer read; startup fails with a rename hint when they are set.
With no OIDC variables at all, the single dev mock provider named
"default" is used, as before.

The platform now authenticates through OIDCProviders: auth.current_user
returns the session's user (sessions stay single-identity, so at most one
provider is logged in) and require_auth/route handlers use it, while the
websocket path keeps scanning for any provider's user. New routes expose
the configuration to the SPA and keep old links working: GET
/api/auth/providers lists the login URLs, GET /auth/login redirects to
the first provider and GET /auth/logout to whichever provider the
session is logged in with. The lobby renders one login button per
provider from that endpoint.

Update the env examples, compose and k8s manifests to the indexed scheme,
document the multiple-provider setup and the migration, and cover it with
config, auth and route tests.
2026-10-01 12:38:29 +00:00
woggioni 3abbbc6b68 Upgrade kaya to 0.0.6
kaya-oidc 0.0.6 namespaces its session state per provider as
oidc:{provider}:user (the default provider is "default", so the key is
oidc:default:user). The unprefixed oidc_user key is no longer written:
left unadapted, every websocket upgrade would be rejected with 401 and
traced requests would silently lose their user.id attribute. Adopt it
with a session_user helper on tavolo.platform.auth that finds any
provider's user, used by both get_ws_user and telemetry's
session_tag_hook. Websocket and route handlers keep their signatures.

kaya-otel 0.0.6 only enriches websocket spans with standard URL, network
and peer attributes (plus error.type), which needs no application change.

Regenerate the lock: the eight kaya pins move to 0.0.6 and, the wheels'
declared metadata being unchanged, nothing else moves. tavolo-platform
now floors kaya-core>=0.0.6 (it reads the new session layout) and
tavolo-app floors kaya-otel>=0.0.6.

As kaya documents, the session-key move signs existing logged-in users
out once; they simply log in again.
2026-10-01 01:18:05 +00:00
woggioni b43e074728 Wake deadline consumers on enqueue instead of polling every second 2026-09-29 23:39:42 +00:00
woggioni e344ce2e30 Sort the player's hand by rank then suit 2026-09-29 16:57:48 +00:00
woggioni e9909cb317 Upgrade kaya to 0.0.5
kaya 0.0.5 surfaces a vanished websocket peer as
kaya.core.WebSocketDisconnect instead of leaking granian's
RSGIProtocolClosed: receive() reports a close message so iteration ends
normally, sends raise the new exception, and KayaApp.handle_websocket
swallows it as normal termination. Adopt it: drop the granian-specific
workaround from the platform websocket handler, catch
WebSocketDisconnect in the send helper, and model the new semantics in
the abrupt-disconnect test (with coverage that a send on a vanished
peer is ignored).

Regenerate the lock: the eight kaya pins move to 0.0.5 (declared
metadata is unchanged) and pip-compile also bumps PyJWT to 2.15.1, a
security release with no breaking changes for kaya-oidc's jwt.decode
usage. tavolo-platform now floors kaya-core>=0.0.5 (it imports
WebSocketDisconnect) and tavolo-app floors kaya-otel>=0.0.5.
2026-09-29 15:43:19 +00:00
woggioni 2910fc4600 Refuse websocket upgrades with HTTP statuses, not close codes
A close() before accept() carries the HTTP status of the refused
upgrade, not a WebSocket close code: RSGI (granian) uses the value
verbatim as the response status, so 4401 panicked on the invalid status
(InvalidStatusCode) and — granian being built with panic = "abort" —
aborted the worker and took the whole app down. ASGI servers answer 403
regardless, and browsers report a refused handshake as an abnormal close
(1006), so the 4401/4403/4404 codes never reached clients anyway.

Reject 401/404/403 to mirror GET /api/games/{id}, and classify a failed
first connection in the game page through the REST API so players still
see "session expired" / "game no longer available" instead of a retry
loop. Add a regression test asserting every pre-accept rejection uses a
status in the HTTP range.
2026-09-29 22:31:41 +08:00
woggioni 21911caf9e Poll deadlines with ZRANGE BYSCORE instead of the deprecated ZRANGEBYSCORE 2026-09-28 05:50:54 +08:00
woggioni 08f158ef28 Fetch match-history participants in one query per page
CI / Build and push docker image (push) Successful in 53s
2026-09-28 05:21:34 +08:00
woggioni 834f1f26a9 Tag traced requests with the session id and OIDC sub 2026-09-28 05:11:22 +08:00
woggioni bcd1371019 Cap urllib3.connectionpool at INFO in the default logging config
The root logger runs at DEBUG, so urllib3.connectionpool emitted a
record per pooled HTTP request. Keep its INFO+ output while dropping
the per-connection chatter.
2026-09-28 04:38:28 +08:00
woggioni 3cf7e2e257 Show match finish times in the browser timezone
CI / Build and push docker image (push) Successful in 2m18s
The history table truncated finished_at, which the API serializes in
UTC, so the Finished column always showed UTC wall time. Parse the
timestamp with the browser's Date and render its local components,
keeping the existing YYYY-MM-DD HH:MM layout.
2026-09-28 04:26:07 +08:00
woggioni 2bdf247cb3 Restore cursor value types before keyset filtering
Cursors are JSON, so a datetime sort key reached the next request as a
string. Tortoise's DatetimeField.to_db_value does not parse strings, so
asyncpg rejected the string where a timestamptz parameter was expected;
sqlite tolerated it, hiding the bug in tests. Coerce decoded cursor
values through their model field's to_python_value before building the
keyset filter.
2026-09-28 04:26:00 +08:00
woggioni a3e8ff3206 Trace Postgres and Redis queries with the contrib instrumentations
CI / Build and push docker image (push) Successful in 2m42s
Builds the TracerProvider in tavolo.telemetry (kaya-otel's is private) and
injects it into kaya-otel, then instruments asyncpg and redis through
opentelemetry-instrumentation-asyncpg/redis. Redis command values are
redacted and asyncpg spans carry the parameterized SQL only.
2026-09-28 03:38:59 +08:00
woggioni 98d4485375 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
woggioni 5be4926885 Bundle kaya-otel in the Docker image
CI / Build and push docker image (push) Successful in 1m57s
The tavolo-app otel extra was never installed: requirements.txt was
locked without it and the Dockerfile installs the local distributions
with --no-deps, so any container with OTEL_ENABLED set died at import
with "kaya-otel is not installed".

Add the extra's full closure (kaya-otel plus its OpenTelemetry/protobuf
dependencies) to the pinned lock and refresh the comments that claimed
the extra had to be added separately.
2026-09-26 15:47:16 +08:00
59 changed files with 3245 additions and 374 deletions
+15 -2
View File
@@ -52,6 +52,15 @@ player does not move — disconnected or fallen asleep — the server plays a
random legal card for them (randomizing among the legal captures when the random legal card for them (randomizing among the legal captures when the
rules require a capture), so one absent player cannot stall the table. rules require a capture), so one absent player cannot stall the table.
## Watching
Anyone who has the game URL (`/game/<id>`) can watch, logged in or not:
spectators see the table, the score and every player's captured cards and
scope, but never a hand and never the join code. The "Copy watch link"
button in the waiting room and in the game HUD copies the URL to share.
A spectator who opens the URL before the match starts sees a "game not
started yet" waiting room.
## Development ## Development
Backend (from `server/`): Backend (from `server/`):
@@ -78,8 +87,12 @@ Trunk proxies `/api`, `/auth` and `/ws` to `127.0.0.1:8080` (see
run the backend with: run the backend with:
```sh ```sh
OIDC_POST_LOGIN_REDIRECT=http://localhost:8000/ \ OIDC_0_PROVIDER=default \
OIDC_POST_LOGOUT_REDIRECT=http://localhost:8000/ \ OIDC_0_ISSUER=http://localhost:8180/tavolo \
OIDC_0_CLIENT_ID=tavolo \
OIDC_0_REDIRECT_URI=http://localhost:8080/auth/default/callback \
OIDC_0_POST_LOGIN_REDIRECT=http://localhost:8000/ \
OIDC_0_POST_LOGOUT_REDIRECT=http://localhost:8000/ \
.venv/bin/granian --host 127.0.0.1 --port 8080 tavolo.app:app .venv/bin/granian --host 127.0.0.1 --port 8080 tavolo.app:app
``` ```
+22 -12
View File
@@ -21,7 +21,7 @@
# - This file deploys a ClusterIP Service only. Until you add an Ingress, # - This file deploys a ClusterIP Service only. Until you add an Ingress,
# reach the app with: # reach the app with:
# kubectl port-forward -n tavolo svc/tavolo 8080:80 # kubectl port-forward -n tavolo svc/tavolo 8080:80
# and set OIDC_REDIRECT_URI to match whatever URL the browser uses. # and set OIDC_0_REDIRECT_URI to match whatever URL the browser uses.
--- ---
apiVersion: v1 apiVersion: v1
@@ -71,22 +71,32 @@ data:
# CORS_ALLOW_CREDENTIALS: "false" # CORS_ALLOW_CREDENTIALS: "false"
# CORS_EXPOSE_HEADERS: "" # CORS_EXPOSE_HEADERS: ""
# CORS_MAX_AGE: "600" # CORS_MAX_AGE: "600"
# OpenTelemetry (kaya-otel): traces + metrics via OTLP/HTTP, disabled # OpenTelemetry (kaya-otel + contrib instrumentations): traces
# unless OTEL_ENABLED is set. Requires the otel extra in the image. # (HTTP/WebSocket, Postgres, Redis), metrics and logs via OTLP/HTTP,
# disabled unless OTEL_ENABLED is set. The image bundles the otel extra.
# OTEL_ENABLED: "true" # OTEL_ENABLED: "true"
# OTEL_SERVICE_NAME: "tavolo" # OTEL_SERVICE_NAME: "tavolo"
# OTEL_EXPORTER_OTLP_ENDPOINT: "http://otel-collector.observability:4318" # OTEL_EXPORTER_OTLP_ENDPOINT: "http://otel-collector.observability:4318"
# OTEL_EXPORTER_OTLP_HEADERS: "Authorization=Bearer ..." # OTEL_EXPORTER_OTLP_HEADERS: "Authorization=Bearer ..."
# OTEL_EXCLUDED_PATHS: "/api/health" # default; paths skipped by tracing # OTEL_EXCLUDED_PATHS: "/api/health" # default; paths skipped by tracing
# OIDC (provider lives in another namespace). # OTEL_METRIC_EXPORT_INTERVAL_MILLIS: "60000"
OIDC_CLIENT_ID: tavolo # OTEL_METRICS_POLL_INTERVAL_SECONDS: "15"
OIDC_POST_LOGIN_REDIRECT: / # OIDC (the provider lives in another namespace). Each provider is an
OIDC_POST_LOGOUT_REDIRECT: / # indexed block: OIDC_<n>_PROVIDER names it in the login URL
# (/auth/<provider>/login) and indices must be contiguous from 0. Add
# OIDC_1_*, OIDC_2_*, ... blocks for more providers; add their secrets
# to tavolo-secrets as well.
OIDC_0_PROVIDER: default
OIDC_0_CLIENT_ID: tavolo
OIDC_0_POST_LOGIN_REDIRECT: /
OIDC_0_POST_LOGOUT_REDIRECT: /
# Issuer URL as seen by the browser AND the pods (see notes above). # Issuer URL as seen by the browser AND the pods (see notes above).
OIDC_ISSUER: REPLACE_ME OIDC_0_ISSUER: REPLACE_ME
# Public callback URL of this deployment, e.g. # Public callback URL of this deployment, e.g.
# http://<host>:<port>/auth/callback (must be allowed at the provider). # http://<host>:<port>/auth/default/callback (must be allowed at the
OIDC_REDIRECT_URI: REPLACE_ME # provider). When unset it defaults to
# http://localhost:8080/auth/<provider>/callback.
OIDC_0_REDIRECT_URI: REPLACE_ME
--- ---
apiVersion: v1 apiVersion: v1
@@ -103,8 +113,8 @@ stringData:
# The only Postgres secret: the password for DATABASE_USER at # The only Postgres secret: the password for DATABASE_USER at
# DATABASE_HOST (both configured in the tavolo-config ConfigMap). # DATABASE_HOST (both configured in the tavolo-config ConfigMap).
DATABASE_PASSWORD: REPLACE_ME DATABASE_PASSWORD: REPLACE_ME
# Client secret for OIDC_CLIENT_ID at the provider. # Client secret for OIDC_0_CLIENT_ID at the provider.
OIDC_CLIENT_SECRET: REPLACE_ME OIDC_0_CLIENT_SECRET: REPLACE_ME
--- ---
# Redis: sessions + live game state. Ephemeral by design (emptyDir): worst # Redis: sessions + live game state. Ephemeral by design (emptyDir): worst
+13 -8
View File
@@ -103,13 +103,15 @@ services:
DATABASE_USER: tavolo DATABASE_USER: tavolo
DATABASE_PASSWORD: ${DATABASE_PASSWORD:-password} DATABASE_PASSWORD: ${DATABASE_PASSWORD:-password}
# By default the app and browsers reach the mock IdP under the same # By default the app and browsers reach the mock IdP under the same
# name (see README /etc/hosts note); override OIDC_ISSUER and # name (see README /etc/hosts note); override the OIDC_0_* values to
# OIDC_REDIRECT_URI to use a real provider or a different host port. # use a real provider or a different host port. Additional providers
OIDC_ISSUER: ${OIDC_ISSUER:-http://mockoauth:8180/tavolo} # take OIDC_1_*, OIDC_2_*, ... blocks (indices are contiguous from 0).
OIDC_0_PROVIDER: ${OIDC_0_PROVIDER:-default}
OIDC_0_ISSUER: ${OIDC_0_ISSUER:-http://mockoauth:8180/tavolo}
# The mock OIDC server does not validate clients: any id/secret works. # The mock OIDC server does not validate clients: any id/secret works.
OIDC_CLIENT_ID: tavolo OIDC_0_CLIENT_ID: tavolo
OIDC_CLIENT_SECRET: ${OIDC_CLIENT_SECRET:-dev-secret} OIDC_0_CLIENT_SECRET: ${OIDC_0_CLIENT_SECRET:-dev-secret}
OIDC_REDIRECT_URI: ${OIDC_REDIRECT_URI:-http://localhost:8080/auth/callback} OIDC_0_REDIRECT_URI: ${OIDC_0_REDIRECT_URI:-http://localhost:8080/auth/default/callback}
REDIS_URL: redis://redis:6379/0 REDIS_URL: redis://redis:6379/0
HAND_ACK_TIMEOUT_SECONDS: ${HAND_ACK_TIMEOUT_SECONDS:-30} HAND_ACK_TIMEOUT_SECONDS: ${HAND_ACK_TIMEOUT_SECONDS:-30}
TURN_TIMEOUT_SECONDS: ${TURN_TIMEOUT_SECONDS:-30} TURN_TIMEOUT_SECONDS: ${TURN_TIMEOUT_SECONDS:-30}
@@ -122,8 +124,11 @@ services:
CORS_ALLOW_CREDENTIALS: ${CORS_ALLOW_CREDENTIALS:-} CORS_ALLOW_CREDENTIALS: ${CORS_ALLOW_CREDENTIALS:-}
CORS_EXPOSE_HEADERS: ${CORS_EXPOSE_HEADERS:-} CORS_EXPOSE_HEADERS: ${CORS_EXPOSE_HEADERS:-}
CORS_MAX_AGE: ${CORS_MAX_AGE:-} CORS_MAX_AGE: ${CORS_MAX_AGE:-}
# OpenTelemetry (kaya-otel): disabled unless OTEL_ENABLED is set. # OpenTelemetry (kaya-otel + contrib instrumentations): traces
# Requires the otel extra in the image (see server/pyproject.toml). # (HTTP/WebSocket, Postgres, Redis), metrics and logs via OTLP/HTTP,
# disabled unless OTEL_ENABLED is set.
# The image bundles the otel extra (see packages/tavolo-app/pyproject.toml
# and server/requirements.txt).
OTEL_ENABLED: ${OTEL_ENABLED:-} OTEL_ENABLED: ${OTEL_ENABLED:-}
OTEL_SERVICE_NAME: ${OTEL_SERVICE_NAME:-} OTEL_SERVICE_NAME: ${OTEL_SERVICE_NAME:-}
OTEL_EXPORTER_OTLP_ENDPOINT: ${OTEL_EXPORTER_OTLP_ENDPOINT:-} OTEL_EXPORTER_OTLP_ENDPOINT: ${OTEL_EXPORTER_OTLP_ENDPOINT:-}
+40 -6
View File
@@ -13,12 +13,29 @@ DATABASE_OPTIONS=
# test suite (sqlite://:memory:) and handy for managed-DB URLs. # test suite (sqlite://:memory:) and handy for managed-DB URLs.
#DATABASE_URL=postgres://tavolo:password@localhost:5432/tavolo #DATABASE_URL=postgres://tavolo:password@localhost:5432/tavolo
# OIDC (mock-oauth2-server in dev; it does not validate clients, so any # OIDC providers. Each provider is an indexed block: OIDC_<n>_PROVIDER
# client id/secret works. For a real IdP like Keycloak, use its values here.) # (required; names the provider in the login URL /auth/<provider>/login and
OIDC_ISSUER=http://localhost:8180/tavolo # in its session state), OIDC_<n>_ISSUER, OIDC_<n>_CLIENT_ID,
OIDC_CLIENT_ID=tavolo # OIDC_<n>_CLIENT_SECRET (optional: leave unset for public clients),
OIDC_CLIENT_SECRET=dev-secret # OIDC_<n>_REDIRECT_URI (defaults to
OIDC_REDIRECT_URI=http://localhost:8080/auth/callback # http://localhost:8080/auth/<provider>/callback), OIDC_<n>_POST_LOGIN_REDIRECT
# and OIDC_<n>_POST_LOGOUT_REDIRECT (default /). Indices must be contiguous
# from 0; add a second provider with an OIDC_1_* block (and so on).
#
# The values below target the dev mock-oauth2-server (it does not validate
# clients, so any client id/secret works). For a real IdP like Keycloak,
# use its values and register the redirect URI there.
OIDC_0_PROVIDER=default
OIDC_0_ISSUER=http://localhost:8180/tavolo
OIDC_0_CLIENT_ID=tavolo
OIDC_0_CLIENT_SECRET=dev-secret
OIDC_0_REDIRECT_URI=http://localhost:8080/auth/default/callback
# A second provider, e.g. Google:
#OIDC_1_PROVIDER=google
#OIDC_1_ISSUER=https://accounts.google.com
#OIDC_1_CLIENT_ID=...
#OIDC_1_CLIENT_SECRET=...
#OIDC_1_REDIRECT_URI=http://localhost:8080/auth/google/callback
# Redis (session and live-game storage). Unset to fall back to in-memory # Redis (session and live-game storage). Unset to fall back to in-memory
# stores (only sensible for local development with a single worker). # stores (only sensible for local development with a single worker).
@@ -61,6 +78,23 @@ TURN_TIMEOUT_SECONDS=30
# Seconds browsers may cache the preflight response (default 600): # Seconds browsers may cache the preflight response (default 600):
#CORS_MAX_AGE=600 #CORS_MAX_AGE=600
# OpenTelemetry (via kaya-otel + the contrib instrumentation packages).
# Off unless OTEL_ENABLED is truthy; exports traces (HTTP/WebSocket,
# Postgres via asyncpg, Redis), metrics and logs over OTLP/HTTP (requires
# the tavolo-app "otel" extra, bundled in the Docker image). The endpoint
# defaults to the standard OTEL_EXPORTER_OTLP_ENDPOINT (localhost:4318)
# when unset.
#OTEL_ENABLED=true
#OTEL_SERVICE_NAME=tavolo
#OTEL_EXPORTER_OTLP_ENDPOINT=http://otel-collector:4318
#OTEL_EXPORTER_OTLP_HEADERS=Authorization=Bearer token,X-Tenant=one
# Paths excluded from tracing and metrics (default /api/health).
#OTEL_EXCLUDED_PATHS=/api/health
# Metric export interval (ms, default 60000) and session/game count poll
# interval (seconds, default 15).
#OTEL_METRIC_EXPORT_INTERVAL_MILLIS=60000
#OTEL_METRICS_POLL_INTERVAL_SECONDS=15
# App server # App server
APP_HOST=0.0.0.0 APP_HOST=0.0.0.0
APP_PORT=8080 APP_PORT=8080
+2 -1
View File
@@ -60,7 +60,8 @@ RUN --mount=type=cache,target=/root/.cache/pip \
# The three sibling distributions, installed from the local tree (PEP 420 # The three sibling distributions, installed from the local tree (PEP 420
# namespace packages sharing the `tavolo` namespace). --no-deps: every # namespace packages sharing the `tavolo` namespace). --no-deps: every
# third-party dependency is already pinned in requirements.txt. # third-party dependency — including the tavolo-app `otel` extra — is
# already pinned in requirements.txt.
COPY server/packages/ ./packages/ COPY server/packages/ ./packages/
RUN --mount=type=cache,target=/root/.cache/pip \ RUN --mount=type=cache,target=/root/.cache/pip \
/opt/venv/bin/pip install --no-deps ./packages/tavolo-platform ./packages/tavolo-scopone ./packages/tavolo-app /opt/venv/bin/pip install --no-deps ./packages/tavolo-platform ./packages/tavolo-scopone ./packages/tavolo-app
+103 -15
View File
@@ -43,7 +43,8 @@ Because the browser and the app both talk to the OIDC issuer at
echo "127.0.0.1 mockoauth" | sudo tee -a /etc/hosts echo "127.0.0.1 mockoauth" | sudo tee -a /etc/hosts
``` ```
Then log in: `GET http://localhost:8080/auth/login`, type any username Then log in: `GET http://localhost:8080/auth/login` (a provider-agnostic
alias that redirects to the first configured provider), type any username
(e.g. `alice`) in the interactive login, and you are redirected back with a (e.g. `alice`) in the interactive login, and you are redirected back with a
session cookie. That cookie authenticates both the REST API and the session cookie. That cookie authenticates both the REST API and the
WebSocket endpoint. WebSocket endpoint.
@@ -65,14 +66,17 @@ All configuration comes from environment variables (see `.env.example`):
| `DATABASE_OPTIONS` | unset | Extra DSN query parameters, e.g. `ssl=require` | | `DATABASE_OPTIONS` | unset | Extra DSN query parameters, e.g. `ssl=require` |
| `DATABASE_URL` | unset | Full-DSN override; when set, the `DATABASE_*` parts above are ignored (used for sqlite in tests and for managed-DB URLs) | | `DATABASE_URL` | unset | Full-DSN override; when set, the `DATABASE_*` parts above are ignored (used for sqlite in tests and for managed-DB URLs) |
| `REDIS_URL` | unset | Redis DSN for sessions + live games. Unset falls back to in-memory stores | | `REDIS_URL` | unset | Redis DSN for sessions + live games. Unset falls back to in-memory stores |
| `OIDC_ISSUER` | `http://localhost:8180/tavolo` | OIDC issuer URL | | `OIDC_<n>_PROVIDER` | `default` ¹ | Provider name (required when OIDC is configured): used in the login URL `/auth/<provider>/login` and to namespace its session keys. Unique, `[A-Za-z0-9_-]+` |
| `OIDC_CLIENT_ID` | `tavolo` | OIDC client id | | `OIDC_<n>_ISSUER` | `http://localhost:8180/tavolo` ¹ | OIDC issuer URL |
| `OIDC_CLIENT_SECRET` | unset | OIDC client secret | | `OIDC_<n>_CLIENT_ID` | `tavolo` ¹ | OIDC client id |
| `OIDC_REDIRECT_URI` | `http://localhost:8080/auth/callback` | Login callback URL | | `OIDC_<n>_CLIENT_SECRET` | unset | OIDC client secret; leave unset for public clients |
| `OIDC_<n>_REDIRECT_URI` | `http://localhost:8080/auth/<provider>/callback` | Login callback URL; must be registered at the provider |
| `OIDC_<n>_POST_LOGIN_REDIRECT` | `/` | Where the browser is sent after login |
| `OIDC_<n>_POST_LOGOUT_REDIRECT` | `/` | Where the browser is sent after logout |
| `GAME_TTL_SECONDS` | `86400` | Sliding TTL of a live game in Redis | | `GAME_TTL_SECONDS` | `86400` | Sliding TTL of a live game in Redis |
| `HAND_ACK_TIMEOUT_SECONDS` | `30` | Seconds the scopone between-hands scoring summary waits for acknowledgements (wired into the `ScoponeEngine`) | | `HAND_ACK_TIMEOUT_SECONDS` | `30` | Seconds the scopone between-hands scoring summary waits for acknowledgements (wired into the `ScoponeEngine`) |
| `TURN_TIMEOUT_SECONDS` | `30` | Seconds a scopone player has to play before the server plays a random legal card for them (wired into the `ScoponeEngine`) | | `TURN_TIMEOUT_SECONDS` | `30` | Seconds a scopone player has to play before the server plays a random legal card for them (wired into the `ScoponeEngine`) |
| `DEADLINE_HEARTBEAT_MS` | `1000` | Upper bound on the deadline consumer's poll interval (locally enqueued deadlines fire on time regardless) | | `DEADLINE_HEARTBEAT_MS` | `30000` | Backstop poll interval of the deadline consumer; enqueued deadlines are announced over pub/sub and wake consumers immediately, so this only bounds recovery from lost notifications or a dead worker |
| `LOGGING_CONFIG` | unset | Path to a YAML logging configuration file (see below). Unset logs DEBUG to the console | | `LOGGING_CONFIG` | unset | Path to a YAML logging configuration file (see below). Unset logs DEBUG to the console |
| `CORS_ALLOW_ORIGINS` | unset | Comma-separated origins allowed for cross-origin requests, or `*` for any. CORS is disabled unless this or `CORS_ALLOW_ORIGIN_REGEX` is set | | `CORS_ALLOW_ORIGINS` | unset | Comma-separated origins allowed for cross-origin requests, or `*` for any. CORS is disabled unless this or `CORS_ALLOW_ORIGIN_REGEX` is set |
| `CORS_ALLOW_ORIGIN_REGEX` | unset | Regex (fullmatch) additionally matched against request origins, e.g. `https://tavolo-[a-z0-9-]+\.vercel\.app` | | `CORS_ALLOW_ORIGIN_REGEX` | unset | Regex (fullmatch) additionally matched against request origins, e.g. `https://tavolo-[a-z0-9-]+\.vercel\.app` |
@@ -81,13 +85,53 @@ All configuration comes from environment variables (see `.env.example`):
| `CORS_ALLOW_CREDENTIALS` | `false` | `1`/`true`/`yes`/`on` allow cookies/credentials on cross-origin requests | | `CORS_ALLOW_CREDENTIALS` | `false` | `1`/`true`/`yes`/`on` allow cookies/credentials on cross-origin requests |
| `CORS_EXPOSE_HEADERS` | unset | Comma-separated response headers exposed to the browser | | `CORS_EXPOSE_HEADERS` | unset | Comma-separated response headers exposed to the browser |
| `CORS_MAX_AGE` | `600` | Seconds browsers may cache the preflight response | | `CORS_MAX_AGE` | `600` | Seconds browsers may cache the preflight response |
| `OTEL_ENABLED` | `false` | `1`/`true`/`yes`/`on` enable OpenTelemetry traces and metrics (requires the `otel` extra, i.e. `pip install tavolo-app[otel]`) | | `OTEL_ENABLED` | `false` | `1`/`true`/`yes`/`on` enable OpenTelemetry traces, metrics and logs (the Docker image bundles the `otel` extra; non-image installs need `pip install tavolo-app[otel]`) |
| `OTEL_SERVICE_NAME` | `tavolo` | `service.name` resource attribute of the exported telemetry | | `OTEL_SERVICE_NAME` | `tavolo` | `service.name` resource attribute of the exported telemetry |
| `OTEL_EXPORTER_OTLP_ENDPOINT` | unset | Base URL of an OTLP/HTTP collector (e.g. `http://localhost:4318`); unset uses the exporter default | | `OTEL_EXPORTER_OTLP_ENDPOINT` | unset | Base URL of an OTLP/HTTP collector (e.g. `http://localhost:4318`); unset uses the exporter default |
| `OTEL_EXPORTER_OTLP_HEADERS` | unset | Comma-separated `key=value` headers sent to the collector (e.g. authentication) | | `OTEL_EXPORTER_OTLP_HEADERS` | unset | Comma-separated `key=value` headers sent to the collector (e.g. authentication) |
| `OTEL_EXCLUDED_PATHS` | `/api/health` | Comma-separated paths excluded from tracing and metrics (exact matches) | | `OTEL_EXCLUDED_PATHS` | `/api/health` | Comma-separated paths excluded from tracing and metrics (exact matches) |
| `OTEL_METRIC_EXPORT_INTERVAL_MILLIS` | `60000` | How often the OTLP metric reader exports |
| `OTEL_METRICS_POLL_INTERVAL_SECONDS` | `15` | How often the session/game count gauges are refreshed |
| `APP_HOST` / `APP_PORT` | `0.0.0.0` / `8080` | Bind address | | `APP_HOST` / `APP_PORT` | `0.0.0.0` / `8080` | Bind address |
¹ When no `OIDC_<n>_*` variable is set at all, a single provider named
`default` pointing at the dev mock IdP is used. Indices must be contiguous
from 0 (`OIDC_0_*`, `OIDC_1_*`, ...).
### Multiple OIDC providers
Providers are configured in indexed blocks; each becomes its own login
route `/auth/<provider>/login` with its own callback URL and session
namespace:
```sh
OIDC_0_PROVIDER=google
OIDC_0_ISSUER=https://accounts.google.com
OIDC_0_CLIENT_ID=...
OIDC_0_CLIENT_SECRET=...
OIDC_0_REDIRECT_URI=https://tavolo.example.com/auth/google/callback
OIDC_1_PROVIDER=github
OIDC_1_ISSUER=https://github.com/login/oauth
OIDC_1_CLIENT_ID=...
OIDC_1_CLIENT_SECRET=...
OIDC_1_REDIRECT_URI=https://tavolo.example.com/auth/github/callback
```
The login page reads `GET /api/auth/providers` and shows one button per
provider. `GET /auth/login` redirects to the first provider (useful for
bookmarks and scripts), and `GET /auth/logout` signs out whichever
provider the session is logged in with.
Sessions hold a single identity: logging in with another provider
replaces the previous one. Each provider's `sub` is a separate account,
so match history and Elo ratings are per provider identity (someone using
two providers has two accounts). Migrating from the pre-0.0.6 unindexed
variables: rename `OIDC_ISSUER`/`OIDC_CLIENT_ID`/... to `OIDC_0_...`, add
`OIDC_0_PROVIDER`, and register the new callback path
(`/auth/<provider>/callback`) at the provider. Sessions survive only when
the provider is named `default`; otherwise users log in once more.
## Logging ## Logging
The application logs through the Python stdlib `logging` module, one The application logs through the Python stdlib `logging` module, one
@@ -123,6 +167,37 @@ loggers:
level: INFO level: INFO
``` ```
## OpenTelemetry
With `OTEL_ENABLED` set, the app exports over OTLP/HTTP (kaya-otel plus
`tavolo.telemetry`):
- **Traces**: HTTP server spans, one span per WebSocket connection, plus
database spans for the Postgres (asyncpg, through Tortoise) and Redis
clients. Redis command values are redacted by the instrumentation
(`GET ? ?`) and asyncpg spans carry the parameterized SQL only — no
parameter values. Each HTTP/WebSocket span is tagged with the request's
session and user, so all traces of a session can be found together:
`session.id` is a truncated SHA-256 of the session id (the cookie is a
credential, so it is never exported verbatim) and `user.id` is the OIDC
`sub`, present once logged in.
- **Metrics**: the HTTP server metrics from kaya-otel, plus
`tavolo.active_users` (sessions in the session store),
`tavolo.ongoing_games` (live matches not finished yet, maintained by the
game store), and process/CPython metrics from
`opentelemetry-instrumentation-system-metrics`: `process.cpu.time`,
`process.cpu.utilization`, `process.memory.usage`,
`process.memory.virtual`, `cpython.gc.collections`,
`cpython.gc.collected_objects` and `cpython.gc.uncollectable_objects`.
- **Logs**: forwarded from the root logger at the configured console level
(OpenTelemetry's own records are filtered out to avoid export loops).
The session/game gauges are polled every
`OTEL_METRICS_POLL_INTERVAL_SECONDS` (Redis-backed stores report
cluster-wide counts, in-memory stores the process's own). Note that with
multiple workers or replicas each process reports the same global counts:
aggregate with `max`, not `sum`.
## Data model ## Data model
### Redis (live games) ### Redis (live games)
@@ -143,6 +218,9 @@ loggers:
on any player being connected and survive the death of any worker on any player being connected and survive the death of any worker
(delivery is at-least-once; engines revalidate the entry's token (delivery is at-least-once; engines revalidate the entry's token
against the live state before firing). against the live state before firing).
- `tavolo:deadlines:events` — a pub/sub channel announcing newly enqueued
deadlines; it wakes every worker's consumer immediately, while the
heartbeat poll remains as a backstop for lost signals.
### Postgres (statistics, via Tortoise ORM + aerich migrations) ### Postgres (statistics, via Tortoise ORM + aerich migrations)
@@ -184,7 +262,8 @@ existed):
## REST API ## REST API
All endpoints except `/api/health`, `/api/docs`, `/api/openapi.json`, All endpoints except `/api/health`, `/api/docs`, `/api/openapi.json`,
`/api/game-types` and `/api/leaderboard` require authentication. `/api/game-types`, `/api/leaderboard` and the game snapshot
(`GET /api/games/{id}`) require authentication.
| Method | Path | Description | | Method | Path | Description |
|---|---|---| |---|---|---|
@@ -192,29 +271,36 @@ All endpoints except `/api/health`, `/api/docs`, `/api/openapi.json`,
| `POST` | `/api/games` | Create a lobby game. Optional body `{"game_type": "scopone_scientifico", "options": {"target_score": 11, "napola": true, "asso_pigliatutto": false}}`. Returns the lobby (`{id, join_code, players, seats_open, ...}`) | | `POST` | `/api/games` | Create a lobby game. Optional body `{"game_type": "scopone_scientifico", "options": {"target_score": 11, "napola": true, "asso_pigliatutto": false}}`. Returns the lobby (`{id, join_code, players, seats_open, ...}`) |
| `POST` | `/api/games/join` | Join with `{"code": "ABC123"}`. The last player triggers the start and gets the game view back | | `POST` | `/api/games/join` | Join with `{"code": "ABC123"}`. The last player triggers the start and gets the game view back |
| `GET` | `/api/games/preview?code=ABC123` | Pre-join summary of the lobby behind a code: game name/description, seated players, open seats and the engine-echoed options. `404` when the code is unknown | | `GET` | `/api/games/preview?code=ABC123` | Pre-join summary of the lobby behind a code: game name/description, seated players, open seats and the engine-echoed options. `404` when the code is unknown |
| `GET` | `/api/games/{id}` | Personalized snapshot (only your own hand is visible) | | `GET` | `/api/games/{id}` | Game snapshot. Seated players get the personalized view (own hand plus the join code); anyone else gets the read-only spectator view: table, scores, captured counts and scope, no hands and no join code |
| `GET` | `/api/me/matches` | Cursor-paginated match history with the engine's `result` and per-player Elo deltas (`?limit=&cursor=&game_type=`) | | `GET` | `/api/me/matches` | Cursor-paginated match history with the engine's `result` and per-player Elo deltas (`?limit=&cursor=&game_type=`) |
| `GET` | `/api/me/ratings` | The caller's Elo rating per game type | | `GET` | `/api/me/ratings` | The caller's Elo rating per game type |
| `GET` | `/api/leaderboard` | Elo rating, aggregated wins / matches / points per player, sorted by Elo (`?game_type=`) | | `GET` | `/api/leaderboard` | Elo rating, aggregated wins / matches / points per player, sorted by Elo (`?game_type=`) |
## WebSocket protocol ## WebSocket protocol
Connect to `/ws/games/{game_id}` with your session cookie. Only seated Connect to `/ws/games/{game_id}`; the session cookie is optional. Anyone
players are accepted; unauthenticated, unknown-game and non-player who knows the game id is accepted: seated players get their personalized
connections are closed with codes `4401`, `4404` and `4403` respectively. view, everyone else — anonymous users and authenticated non-players — is
a read-only **spectator**. Only an unknown game id is refused before the
handshake completes, with `404`. (Browsers report a refused handshake as
an abnormal close — there is no close code to read.)
Server → client messages are JSON objects with a `type`: Server → client messages are JSON objects with a `type`:
- `{"type": "state", "game": {...}}` — personalized game view. Your own - `{"type": "state", "game": {...}}` — personalized game view. Your own
hand is included; other players expose only their card counts. hand is included; other players expose only their card counts.
`your_turn: true` is present when it is your move. Spectators get no hand at all, `spectator: true` marks the view and
`join_code` is omitted for them. `your_turn: true` is present when it
is your move.
- `{"type": "game_over", "scores": {"A": 11, "B": 7}, "winner": "A"}` - `{"type": "game_over", "scores": {"A": 11, "B": 7}, "winner": "A"}`
- `{"type": "error", "code": "illegal_move", "message": "..."}` - `{"type": "error", "code": "illegal_move", "message": "..."}`
Client → server messages are JSON objects with an `action`. The Client → server messages are JSON objects with an `action`. The
platform handles `state` (resend the snapshot) itself; every other action platform handles `state` (resend the snapshot) itself; every other action
is dispatched to the session's game engine with the rest of the message is dispatched to the session's game engine with the rest of the message
as its payload. For scopone scientifico: as its payload. Spectators may only send `state`/`sync`: any other
action is answered with an `error` whose `code` is `forbidden`. For
scopone scientifico:
```json ```json
{"action": "play", "card": "07D", "capture": ["02D", "05C"]} {"action": "play", "card": "07D", "capture": ["02D", "05C"]}
@@ -233,7 +319,8 @@ as its payload. For scopone scientifico:
after `HAND_ACK_TIMEOUT_SECONDS`. after `HAND_ACK_TIMEOUT_SECONDS`.
- `state` asks for a fresh snapshot. - `state` asks for a fresh snapshot.
After every accepted move the new state is broadcast to all four players. After every accepted move the new state is broadcast to every connected
client, spectators included.
### Turn timeout ### Turn timeout
@@ -356,6 +443,7 @@ packages/tavolo-app/ # deployable app: composition root (own pyproject)
src/tavolo/ src/tavolo/
├── app.py # session/OIDC/Tortoise/OpenAPI/Platform/DeadlineScheduler mixins ├── app.py # session/OIDC/Tortoise/OpenAPI/Platform/DeadlineScheduler mixins
├── config.py # env -> frozen Settings ├── config.py # env -> frozen Settings
├── telemetry.py # OTLP metric/log providers, app gauges, count poller
├── aerich_config.py # aerich CLI configuration ├── aerich_config.py # aerich CLI configuration
├── logging_config.py # logging setup ├── logging_config.py # logging setup
└── static.py # SPA shell hosting └── static.py # SPA shell hosting
+4 -3
View File
@@ -10,7 +10,8 @@ from __future__ import annotations
import os import os
os.environ.setdefault("DATABASE_URL", "sqlite://:memory:") os.environ.setdefault("DATABASE_URL", "sqlite://:memory:")
os.environ.setdefault("OIDC_ISSUER", "http://localhost:8180/tavolo") os.environ.setdefault("OIDC_0_PROVIDER", "default")
os.environ.setdefault("OIDC_CLIENT_ID", "tavolo") os.environ.setdefault("OIDC_0_ISSUER", "http://localhost:8180/tavolo")
os.environ.setdefault("OIDC_REDIRECT_URI", "http://localhost:8080/auth/callback") os.environ.setdefault("OIDC_0_CLIENT_ID", "tavolo")
os.environ.setdefault("OIDC_0_REDIRECT_URI", "http://localhost:8080/auth/default/callback")
os.environ.pop("REDIS_URL", None) os.environ.pop("REDIS_URL", None)
+7
View File
@@ -11,6 +11,12 @@ The deployable tavolo application: the composition root wiring
`ScoponeEngine` with its configured timeouts, and exposes `ScoponeEngine` with its configured timeouts, and exposes
`tavolo.app:app` for granian. `tavolo.app:app` for granian.
- `config.py` — environment-driven frozen `Settings`. - `config.py` — environment-driven frozen `Settings`.
- `telemetry.py` — the optional OpenTelemetry wiring beyond kaya-otel:
meter/tracer/logger providers, the
`tavolo.active_users`/`tavolo.ongoing_games` gauges (polled from the
stores), process/CPython metrics via
`opentelemetry-instrumentation-system-metrics`, Postgres/Redis query
spans via the asyncpg/redis instrumentations, and OTLP log export.
- `aerich_config.py` — Tortoise ORM configuration consumed by the aerich - `aerich_config.py` — Tortoise ORM configuration consumed by the aerich
CLI (migrations live in `server/migrations/`; aerich itself is CLI (migrations live in `server/migrations/`; aerich itself is
configured via the `[tool.aerich]` section of `server/pyproject.toml`). configured via the `[tool.aerich]` section of `server/pyproject.toml`).
@@ -23,6 +29,7 @@ The deployable tavolo application: the composition root wiring
.venv/bin/python -m mypy --namespace-packages --explicit-package-bases \ .venv/bin/python -m mypy --namespace-packages --explicit-package-bases \
src/tavolo/app.py \ src/tavolo/app.py \
src/tavolo/config.py \ src/tavolo/config.py \
src/tavolo/telemetry.py \
src/tavolo/static.py \ src/tavolo/static.py \
src/tavolo/aerich_config.py \ src/tavolo/aerich_config.py \
src/tavolo/logging_config.py src/tavolo/logging_config.py
+4 -1
View File
@@ -34,7 +34,10 @@ dev = [
"httpx-ws", "httpx-ws",
] ]
otel = [ otel = [
"kaya-otel>=0.0.4", "kaya-otel>=0.0.7",
"opentelemetry-instrumentation-asyncpg>=0.66b0",
"opentelemetry-instrumentation-redis>=0.66b0",
"opentelemetry-instrumentation-system-metrics>=0.66b0",
] ]
[tool.setuptools.packages.find] [tool.setuptools.packages.find]
+75 -37
View File
@@ -6,7 +6,7 @@ platform:
- :class:`~kaya.session.SessionMixin` (sessions persisted in Redis via - :class:`~kaya.session.SessionMixin` (sessions persisted in Redis via
:class:`~kaya.session.redis.RedisSessionStore` when ``REDIS_URL`` is set, :class:`~kaya.session.redis.RedisSessionStore` when ``REDIS_URL`` is set,
otherwise an in-memory store — e.g. for tests) otherwise an in-memory store — e.g. for tests)
- :class:`~kaya.oidc.OIDCMixin` (OIDC login) - :class:`~kaya.oidc.OIDCProviders` (the configured OIDC providers)
- :class:`~tavolo.platform.tortoise_mixin.TortoiseMixin` (Postgres match - :class:`~tavolo.platform.tortoise_mixin.TortoiseMixin` (Postgres match
statistics; skipped for ``/api/health`` and the OpenAPI documentation statistics; skipped for ``/api/health`` and the OpenAPI documentation
endpoints) endpoints)
@@ -31,14 +31,13 @@ from __future__ import annotations
from importlib.metadata import version as _pkg_version from importlib.metadata import version as _pkg_version
from logging import getLogger from logging import getLogger
from typing import Optional from typing import Any, Dict, Optional
from kaya.core import KayaApp, KayaMixin from kaya.core import KayaApp, KayaMixin
from kaya.cors import CorsMixin from kaya.cors import CorsMixin
from kaya.oidc import OIDCConfig, OIDCMixin from kaya.oidc import OIDCConfig, OIDCProviders
from kaya.openapi import OpenAPIMixin from kaya.openapi import OpenAPIMixin
from kaya.session import InMemorySessionStore, SessionMixin, SessionStore from kaya.session import SessionMixin
from kaya.session.redis import RedisSessionStore
from redis.asyncio import Redis from redis.asyncio import Redis
from tavolo.platform import GameRegistry, Platform, PlatformMixin from tavolo.platform import GameRegistry, Platform, PlatformMixin
from tavolo.platform.deadlines import DeadlineScheduler, DeadlineSchedulerMixin from tavolo.platform.deadlines import DeadlineScheduler, DeadlineSchedulerMixin
@@ -48,6 +47,14 @@ from tavolo.scopone import ScoponeEngine
from .config import Settings, settings from .config import Settings, settings
from .logging_config import configure_logging from .logging_config import configure_logging
from .telemetry import (
CountableSessionStore,
CountingInMemorySessionStore,
CountingRedisSessionStore,
otel_headers,
session_tag_hook,
telemetry_from_settings,
)
configure_logging(settings.logging_config) configure_logging(settings.logging_config)
log = getLogger(__name__) log = getLogger(__name__)
@@ -73,13 +80,23 @@ 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,
tracer_provider: Optional[Any] = None,
) -> Optional[KayaMixin]:
"""Build a :class:`~kaya.otel.OTelMixin` from the OTEL_* settings. """Build a :class:`~kaya.otel.OTelMixin` from the OTEL_* settings.
Returns ``None`` — telemetry disabled — unless ``OTEL_ENABLED`` is Returns ``None`` — telemetry disabled — unless ``OTEL_ENABLED`` is
truthy. kaya-otel is an optional dependency (the ``otel`` extra), so it truthy. kaya-otel is an optional dependency (the ``otel`` extra), so it
is imported lazily here: default installs and the test suite never need is imported lazily here: default installs and the test suite never need
the OpenTelemetry packages. the OpenTelemetry packages.
``meter_provider``/``tracer_provider`` are the providers built by
:func:`tavolo.telemetry.telemetry_from_settings`; injecting them lets
the custom instruments and the database instrumentations share the same
OTLP pipelines (kaya-otel then leaves their shutdown to the telemetry
mixin).
""" """
if not settings.otel_enabled: if not settings.otel_enabled:
return None return None
@@ -90,17 +107,21 @@ def otel_mixin_from_settings(settings: Settings) -> Optional[KayaMixin]:
"OTEL_ENABLED is set but kaya-otel is not installed; " "OTEL_ENABLED is set but kaya-otel is not installed; "
"install tavolo-app with the 'otel' extra" "install tavolo-app with the 'otel' extra"
) from exc ) from exc
headers = dict( kwargs: Dict[str, Any] = {
pair.split("=", 1) "service_name": settings.otel_service_name,
for pair in (settings.otel_exporter_headers or ()) "endpoint": settings.otel_exporter_endpoint,
if "=" in pair "headers": otel_headers(settings),
) "excluded_paths": settings.otel_excluded_paths,
return OTelMixin( # Runs after the session mixin's after hook, so the session id has
service_name=settings.otel_service_name, # been persisted/minted and the OIDC claims are on the session.
endpoint=settings.otel_exporter_endpoint, "server_response_hook": session_tag_hook,
headers=headers or None, "websocket_close_hook": session_tag_hook,
excluded_paths=settings.otel_excluded_paths, }
) if meter_provider is not None:
kwargs["meter_provider"] = meter_provider
if tracer_provider is not None:
kwargs["tracer_provider"] = tracer_provider
return OTelMixin(**kwargs)
registry = GameRegistry() registry = GameRegistry()
@@ -117,12 +138,12 @@ log.debug(
settings.turn_timeout_seconds, settings.turn_timeout_seconds,
) )
session_store: SessionStore session_store: CountableSessionStore
game_store: GameStore game_store: GameStore
if settings.redis_url is not None: if settings.redis_url is not None:
# Lazy client: no connection is opened until a session is actually # Lazy client: no connection is opened until a session is actually
# loaded/saved, so importing this module never requires a live Redis. # 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( game_store = RedisGameStore(
Redis.from_url(settings.redis_url, decode_responses=False), Redis.from_url(settings.redis_url, decode_responses=False),
registry, registry,
@@ -130,23 +151,31 @@ if settings.redis_url is not None:
) )
log.info("using Redis stores (sessions + live games, game TTL %ds)", settings.game_ttl_seconds) log.info("using Redis stores (sessions + live games, game TTL %ds)", settings.game_ttl_seconds)
else: else:
session_store = InMemorySessionStore() session_store = CountingInMemorySessionStore()
game_store = InMemoryGameStore(registry) game_store = InMemoryGameStore(registry)
log.info("REDIS_URL unset: using in-memory stores (sessions + live games)") log.info("REDIS_URL unset: using in-memory stores (sessions + live games)")
session_mixin = SessionMixin(session_store) session_mixin = SessionMixin(session_store)
oidc_mixin = OIDCMixin( oidc = OIDCProviders(
OIDCConfig( [
issuer=settings.oidc_issuer, OIDCConfig(
client_id=settings.oidc_client_id, provider=provider.provider,
client_secret=settings.oidc_client_secret, issuer=provider.issuer,
redirect_uri=settings.oidc_redirect_uri, client_id=provider.client_id,
post_login_redirect=settings.oidc_post_login_redirect, client_secret=provider.client_secret,
post_logout_redirect=settings.oidc_post_logout_redirect, redirect_uri=provider.redirect_uri,
fetch_userinfo=True, post_login_redirect=provider.post_login_redirect,
), post_logout_redirect=provider.post_logout_redirect,
fetch_userinfo=True,
)
for provider in settings.oidc_providers
],
session=session_mixin, session=session_mixin,
) )
log.info(
"OIDC providers: %s",
", ".join(provider.provider for provider in settings.oidc_providers),
)
openapi_mixin = OpenAPIMixin( openapi_mixin = OpenAPIMixin(
title="tavolo", title="tavolo",
version=_pkg_version("tavolo-app"), version=_pkg_version("tavolo-app"),
@@ -171,20 +200,29 @@ platform = Platform(
registry=registry, registry=registry,
game_store=game_store, game_store=game_store,
scheduler=scheduler, scheduler=scheduler,
oidc=oidc_mixin, oidc=oidc,
) )
mixins: list[KayaMixin] = [session_mixin, oidc_mixin, tortoise_mixin, openapi_mixin, mixins: list[KayaMixin] = [session_mixin, oidc, tortoise_mixin, openapi_mixin,
PlatformMixin(platform), DeadlineSchedulerMixin(scheduler)] PlatformMixin(platform), DeadlineSchedulerMixin(scheduler)]
otel_mixin = otel_mixin_from_settings(settings) telemetry = telemetry_from_settings(settings, session_store, game_store)
if otel_mixin is not None: if telemetry is not None:
# First in the list: before hooks run in registration order (after hooks otel_mixin = otel_mixin_from_settings(
settings,
meter_provider=telemetry.meter_provider,
tracer_provider=telemetry.tracer_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 # in reverse), so the span covers session loading, OIDC handling and the
# handler itself. CORS, when enabled, is still prepended before it so # 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) mixins.insert(0, otel_mixin)
log.info( log.info(
"OpenTelemetry enabled: service=%s endpoint=%s", "OpenTelemetry enabled: service=%s endpoint=%s (traces, metrics, logs)",
settings.otel_service_name, settings.otel_service_name,
settings.otel_exporter_endpoint or "(OTLP default)", settings.otel_exporter_endpoint or "(OTLP default)",
) )
+152 -22
View File
@@ -6,6 +6,7 @@ dataclass. No pydantic-settings, no settings module.
from __future__ import annotations from __future__ import annotations
import os import os
import re
from dataclasses import dataclass from dataclasses import dataclass
from typing import Optional, Tuple from typing import Optional, Tuple
from urllib.parse import quote from urllib.parse import quote
@@ -39,6 +40,135 @@ def _env_bool(name: str, default: bool = False) -> bool:
return value.strip().lower() in ("1", "true", "yes", "on") return value.strip().lower() in ("1", "true", "yes", "on")
# Per-provider OIDC settings are indexed with a 0-based suffix:
# OIDC_<n>_PROVIDER, OIDC_<n>_ISSUER, ... Indices must be contiguous from 0.
_OIDC_VARIABLES = (
"PROVIDER",
"ISSUER",
"CLIENT_ID",
"CLIENT_SECRET",
"REDIRECT_URI",
"POST_LOGIN_REDIRECT",
"POST_LOGOUT_REDIRECT",
)
_OIDC_ENV = re.compile(r"^OIDC_(\d+)_([A-Z0-9_]+)$")
_PROVIDER_NAME = re.compile(r"^[A-Za-z0-9_-]+$")
# The pre-0.0.6 unindexed names, detected only to point the migration out.
_LEGACY_OIDC_VARIABLES = (
"OIDC_ISSUER",
"OIDC_CLIENT_ID",
"OIDC_CLIENT_SECRET",
"OIDC_REDIRECT_URI",
"OIDC_POST_LOGIN_REDIRECT",
"OIDC_POST_LOGOUT_REDIRECT",
)
_DEFAULT_OIDC_ISSUER = "http://localhost:8180/tavolo"
_DEFAULT_OIDC_CLIENT_ID = "tavolo"
@dataclass(frozen=True)
class OidcProviderSettings:
"""One OIDC provider, parsed from the ``OIDC_<n>_*`` variables.
``provider`` names the provider in URLs (``/auth/<provider>/login``)
and namespaces its session state (``oidc:<provider>:...``); it must
match kaya-oidc's ``[A-Za-z0-9_-]+`` pattern.
"""
provider: str
issuer: str
client_id: str
client_secret: Optional[str]
redirect_uri: str
post_login_redirect: str
post_logout_redirect: str
def _oidc_providers_from_env() -> Tuple[OidcProviderSettings, ...]:
"""Parse the indexed ``OIDC_<n>_*`` provider configuration.
With no indexed variables at all, fall back to a single provider
named ``default`` pointing at the dev mock IdP, mirroring the
pre-0.0.6 defaults. The old unindexed variables are rejected with a
migration hint: they would otherwise be silently ignored.
"""
legacy = [name for name in _LEGACY_OIDC_VARIABLES if os.environ.get(name)]
if legacy:
renamed = ", ".join(
f"{name} -> OIDC_0_{name[len('OIDC_'):]}" for name in legacy
)
raise RuntimeError(
"the OIDC variables are now indexed per provider "
f"({renamed}; add OIDC_0_PROVIDER)"
)
indices: set = set()
for name in os.environ:
match = _OIDC_ENV.match(name)
if match is None:
continue
index, variable = match.groups()
if variable not in _OIDC_VARIABLES:
raise RuntimeError(
f"unknown OIDC environment variable {name!r}; expected one of "
+ ", ".join(f"OIDC_<n>_{v}" for v in _OIDC_VARIABLES)
)
indices.add(int(index))
if not indices:
return (
OidcProviderSettings(
provider="default",
issuer=_DEFAULT_OIDC_ISSUER,
client_id=_DEFAULT_OIDC_CLIENT_ID,
client_secret=None,
redirect_uri="http://localhost:8080/auth/default/callback",
post_login_redirect="/",
post_logout_redirect="/",
),
)
missing = sorted(set(range(max(indices) + 1)) - indices)
if missing:
raise RuntimeError(
"OIDC provider indices must be contiguous from 0; missing "
+ ", ".join(f"OIDC_{index}_*" for index in missing)
)
providers = []
seen = set()
for index in sorted(indices):
prefix = f"OIDC_{index}_"
provider = _env(prefix + "PROVIDER")
if _PROVIDER_NAME.match(provider) is None:
raise RuntimeError(
f"{prefix}PROVIDER must be a non-empty string of letters, "
f"digits, underscores or hyphens (got {provider!r})"
)
if provider in seen:
raise RuntimeError(f"duplicate OIDC provider name {provider!r}")
seen.add(provider)
providers.append(
OidcProviderSettings(
provider=provider,
issuer=_env(prefix + "ISSUER"),
client_id=_env(prefix + "CLIENT_ID"),
client_secret=os.environ.get(prefix + "CLIENT_SECRET") or None,
# kaya-oidc's OIDCProviders derives the callback route from
# the provider name; keep the default in sync with it.
redirect_uri=_env(
prefix + "REDIRECT_URI",
f"http://localhost:8080/auth/{provider}/callback",
),
post_login_redirect=_env(prefix + "POST_LOGIN_REDIRECT", "/"),
post_logout_redirect=_env(prefix + "POST_LOGOUT_REDIRECT", "/"),
)
)
return tuple(providers)
def _database_url_from_parts(engine: str, def _database_url_from_parts(engine: str,
user: str, user: str,
password: Optional[str], password: Optional[str],
@@ -70,15 +200,10 @@ def _database_url_from_parts(engine: str,
@dataclass(frozen=True) @dataclass(frozen=True)
class Settings: class Settings:
database_url: str database_url: str
oidc_issuer: str # The configured OIDC providers, in index order. Each provider
oidc_client_id: str # authenticates users independently and namespaces its session state;
oidc_client_secret: Optional[str] # see OidcProviderSettings above.
oidc_redirect_uri: str oidc_providers: Tuple[OidcProviderSettings, ...]
# Where the browser is sent after login/logout. In production the SPA is
# served by this app ("/"); in development point these at the trunk dev
# server (e.g. "http://localhost:8000/").
oidc_post_login_redirect: str
oidc_post_logout_redirect: str
app_host: str app_host: str
app_port: int app_port: int
redis_url: Optional[str] redis_url: Optional[str]
@@ -97,10 +222,10 @@ class Settings:
# card for them, covering disconnects and idle players (wired into the # card for them, covering disconnects and idle players (wired into the
# ScoponeEngine; see tavolo.app). # ScoponeEngine; see tavolo.app).
turn_timeout_seconds: int turn_timeout_seconds: int
# Upper bound on how long the deadline consumer sleeps between polls. # Upper bound on how long the deadline consumer sleeps between polls
# Locally enqueued deadlines wake the consumer immediately; the # when nothing else wakes it. Enqueued deadlines are announced over a
# heartbeat only bounds the discovery delay for deadlines enqueued by # pub/sub channel, so this is only the backstop recovering lost
# other workers. # notifications and deadlines orphaned by a dead worker.
deadline_heartbeat_ms: int deadline_heartbeat_ms: int
# Path to a YAML logging configuration file (logging.config.dictConfig # Path to a YAML logging configuration file (logging.config.dictConfig
# schema). Unset uses the built-in default: DEBUG to the console. # schema). Unset uses the built-in default: DEBUG to the console.
@@ -125,6 +250,10 @@ class Settings:
# Paths excluded from tracing and metrics (exact matches). Defaults to # Paths excluded from tracing and metrics (exact matches). Defaults to
# the health endpoint, which k8s probes would otherwise spam. # the health endpoint, which k8s probes would otherwise spam.
otel_excluded_paths: Tuple[str, ...] 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 @staticmethod
def from_env() -> "Settings": def from_env() -> "Settings":
@@ -143,12 +272,7 @@ class Settings:
# Raw DSN query string (e.g. "ssl=require"); empty = none. # Raw DSN query string (e.g. "ssl=require"); empty = none.
options=os.environ.get("DATABASE_OPTIONS", ""), options=os.environ.get("DATABASE_OPTIONS", ""),
), ),
oidc_issuer=_env("OIDC_ISSUER", "http://localhost:8180/tavolo"), oidc_providers=_oidc_providers_from_env(),
oidc_client_id=_env("OIDC_CLIENT_ID", "tavolo"),
oidc_client_secret=os.environ.get("OIDC_CLIENT_SECRET"),
oidc_redirect_uri=_env("OIDC_REDIRECT_URI", "http://localhost:8080/auth/callback"),
oidc_post_login_redirect=_env("OIDC_POST_LOGIN_REDIRECT", "/"),
oidc_post_logout_redirect=_env("OIDC_POST_LOGOUT_REDIRECT", "/"),
app_host=_env("APP_HOST", "0.0.0.0"), app_host=_env("APP_HOST", "0.0.0.0"),
app_port=int(_env("APP_PORT", "8080")), app_port=int(_env("APP_PORT", "8080")),
# When unset, sessions and live games fall back to in-memory # When unset, sessions and live games fall back to in-memory
@@ -159,7 +283,7 @@ class Settings:
static_dir=_env("STATIC_DIR", "web/dist"), static_dir=_env("STATIC_DIR", "web/dist"),
hand_ack_timeout_seconds=int(_env("HAND_ACK_TIMEOUT_SECONDS", "30")), hand_ack_timeout_seconds=int(_env("HAND_ACK_TIMEOUT_SECONDS", "30")),
turn_timeout_seconds=int(_env("TURN_TIMEOUT_SECONDS", "30")), turn_timeout_seconds=int(_env("TURN_TIMEOUT_SECONDS", "30")),
deadline_heartbeat_ms=int(_env("DEADLINE_HEARTBEAT_MS", "1000")), deadline_heartbeat_ms=int(_env("DEADLINE_HEARTBEAT_MS", "30000")),
logging_config=os.environ.get("LOGGING_CONFIG"), logging_config=os.environ.get("LOGGING_CONFIG"),
# CORS is disabled unless CORS_ALLOW_ORIGINS (a comma-separated # CORS is disabled unless CORS_ALLOW_ORIGINS (a comma-separated
# list of origins, or "*" for any) or CORS_ALLOW_ORIGIN_REGEX # list of origins, or "*" for any) or CORS_ALLOW_ORIGIN_REGEX
@@ -171,14 +295,20 @@ class Settings:
cors_allow_credentials=_env_bool("CORS_ALLOW_CREDENTIALS"), cors_allow_credentials=_env_bool("CORS_ALLOW_CREDENTIALS"),
cors_expose_headers=_env_list("CORS_EXPOSE_HEADERS"), cors_expose_headers=_env_list("CORS_EXPOSE_HEADERS"),
cors_max_age=int(_env("CORS_MAX_AGE", "600")), cors_max_age=int(_env("CORS_MAX_AGE", "600")),
# OpenTelemetry is opt-in: set OTEL_ENABLED=1 to export traces # OpenTelemetry is opt-in: set OTEL_ENABLED=1 to export traces,
# and metrics via OTLP/HTTP (requires the ``otel`` extra). # metrics and logs via OTLP/HTTP (requires the ``otel`` extra).
otel_enabled=_env_bool("OTEL_ENABLED"), otel_enabled=_env_bool("OTEL_ENABLED"),
otel_service_name=_env("OTEL_SERVICE_NAME", "tavolo"), otel_service_name=_env("OTEL_SERVICE_NAME", "tavolo"),
otel_exporter_endpoint=os.environ.get("OTEL_EXPORTER_OTLP_ENDPOINT") or None, otel_exporter_endpoint=os.environ.get("OTEL_EXPORTER_OTLP_ENDPOINT") or None,
# Comma-separated key=value pairs, e.g. "Authorization=Bearer x". # Comma-separated key=value pairs, e.g. "Authorization=Bearer x".
otel_exporter_headers=_env_list("OTEL_EXPORTER_OTLP_HEADERS"), otel_exporter_headers=_env_list("OTEL_EXPORTER_OTLP_HEADERS"),
otel_excluded_paths=_env_list("OTEL_EXCLUDED_PATHS") or ("/api/health",), 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")
),
) )
@@ -2,7 +2,9 @@
Configured once at app import time (see :mod:`tavolo.app`). By default a Configured once at app import time (see :mod:`tavolo.app`). By default a
single console handler at DEBUG level is installed, formatting records as single console handler at DEBUG level is installed, formatting records as
``{asctime} [{levelname}] ({processName}/{threadName}) - {name} - {message}``. ``{asctime} [{levelname}] ({processName}/{threadName}) - {name} - {message}``;
``urllib3.connectionpool`` is capped at INFO so its per-request DEBUG
chatter does not drown the log.
Point the ``LOGGING_CONFIG`` environment variable at a YAML file to take Point the ``LOGGING_CONFIG`` environment variable at a YAML file to take
over the configuration entirely; the file follows the over the configuration entirely; the file follows the
:data:`logging.config.dictConfig` schema, e.g.:: :data:`logging.config.dictConfig` schema, e.g.::
@@ -55,6 +57,13 @@ DEFAULT_CONFIG = {
"level": "DEBUG", "level": "DEBUG",
"handlers": ["console"], "handlers": ["console"],
}, },
"loggers": {
# Every pooled HTTP request is a DEBUG record; keep the logger's
# INFO+ output, drop the per-connection noise.
"urllib3.connectionpool": {
"level": "INFO",
},
},
} }
@@ -0,0 +1,375 @@
"""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;
* session/user tagging of request and websocket spans through
:func:`session_tag_hook` (kaya-otel's response/close hooks), so traces
can be correlated with a session (hashed id) and an OIDC user.
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 hashlib
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.auth import session_user
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
from opentelemetry.sdk.trace import TracerProvider
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
def session_tag_hook(span: Any, carrier: Any) -> None:
"""Tag an HTTP request/websocket span with the request's session/user.
Wired into kaya-otel as its ``server_response_hook`` and
``websocket_close_hook``: those run after the session mixin's after
hook has persisted (and, at login, minted) the session, so the final
session id is visible here.
The session id is the session cookie, i.e. a credential, so only a
truncated SHA-256 of it is exported; session ids are
``secrets.token_urlsafe(32)`` (256 bits), which makes the digest
irreversible in practice. The OIDC ``sub`` is a pseudonymous
identifier already persisted in Postgres and served by the API, so it
is exported verbatim as ``user.id`` (OpenTelemetry's current end-user
attribute; ``enduser.id`` is PII-flagged and being superseded).
"""
session = getattr(carrier, "session", None)
if session is None:
return
session_id = getattr(session, "id", None)
if session_id:
span.set_attribute(
"session.id",
hashlib.sha256(session_id.encode()).hexdigest()[:16],
)
user = session_user(session)
if user is not None:
sub = user.get("sub")
if sub:
span.set_attribute("user.id", str(sub))
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, DB instrumentation and log export.
The meter, tracer 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
(and uninstruments the database clients it instrumented).
"""
def __init__(
self,
*,
meter_provider: "MeterProvider",
tracer_provider: "TracerProvider",
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._tracer_provider = tracer_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
self._db_instrumentors: list = []
def apply(self, app: KayaApp) -> None:
from opentelemetry.instrumentation.asyncpg import AsyncPGInstrumentor
from opentelemetry.instrumentation.redis import RedisInstrumentor
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)
# Query spans for the Postgres (asyncpg, used by Tortoise) and Redis
# clients. Redis command values are redacted by the instrumentation
# (``SET ? ?``); asyncpg spans carry the parameterized SQL only.
self._db_instrumentors = [
AsyncPGInstrumentor(capture_parameters=False),
RedisInstrumentor(),
]
for db_instrumentor in self._db_instrumentors:
db_instrumentor.instrument(tracer_provider=self._tracer_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
for db_instrumentor in self._db_instrumentors:
db_instrumentor.uninstrument()
self._db_instrumentors = []
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._tracer_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"
tracer_provider: "TracerProvider"
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/trace/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.exporter.otlp.proto.http.trace_exporter import (
OTLPSpanExporter,
)
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
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
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})
tracer_provider = TracerProvider(resource=resource)
tracer_provider.add_span_processor(
BatchSpanProcessor(
OTLPSpanExporter(
endpoint=_signal_endpoint(endpoint, "v1/traces"),
headers=headers,
)
)
)
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,
tracer_provider=tracer_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,
tracer_provider=tracer_provider,
logger_provider=logger_provider,
log_handler=handler,
mixin=mixin,
)
@@ -9,9 +9,9 @@ description = "Game-independent multiplayer game platform (lobby, live play, mat
readme = "README.md" readme = "README.md"
requires-python = ">=3.10" requires-python = ">=3.10"
dependencies = [ dependencies = [
"kaya-core", "kaya-core>=0.0.7",
"kaya-session", "kaya-session",
"kaya-oidc", "kaya-oidc>=0.0.7",
"kaya-openapi", "kaya-openapi",
"tortoise-orm", "tortoise-orm",
"redis", "redis",
@@ -9,7 +9,29 @@ from __future__ import annotations
from typing import Any, Callable, Mapping, Optional from typing import Any, Callable, Mapping, Optional
from kaya.core import HttpContext, WebSocket from kaya.core import HttpContext, WebSocket
from kaya.oidc import OIDCMixin, OIDCUser from kaya.oidc import OIDCProviders, OIDCUser
from kaya.session import Session
# kaya-oidc namespaces its session state per provider as
# ``oidc:{provider}:user`` (since 0.0.6), so the lookup cannot assume a
# fixed key or provider name.
_OIDC_PREFIX = "oidc:"
_USER_SUFFIX = ":user"
def session_user(session: Session) -> Optional[OIDCUser]:
"""Return the authenticated user stored in a session, if any.
The kaya-oidc mixin writes the userinfo at login under a per-provider
key, so any provider's user is accepted (the configured provider is
``default`` unless the app says otherwise).
"""
for key in session:
if key.startswith(_OIDC_PREFIX) and key.endswith(_USER_SUFFIX):
claims = session.get(key)
if isinstance(claims, Mapping):
return OIDCUser(claims)
return None
def get_ws_user(ws: WebSocket) -> Optional[OIDCUser]: def get_ws_user(ws: WebSocket) -> Optional[OIDCUser]:
@@ -21,10 +43,7 @@ def get_ws_user(ws: WebSocket) -> Optional[OIDCUser]:
session = getattr(ws, "session", None) session = getattr(ws, "session", None)
if session is None: if session is None:
return None return None
claims = session.get("oidc_user") return session_user(session)
if not isinstance(claims, Mapping):
return None
return OIDCUser(claims)
def display_name(user: OIDCUser) -> str: def display_name(user: OIDCUser) -> str:
@@ -36,7 +55,16 @@ def display_name(user: OIDCUser) -> str:
return user.sub return user.sub
def require_auth(oidc: OIDCMixin) -> Callable[[Callable[..., Any]], Callable[..., Any]]: def current_user(oidc: OIDCProviders, ctx: HttpContext) -> Optional[OIDCUser]:
"""Return the authenticated user of an HTTP request, if any.
Sessions are single-identity by default, so at most one provider has
a user; if several do, the first configured provider wins.
"""
return next(iter(oidc.get_users(ctx).values()), None)
def require_auth(oidc: OIDCProviders) -> Callable[[Callable[..., Any]], Callable[..., Any]]:
"""Build a decorator gating a handler on being authenticated. """Build a decorator gating a handler on being authenticated.
Responds ``401`` with a JSON error envelope when unauthenticated — Responds ``401`` with a JSON error envelope when unauthenticated —
@@ -46,7 +74,7 @@ def require_auth(oidc: OIDCMixin) -> Callable[[Callable[..., Any]], Callable[...
def decorator(handler: Callable[..., Any]) -> Callable[..., Any]: def decorator(handler: Callable[..., Any]) -> Callable[..., Any]:
async def guarded(ctx: HttpContext, *args: Any, **kwargs: Any) -> None: async def guarded(ctx: HttpContext, *args: Any, **kwargs: Any) -> None:
if oidc.get_user(ctx) is None: if current_user(oidc, ctx) is None:
await ctx.send_bytes( await ctx.send_bytes(
401, 401,
b'{"error":"unauthenticated"}', b'{"error":"unauthenticated"}',
@@ -11,8 +11,11 @@ machinery; the *meaning* of each deadline belongs to the game engine:
token), and the scheduler enqueues it in the store's shared deadline token), and the scheduler enqueues it in the store's shared deadline
queue (a Redis sorted set in production, see queue (a Redis sorted set in production, see
:mod:`tavolo.platform.store`); enqueueing is idempotent; :mod:`tavolo.platform.store`); enqueueing is idempotent;
* a background consumer running on **every** worker polls the queue for * enqueueing a deadline announces it on a shared pub/sub channel, so
due entries and hands them to the engine's every worker's consumer wakes immediately instead of waiting for its
next poll; the consumer also wakes on a slow heartbeat, which is the
backstop recovering notifications lost to a reconnect or a worker
crash. Due entries are handed to the engine's
:meth:`~tavolo.platform.engine.GameEngine.fire_deadline` under the :meth:`~tavolo.platform.engine.GameEngine.fire_deadline` under the
per-game lock; the engine revalidates the token against the live state per-game lock; the engine revalidates the token against the live state
and raises :class:`~tavolo.platform.errors.GameError` when the entry and raises :class:`~tavolo.platform.errors.GameError` when the entry
@@ -65,8 +68,8 @@ class DeadlineScheduler:
"""Enqueue and fire the deadlines game engines ask for. """Enqueue and fire the deadlines game engines ask for.
Holds the store and the registry so a single instance serves every Holds the store and the registry so a single instance serves every
game type. One consumer task (and its wake-up event) is kept per game type. One consumer task, its listener task and its wake-up
event loop — tests run each test on a fresh loop. event are kept per event loop — tests run each test on a fresh loop.
``context_binder`` is an optional async callable the consumer awaits ``context_binder`` is an optional async callable the consumer awaits
before persisting a finished match. The consumer is a long-lived task before persisting a finished match. The consumer is a long-lived task
@@ -80,14 +83,19 @@ class DeadlineScheduler:
store: GameStore, store: GameStore,
registry: GameRegistry, registry: GameRegistry,
heartbeat_ms: int = 1000, heartbeat_ms: int = 1000,
retry_ms: int = 1000,
context_binder: Optional[Callable[[], Awaitable[None]]] = None, context_binder: Optional[Callable[[], Awaitable[None]]] = None,
) -> None: ) -> None:
self._store = store self._store = store
self._registry = registry self._registry = registry
self._heartbeat = heartbeat_ms / 1000 self._heartbeat = heartbeat_ms / 1000
# Backoff for entries that failed to process and for listener
# reconnects; deliberately independent of the (slow) heartbeat.
self._retry_delay = retry_ms / 1000
self._context_binder = context_binder self._context_binder = context_binder
self._consumers: Dict[asyncio.AbstractEventLoop, asyncio.Task] = {} self._consumers: Dict[asyncio.AbstractEventLoop, asyncio.Task] = {}
self._wake_events: Dict[asyncio.AbstractEventLoop, asyncio.Event] = {} self._wake_events: Dict[asyncio.AbstractEventLoop, asyncio.Event] = {}
self._listeners: Dict[asyncio.AbstractEventLoop, asyncio.Task] = {}
async def sync_deadline(self, session: GameSession) -> None: async def sync_deadline(self, session: GameSession) -> None:
"""Enqueue the deadline the current state implies, if any. """Enqueue the deadline the current state implies, if any.
@@ -109,6 +117,16 @@ class DeadlineScheduler:
self.ensure_consumer() self.ensure_consumer()
due_at = deadline.due_at.timestamp() due_at = deadline.due_at.timestamp()
await self._store.add_deadline(encode(entry), due_at) await self._store.add_deadline(encode(entry), due_at)
# Announce the enqueue so consumers on other workers wake up too.
# The entry is already queued, so a lost notification only delays
# their discovery to the next poll backstop.
try:
await self._store.notify_deadline()
except Exception:
log.warning(
"deadline enqueue notification failed; relying on the poll backstop",
exc_info=True,
)
wake = self._wake_events.get(asyncio.get_running_loop()) wake = self._wake_events.get(asyncio.get_running_loop())
if wake is not None: if wake is not None:
wake.set() wake.set()
@@ -190,15 +208,22 @@ class DeadlineScheduler:
if old.is_closed(): if old.is_closed():
self._consumers.pop(old, None) self._consumers.pop(old, None)
self._wake_events.pop(old, None) self._wake_events.pop(old, None)
self._listeners.pop(old, None)
task = self._consumers.get(loop) task = self._consumers.get(loop)
if task is None or task.done(): if task is None or task.done():
self._wake_events[loop] = asyncio.Event() self._wake_events[loop] = asyncio.Event()
self._consumers[loop] = loop.create_task(self._run(loop)) self._consumers[loop] = loop.create_task(self._run(loop))
listener = self._listeners.get(loop)
if listener is None or listener.done():
self._listeners[loop] = loop.create_task(self._listen(loop))
log.debug("deadline consumer started") log.debug("deadline consumer started")
def stop_consumer(self, loop: asyncio.AbstractEventLoop) -> None: def stop_consumer(self, loop: asyncio.AbstractEventLoop) -> None:
task = self._consumers.pop(loop, None) task = self._consumers.pop(loop, None)
self._wake_events.pop(loop, None) self._wake_events.pop(loop, None)
listener = self._listeners.pop(loop, None)
if listener is not None:
listener.cancel()
if task is not None: if task is not None:
task.cancel() task.cancel()
@@ -228,21 +253,48 @@ class DeadlineScheduler:
log.exception("deadline consumer: poll failed; retrying") log.exception("deadline consumer: poll failed; retrying")
failed = True failed = True
if failed: if failed:
# A failed entry is still due: back off to the heartbeat # A failed entry is still due: back off briefly instead of
# instead of polling it again immediately. # polling it again immediately. Kept independent of the
delay = self._heartbeat # heartbeat, which is only a slow backstop.
delay = self._retry_delay
try: try:
await asyncio.wait_for(wake.wait(), timeout=delay) await asyncio.wait_for(wake.wait(), timeout=delay)
except asyncio.TimeoutError: except asyncio.TimeoutError:
pass pass
async def _listen(self, loop: asyncio.AbstractEventLoop) -> None:
"""Wake the consumer on deadline-enqueue notifications.
The subscription is best effort: a dropped connection or a lost
message is recovered by the consumer's heartbeat backstop, so
this task only has to reconnect and keep forwarding signals.
"""
while True:
try:
async with self._store.subscribe_deadlines() as events:
async for _ in events:
wake = self._wake_events.get(loop)
if wake is None:
# Shut down: the consumer is gone.
return
wake.set()
except asyncio.CancelledError:
raise
except Exception:
log.exception("deadline listener failed; reconnecting")
try:
await asyncio.sleep(self._retry_delay)
except asyncio.CancelledError:
raise
class DeadlineSchedulerMixin(KayaMixin): class DeadlineSchedulerMixin(KayaMixin):
"""Run the deadline consumer for the whole app lifetime. """Run the deadline consumer for the whole app lifetime.
Every worker (and every pod) runs the same consumer; coordination Every worker (and every pod) runs the same consumer and listener;
happens exclusively through the shared deadline queue and the coordination happens exclusively through the shared deadline queue,
per-game locks, so any worker may fire any game's deadline. its notification channel and the per-game locks, so any worker may
fire any game's deadline.
""" """
def __init__(self, scheduler: DeadlineScheduler) -> None: def __init__(self, scheduler: DeadlineScheduler) -> None:
@@ -191,12 +191,17 @@ class GameEngine(ABC):
""" """
@abstractmethod @abstractmethod
def view_for(self, session: GameSession, user_sub: str) -> Dict[str, Any]: def view_for(self, session: GameSession, user_sub: Optional[str]) -> Dict[str, Any]:
"""The personalized game view for one player. """The personalized game view for one viewer.
Must hide information the player is not entitled to (e.g. other Must hide information the viewer is not entitled to (e.g. other
players' hands). The platform merges in the envelope fields players' hands). ``user_sub`` is ``None`` for a spectator — an
(``id``, ``join_code``, ``game_type``) itself. anonymous user or an authenticated user who is not seated — and
the view must then contain no privileged information either (no
hand, no legal moves: only what everyone at the table can see).
The platform merges in the envelope fields (``id``,
``join_code`` for seated players, ``game_type``, ``spectator``)
itself.
""" """
@abstractmethod @abstractmethod
@@ -248,6 +253,13 @@ class GameEngine(ABC):
``kind``/``token`` no longer match the live state. ``kind``/``token`` no longer match the live state.
""" """
def game_over_view(self, session: GameSession, user_sub: str) -> Dict[str, Any]: def game_over_view(
"""Extra fields merged into the ``game_over`` websocket message.""" self, session: GameSession, user_sub: Optional[str]
) -> Dict[str, Any]:
"""Extra fields merged into the ``game_over`` websocket message.
``user_sub`` is ``None`` for a spectator (see :meth:`view_for`);
the default fields are public, so the default implementation
ignores it.
"""
return {} return {}
@@ -2,7 +2,7 @@
``Platform`` bundles the collaborators every platform endpoint needs — ``Platform`` bundles the collaborators every platform endpoint needs —
the game registry, the live-game store, the deadline scheduler and the the game registry, the live-game store, the deadline scheduler and the
OIDC mixin used for authentication — so route and websocket modules OIDC providers used for authentication — so route and websocket modules
never reach for module-global state. :class:`PlatformMixin` is a plain never reach for module-global state. :class:`PlatformMixin` is a plain
:class:`~kaya.core.KayaMixin`: applying it registers the lobby, stats :class:`~kaya.core.KayaMixin`: applying it registers the lobby, stats
and health HTTP routes plus the live-play websocket endpoint, and the and health HTTP routes plus the live-play websocket endpoint, and the
@@ -14,7 +14,7 @@ from __future__ import annotations
from dataclasses import dataclass from dataclasses import dataclass
from kaya.core import KayaApp, KayaMixin from kaya.core import KayaApp, KayaMixin
from kaya.oidc import OIDCMixin from kaya.oidc import OIDCProviders
from .deadlines import DeadlineScheduler from .deadlines import DeadlineScheduler
from .registry import GameRegistry from .registry import GameRegistry
@@ -28,7 +28,7 @@ class Platform:
registry: GameRegistry registry: GameRegistry
game_store: GameStore game_store: GameStore
scheduler: DeadlineScheduler scheduler: DeadlineScheduler
oidc: OIDCMixin oidc: OIDCProviders
class PlatformMixin(KayaMixin): class PlatformMixin(KayaMixin):
@@ -41,9 +41,10 @@ class PlatformMixin(KayaMixin):
# Imported here so module import order stays acyclic: the route # Imported here so module import order stays acyclic: the route
# modules reference Platform only for typing. # modules reference Platform only for typing.
from . import ws from . import ws
from .routes import games, health, me, stats from .routes import auth, games, health, me, stats
health.register(app, self.platform) health.register(app, self.platform)
auth.register(app, self.platform)
me.register(app, self.platform) me.register(app, self.platform)
games.register(app, self.platform) games.register(app, self.platform)
stats.register(app, self.platform) stats.register(app, self.platform)
@@ -15,12 +15,15 @@ from __future__ import annotations
import base64 import base64
import json import json
from dataclasses import dataclass from dataclasses import dataclass
from typing import Any, Dict, List, Optional, Tuple from typing import TYPE_CHECKING, Any, Dict, List, Optional, Tuple
from tortoise.queryset import QuerySet from tortoise.queryset import QuerySet
from .http import extract_query_params from .http import extract_query_params
if TYPE_CHECKING:
from tortoise.models import Model
DEFAULT_LIMIT = 20 DEFAULT_LIMIT = 20
MAX_LIMIT = 100 MAX_LIMIT = 100
MIN_LIMIT = 1 MIN_LIMIT = 1
@@ -81,6 +84,28 @@ def parse_cursor_params(query_string: str) -> Cursor:
Sort = List[Tuple[str, str]] Sort = List[Tuple[str, str]]
def _coerce_sort_values(
model: "type[Model]", sort: Sort, after: Dict[str, Any]
) -> Dict[str, Any]:
"""Restore the Python type of every decoded cursor value.
The cursor is JSON, so field values travel as strings (datetimes via
``default=str``). asyncpg rejects strings where a ``timestamp``
parameter is expected, and sqlite only tolerates them by accident, so
each value is run through its model field's ``to_python_value`` before
it reaches the keyset filter.
"""
fields = model._meta.fields_map
coerced: Dict[str, Any] = {}
for field, _ in sort:
value = after.get(field)
field_obj = fields.get(field)
if value is not None and field_obj is not None:
value = field_obj.to_python_value(value)
coerced[field] = value
return coerced
def _keyset_where(sort: Sort, after: Dict[str, Any]): def _keyset_where(sort: Sort, after: Dict[str, Any]):
"""Build a Tortoise ``Q`` filter from a cursor.""" """Build a Tortoise ``Q`` filter from a cursor."""
from tortoise.queryset import Q # local to keep import edge narrow from tortoise.queryset import Q # local to keep import edge narrow
@@ -116,7 +141,8 @@ async def paginate(
order_by.append(field if direction == "ASC" else f"-{field}") order_by.append(field if direction == "ASC" else f"-{field}")
qs = queryset.order_by(*order_by) qs = queryset.order_by(*order_by)
if cursor.after: if cursor.after:
qs = qs.filter(_keyset_where(sort, cursor.after)) after = _coerce_sort_values(queryset.model, sort, cursor.after)
qs = qs.filter(_keyset_where(sort, after))
rows = await qs.limit(cursor.limit + 1) rows = await qs.limit(cursor.limit + 1)
if len(rows) <= cursor.limit: if len(rows) <= cursor.limit:
return rows, None return rows, None
@@ -0,0 +1,62 @@
"""Authentication plumbing for the single-page app.
kaya-oidc registers one login/callback/logout route per provider
(``/auth/<provider>/...``); this module adds the provider-agnostic
pieces:
* ``GET /api/auth/providers`` lists the configured providers with their
login URLs, so the SPA can offer a choice without hardcoding names;
* ``GET /auth/login`` redirects to the first configured provider;
* ``GET /auth/logout`` redirects to the provider this session is logged
in with (or the SPA root when not logged in), so the UI can keep a
single logout link.
"""
from __future__ import annotations
from typing import TYPE_CHECKING
from kaya.core import HttpContext, KayaApp
from kaya.openapi import operation
from ..http import send_json
if TYPE_CHECKING:
from ..mixin import Platform
def _login_url(platform: "Platform", provider: str) -> str:
return f"{platform.oidc.base_path}/{provider}/login"
def register(app: KayaApp, platform: "Platform") -> None:
providers = list(platform.oidc.providers)
@app.GET("/api/auth/providers")
@operation(summary="Configured login providers",
description="Names and login URLs of the configured OIDC "
"providers, in configuration order.",
tags=["auth"],
responses={200: {"description": "The configured providers"}})
async def auth_providers(ctx: HttpContext) -> None:
await send_json(
ctx,
200,
{
"providers": [
{"name": name, "login_url": _login_url(platform, name)}
for name in providers
]
},
)
@app.GET("/auth/login")
async def login(ctx: HttpContext) -> None:
await ctx.send_empty(302, {"Location": _login_url(platform, providers[0])})
@app.GET("/auth/logout")
async def logout(ctx: HttpContext) -> None:
logged_in = platform.oidc.logged_in_providers(ctx)
location = (
f"{platform.oidc.base_path}/{logged_in[0]}/logout" if logged_in else "/"
)
await ctx.send_empty(302, {"Location": location})
@@ -16,12 +16,12 @@ import secrets
import uuid import uuid
from datetime import datetime, timezone from datetime import datetime, timezone
from logging import getLogger from logging import getLogger
from typing import TYPE_CHECKING, Any, Dict from typing import TYPE_CHECKING, Any, Dict, Optional
from kaya.core import HttpContext, KayaApp from kaya.core import HttpContext, KayaApp
from kaya.openapi import operation from kaya.openapi import operation
from ..auth import display_name, require_auth from ..auth import current_user, display_name, require_auth
from ..engine import GameSession, Seat from ..engine import GameSession, Seat
from ..errors import GameError from ..errors import GameError
from ..http import ( from ..http import (
@@ -76,14 +76,21 @@ def _lobby_payload(platform: "Platform", session: GameSession) -> Dict[str, Any]
} }
def _view_payload(platform: "Platform", session: GameSession, sub: str) -> Dict[str, Any]: def _view_payload(
platform: "Platform", session: GameSession, sub: Optional[str]
) -> Dict[str, Any]:
engine = platform.registry.require(session.game_type) engine = platform.registry.require(session.game_type)
return { seated = sub is not None and session.seated(sub)
payload: Dict[str, Any] = {
"id": session.id, "id": session.id,
"join_code": session.join_code,
"game_type": session.game_type, "game_type": session.game_type,
**engine.view_for(session, sub), "spectator": not seated,
} }
if seated:
# The join code is a credential: only players get it.
payload["join_code"] = session.join_code
payload.update(engine.view_for(session, sub))
return payload
def _preview_payload(platform: "Platform", session: GameSession) -> Dict[str, Any]: def _preview_payload(platform: "Platform", session: GameSession) -> Dict[str, Any]:
@@ -183,7 +190,7 @@ def register(app: KayaApp, platform: "Platform") -> None:
await send_error(ctx, 400, "options must be an object") await send_error(ctx, 400, "options must be an object")
return return
user = platform.oidc.get_user(ctx) user = current_user(platform.oidc, ctx)
assert user is not None # enforced by @require_auth assert user is not None # enforced by @require_auth
session = GameSession( session = GameSession(
id=str(uuid.uuid4()), id=str(uuid.uuid4()),
@@ -244,7 +251,7 @@ def register(app: KayaApp, platform: "Platform") -> None:
await send_error(ctx, 400, "code is required") await send_error(ctx, 400, "code is required")
return return
user = platform.oidc.get_user(ctx) user = current_user(platform.oidc, ctx)
assert user is not None assert user is not None
existing = await platform.game_store.find_by_code(code) existing = await platform.game_store.find_by_code(code)
if existing is None: if existing is None:
@@ -310,24 +317,24 @@ def register(app: KayaApp, platform: "Platform") -> None:
@app.GET("/api/games/${game_id}") @app.GET("/api/games/${game_id}")
@operation(summary="Get a game snapshot", @operation(summary="Get a game snapshot",
description="Only seated players may read a game; the view is " description="Anyone who knows the game id may read a game. "
"personalized by the engine (e.g. hidden hands).", "Seated players get the personalized view (their "
"own hand plus the join code); everyone else — "
"anonymous users and authenticated non-players — "
"gets the read-only spectator view: no hands, no "
"join code, just what is public at the table.",
tags=["games"], tags=["games"],
responses={ responses={
200: {"description": "The personalized game state"}, 200: {"description": "The personalized (player) or "
401: {"description": "Authentication required"}, "read-only (spectator) game state"},
403: {"description": "Not a player in this game"},
404: {"description": "Game not found"}, 404: {"description": "Game not found"},
}) })
@require_auth(platform.oidc)
async def get_game(ctx: HttpContext, game_id: str) -> None: async def get_game(ctx: HttpContext, game_id: str) -> None:
session = await platform.game_store.load(game_id) session = await platform.game_store.load(game_id)
if session is None: if session is None:
await send_error(ctx, 404, "game not found") await send_error(ctx, 404, "game not found")
return return
user = platform.oidc.get_user(ctx) user = current_user(platform.oidc, ctx)
assert user is not None await send_json(
if not session.seated(user.sub): ctx, 200, _view_payload(platform, session, user.sub if user else None)
await send_error(ctx, 403, "forbidden") )
return
await send_json(ctx, 200, _view_payload(platform, session, user.sub))
@@ -6,7 +6,7 @@ from typing import TYPE_CHECKING
from kaya.core import HttpContext, KayaApp from kaya.core import HttpContext, KayaApp
from kaya.openapi import operation from kaya.openapi import operation
from ..auth import display_name, require_auth from ..auth import current_user, display_name, require_auth
from ..http import send_json from ..http import send_json
if TYPE_CHECKING: if TYPE_CHECKING:
@@ -25,6 +25,6 @@ def register(app: KayaApp, platform: "Platform") -> None:
}) })
@require_auth(platform.oidc) @require_auth(platform.oidc)
async def me(ctx: HttpContext) -> None: async def me(ctx: HttpContext) -> None:
user = platform.oidc.get_user(ctx) user = current_user(platform.oidc, ctx)
assert user is not None # enforced by @require_auth assert user is not None # enforced by @require_auth
await send_json(ctx, 200, {"sub": user.sub, "name": display_name(user)}) await send_json(ctx, 200, {"sub": user.sub, "name": display_name(user)})
@@ -8,12 +8,14 @@ verbatim through the ``result`` JSON column.
""" """
from __future__ import annotations from __future__ import annotations
from collections import defaultdict
from typing import TYPE_CHECKING, Any, Dict, List, Optional, Tuple from typing import TYPE_CHECKING, Any, Dict, List, Optional, Tuple
from uuid import UUID
from kaya.core import HttpContext, KayaApp from kaya.core import HttpContext, KayaApp
from kaya.openapi import operation from kaya.openapi import operation
from ..auth import require_auth from ..auth import current_user, require_auth
from ..elo import INITIAL_RATING from ..elo import INITIAL_RATING
from ..http import extract_query_params, send_error, send_json from ..http import extract_query_params, send_error, send_json
from ..models import Match, MatchPlayer, PlayerRating from ..models import Match, MatchPlayer, PlayerRating
@@ -48,8 +50,9 @@ def _parse_game_type(
return game_type, None return game_type, None
async def _serialize_match(match: Match, viewer: str) -> Dict[str, Any]: def _serialize_match(
participants = await MatchPlayer.filter(match_id=match.id).order_by("seat") match: Match, viewer: str, participants: List[MatchPlayer]
) -> Dict[str, Any]:
return { return {
"id": str(match.id), "id": str(match.id),
"game_type": match.game_type, "game_type": match.game_type,
@@ -101,7 +104,7 @@ def register(app: KayaApp, platform: "Platform") -> None:
if error is not None: if error is not None:
await send_error(ctx, 400, error) await send_error(ctx, 400, error)
return return
user = platform.oidc.get_user(ctx) user = current_user(platform.oidc, ctx)
assert user is not None assert user is not None
queryset = Match.filter(players__user_sub=user.sub).distinct() queryset = Match.filter(players__user_sub=user.sub).distinct()
if game_type is not None: if game_type is not None:
@@ -109,7 +112,21 @@ def register(app: KayaApp, platform: "Platform") -> None:
matches, next_cursor = await paginate( matches, next_cursor = await paginate(
queryset, [("finished_at", "DESC"), ("id", "ASC")], cursor queryset, [("finished_at", "DESC"), ("id", "ASC")], cursor
) )
results = [await _serialize_match(m, user.sub) for m in matches] # One participants query for the whole page instead of one per
# match; the grouping keeps each match's players seat-ordered.
by_match: Dict[UUID, List[MatchPlayer]] = defaultdict(list)
if matches:
participants = (
await MatchPlayer.filter(match_id__in=[m.id for m in matches])
.order_by("match_id", "seat")
)
for player in participants:
# Tortoise sets the FK's raw id on instances but does not
# declare it on the model class, hence the getattr.
by_match[getattr(player, "match_id")].append(player)
results = [
_serialize_match(m, user.sub, by_match.get(m.id, [])) for m in matches
]
await send_json(ctx, 200, {"results": results, "next_cursor": next_cursor}) await send_json(ctx, 200, {"results": results, "next_cursor": next_cursor})
@app.GET("/api/leaderboard") @app.GET("/api/leaderboard")
@@ -176,7 +193,7 @@ def register(app: KayaApp, platform: "Platform") -> None:
}) })
@require_auth(platform.oidc) @require_auth(platform.oidc)
async def my_ratings(ctx: HttpContext) -> None: async def my_ratings(ctx: HttpContext) -> None:
user = platform.oidc.get_user(ctx) user = current_user(platform.oidc, ctx)
assert user is not None assert user is not None
rows = await PlayerRating.filter(user_sub=user.sub).order_by("game_type") rows = await PlayerRating.filter(user_sub=user.sub).order_by("game_type")
await send_json(ctx, 200, { await send_json(ctx, 200, {
@@ -26,8 +26,9 @@ signal (never the state) means updated state reaches connections on
every worker without leaking hidden information into the channel. every worker without leaking hidden information into the channel.
Timeouts are driven by a shared delayed-deadline queue: producers Timeouts are driven by a shared delayed-deadline queue: producers
enqueue an opaque ``member`` string with a due timestamp, and a consumer enqueue an opaque ``member`` string with a due timestamp and announce it
on every worker polls for due entries (see on a shared "deadline enqueued" channel, and a consumer on every worker
wakes on that signal to process due entries (see
:mod:`tavolo.platform.deadlines`). Delivery is at-least-once — entries :mod:`tavolo.platform.deadlines`). Delivery is at-least-once — entries
are removed only after they are processed — so a worker dying are removed only after they are processed — so a worker dying
mid-processing cannot lose a deadline; engines revalidate entries mid-processing cannot lose a deadline; engines revalidate entries
@@ -38,6 +39,7 @@ from __future__ import annotations
import asyncio import asyncio
import contextlib import contextlib
import json import json
import time
from abc import ABC, abstractmethod from abc import ABC, abstractmethod
from datetime import datetime from datetime import datetime
from logging import getLogger from logging import getLogger
@@ -54,6 +56,13 @@ GAME_KEY_PREFIX = "tavolo:game:"
CODE_KEY_PREFIX = "tavolo:code:" CODE_KEY_PREFIX = "tavolo:code:"
CHANNEL_PREFIX = "tavolo:game:" CHANNEL_PREFIX = "tavolo:game:"
DEADLINES_KEY = "tavolo:deadlines" DEADLINES_KEY = "tavolo:deadlines"
# Channel announcing newly enqueued deadline entries, so every worker's
# consumer can wake immediately instead of polling on its heartbeat.
DEADLINES_CHANNEL = "tavolo:deadlines:events"
# 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. # Sentinel pushed into in-memory subscriber queues to signal a change.
_BUMP = b"update" _BUMP = b"update"
@@ -120,6 +129,14 @@ class GameStore(ABC):
async def find_by_code(self, code: str) -> Optional[GameSession]: async def find_by_code(self, code: str) -> Optional[GameSession]:
"""Return the live session for a join ``code`` or ``None``.""" """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 @abstractmethod
def lock(self, game_id: str) -> AsyncContextManager[None]: def lock(self, game_id: str) -> AsyncContextManager[None]:
"""Async context manager serializing mutations of one game.""" """Async context manager serializing mutations of one game."""
@@ -152,6 +169,18 @@ class GameStore(ABC):
async def remove_deadline(self, member: str) -> None: async def remove_deadline(self, member: str) -> None:
"""Remove ``member`` from the queue; a no-op when absent.""" """Remove ``member`` from the queue; a no-op when absent."""
@abstractmethod
async def notify_deadline(self) -> None:
"""Wake every worker's deadline consumer (best effort).
Notification is advisory: consumers stay correct when a signal
is lost, because their poll backstop recovers it.
"""
@abstractmethod
def subscribe_deadlines(self) -> AsyncContextManager[AsyncIterator[None]]:
"""Async context manager yielding deadline-enqueue signals."""
def _channel(game_id: str) -> str: def _channel(game_id: str) -> str:
return f"{CHANNEL_PREFIX}{game_id}:events" return f"{CHANNEL_PREFIX}{game_id}:events"
@@ -181,9 +210,15 @@ class RedisGameStore(GameStore):
async def save(self, session: GameSession) -> None: async def save(self, session: GameSession) -> None:
payload = json.dumps(session_to_json(session, self._registry)) 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: async with self._redis.pipeline(transaction=True) as pipe:
pipe.set(f"{GAME_KEY_PREFIX}{session.id}", payload, ex=self._ttl) 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) 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() await pipe.execute()
log.debug("redis save %s (game %s, ttl %ds)", session.id, session.game_type, self._ttl) log.debug("redis save %s (game %s, ttl %ds)", session.id, session.game_type, self._ttl)
@@ -195,6 +230,17 @@ class RedisGameStore(GameStore):
game_id = game_id.decode("utf-8") game_id = game_id.decode("utf-8")
return await self.load(str(game_id)) 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 @contextlib.asynccontextmanager
async def subscribe(self, game_id: str) -> AsyncIterator[AsyncIterator[None]]: async def subscribe(self, game_id: str) -> AsyncIterator[AsyncIterator[None]]:
pubsub = self._redis.pubsub() pubsub = self._redis.pubsub()
@@ -214,10 +260,17 @@ class RedisGameStore(GameStore):
await self._redis.zadd(DEADLINES_KEY, {member: due_at}) await self._redis.zadd(DEADLINES_KEY, {member: due_at})
async def due_deadlines(self, now: float, limit: int = 32) -> List[str]: async def due_deadlines(self, now: float, limit: int = 32) -> List[str]:
# ``ZRANGE ... BYSCORE`` is the non-deprecated spelling of the
# old ``ZRANGEBYSCORE``.
members = cast( members = cast(
list, list,
await self._redis.zrangebyscore( await self._redis.zrange(
DEADLINES_KEY, "-inf", now, start=0, num=limit DEADLINES_KEY,
"-inf",
now,
byscore=True,
offset=0,
num=limit,
), ),
) )
return [m.decode("utf-8") if isinstance(m, bytes) else m for m in members] return [m.decode("utf-8") if isinstance(m, bytes) else m for m in members]
@@ -229,6 +282,20 @@ class RedisGameStore(GameStore):
async def remove_deadline(self, member: str) -> None: async def remove_deadline(self, member: str) -> None:
await self._redis.zrem(DEADLINES_KEY, member) await self._redis.zrem(DEADLINES_KEY, member)
async def notify_deadline(self) -> None:
await self._redis.publish(DEADLINES_CHANNEL, "deadline")
@contextlib.asynccontextmanager
async def subscribe_deadlines(self) -> AsyncIterator[AsyncIterator[None]]:
pubsub = self._redis.pubsub()
await pubsub.subscribe(DEADLINES_CHANNEL)
try:
yield _redis_events(pubsub)
finally:
with contextlib.suppress(Exception):
await pubsub.unsubscribe(DEADLINES_CHANNEL)
await pubsub.aclose()
async def _redis_events(pubsub) -> AsyncIterator[None]: async def _redis_events(pubsub) -> AsyncIterator[None]:
async for message in pubsub.listen(): async for message in pubsub.listen():
@@ -243,9 +310,11 @@ class InMemoryGameStore(GameStore):
self._registry = registry self._registry = registry
self._games: Dict[str, str] = {} self._games: Dict[str, str] = {}
self._codes: Dict[str, str] = {} self._codes: Dict[str, str] = {}
self._ongoing: Set[str] = set()
self._locks: Dict[str, asyncio.Lock] = {} self._locks: Dict[str, asyncio.Lock] = {}
self._subscribers: Dict[str, Set[asyncio.Queue]] = {} self._subscribers: Dict[str, Set[asyncio.Queue]] = {}
self._deadlines: Dict[str, float] = {} self._deadlines: Dict[str, float] = {}
self._deadline_subscribers: Set[asyncio.Queue] = set()
def _lock_for(self, game_id: str) -> asyncio.Lock: def _lock_for(self, game_id: str) -> asyncio.Lock:
lock = self._locks.get(game_id) lock = self._locks.get(game_id)
@@ -268,6 +337,11 @@ class InMemoryGameStore(GameStore):
async def save(self, session: GameSession) -> None: async def save(self, session: GameSession) -> None:
self._games[session.id] = json.dumps(session_to_json(session, self._registry)) self._games[session.id] = json.dumps(session_to_json(session, self._registry))
self._codes[session.join_code] = session.id 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]: async def find_by_code(self, code: str) -> Optional[GameSession]:
game_id = self._codes.get(code.upper()) game_id = self._codes.get(code.upper())
@@ -275,6 +349,9 @@ class InMemoryGameStore(GameStore):
return None return None
return await self.load(game_id) return await self.load(game_id)
async def count_ongoing(self) -> int:
return len(self._ongoing)
@contextlib.asynccontextmanager @contextlib.asynccontextmanager
async def subscribe(self, game_id: str) -> AsyncIterator[AsyncIterator[None]]: async def subscribe(self, game_id: str) -> AsyncIterator[AsyncIterator[None]]:
queue: asyncio.Queue = asyncio.Queue() queue: asyncio.Queue = asyncio.Queue()
@@ -306,6 +383,19 @@ class InMemoryGameStore(GameStore):
async def remove_deadline(self, member: str) -> None: async def remove_deadline(self, member: str) -> None:
self._deadlines.pop(member, None) self._deadlines.pop(member, None)
async def notify_deadline(self) -> None:
for queue in list(self._deadline_subscribers):
queue.put_nowait(_BUMP)
@contextlib.asynccontextmanager
async def subscribe_deadlines(self) -> AsyncIterator[AsyncIterator[None]]:
queue: asyncio.Queue = asyncio.Queue()
self._deadline_subscribers.add(queue)
try:
yield _queue_events(queue)
finally:
self._deadline_subscribers.discard(queue)
async def _queue_events(queue: asyncio.Queue) -> AsyncIterator[None]: async def _queue_events(queue: asyncio.Queue) -> AsyncIterator[None]:
while True: while True:
@@ -2,12 +2,21 @@
Clients connect to ``/ws/games/{game_id}`` using their session cookie Clients connect to ``/ws/games/{game_id}`` using their session cookie
(the OIDC login stores the user in the session, which the session mixin (the OIDC login stores the user in the session, which the session mixin
loads onto the websocket). Only seated players are accepted. loads onto the websocket). Anyone who knows the game id is accepted:
seated players get their personalized view, everyone else — anonymous
users and authenticated users who are not seated — is a read-only
**spectator** receiving the view without privileged information (no
hands, no legal moves). The game view carries ``spectator: true`` for
them and never carries the join code. Spectator actions other than
``state``/``sync`` are rejected with a ``forbidden`` error; seating is
re-checked under the game lock so a viewer who joins mid-connection
immediately gains (and, if they somehow left, would lose) the right to
act.
The platform owns the connection lifecycle — authentication, seat The platform owns the connection lifecycle — authentication, the
check, the subscription/publish fan-out and the message envelope — and subscription/publish fan-out and the message envelope — and delegates
delegates the game-specific actions to the engine registered for the the game-specific actions to the engine registered for the session's
session's ``game_type``. ``game_type``.
Protocol Protocol
-------- --------
@@ -48,9 +57,9 @@ import asyncio
import json import json
from contextlib import suppress from contextlib import suppress
from logging import getLogger from logging import getLogger
from typing import TYPE_CHECKING, Any, Awaitable, Callable, Dict from typing import TYPE_CHECKING, Any, Awaitable, Callable, Dict, Optional
from kaya.core import KayaApp, WebSocket from kaya.core import KayaApp, WebSocket, WebSocketDisconnect
from . import auth from . import auth
from .engine import GameSession from .engine import GameSession
@@ -68,17 +77,36 @@ def _error(message: str, code: str = "invalid") -> Dict[str, Any]:
return {"type": "error", "code": code, "message": message} return {"type": "error", "code": code, "message": message}
def _state_message(platform: "Platform", session: GameSession, sub: str) -> Dict[str, Any]: async def _reject(ws: WebSocket, status: int) -> None:
"""Refuse a websocket upgrade with the HTTP ``status``.
Rejections happen before :meth:`WebSocket.accept`, so ``close()``
carries the HTTP status of the refused upgrade rather than a
WebSocket close code: ASGI servers answer ``403`` per the spec, and
RSGI (granian) uses the value verbatim as the response status.
Passing a 4xxx close code here would panic granian's RSGI layer on
the invalid status — and since granian is built with
``panic = "abort"``, that kills the worker and takes the whole app
down.
"""
await ws.close(status)
def _state_message(
platform: "Platform", session: GameSession, sub: Optional[str]
) -> Dict[str, Any]:
engine = platform.registry.require(session.game_type) engine = platform.registry.require(session.game_type)
return { seated = sub is not None and session.seated(sub)
"type": "state", payload: Dict[str, Any] = {
"game": { "id": session.id,
"id": session.id, "game_type": session.game_type,
"join_code": session.join_code, "spectator": not seated,
"game_type": session.game_type,
**engine.view_for(session, sub),
},
} }
if seated:
# The join code is a credential: only players get it.
payload["join_code"] = session.join_code
payload.update(engine.view_for(session, sub))
return {"type": "state", "game": payload}
def register(app: KayaApp, platform: "Platform") -> None: def register(app: KayaApp, platform: "Platform") -> None:
@@ -86,32 +114,33 @@ def register(app: KayaApp, platform: "Platform") -> None:
@app.websocket("/ws/games/${game_id}") @app.websocket("/ws/games/${game_id}")
async def game_socket(ws: WebSocket, game_id: str) -> None: async def game_socket(ws: WebSocket, game_id: str) -> None:
# Anonymous and non-seated viewers are welcome as read-only
# spectators; only an unknown game id is refused.
user = auth.get_ws_user(ws) user = auth.get_ws_user(ws)
if user is None: sub: Optional[str] = user.sub if user is not None else None
log.debug("websocket %s rejected: no authenticated user", game_id)
await ws.close(4401)
return
session = await platform.game_store.load(game_id) session = await platform.game_store.load(game_id)
if session is None: if session is None:
log.debug("websocket rejected: unknown game %s", game_id) log.debug("websocket rejected: unknown game %s", game_id)
await ws.close(4404) await _reject(ws, 404)
return
if not session.seated(user.sub):
log.debug("websocket %s rejected: %s is not seated", game_id, user.sub)
await ws.close(4403)
return return
await ws.accept() await ws.accept()
log.info("%s connected to game %s", user.sub, game_id) log.info("%s connected to game %s", sub or "spectator", game_id)
send_lock = asyncio.Lock() send_lock = asyncio.Lock()
async def send(payload: Dict[str, Any]) -> None: async def send(payload: Dict[str, Any]) -> None:
async with send_lock: async with send_lock:
await ws.send_text(json.dumps(payload)) try:
await ws.send_text(json.dumps(payload))
except WebSocketDisconnect:
# The peer went away without a close handshake: kaya
# raises this from the transport, and the receive loop
# ends momentarily.
pass
await send(_state_message(platform, session, user.sub)) await send(_state_message(platform, session, sub))
# Backstop: make sure the current deadline is queued even if its # Backstop: make sure the current deadline is queued even if its
# entry was lost (e.g. the queue was flushed while the game lived # entry was lost (e.g. the queue was flushed while the game lived
# on thanks to its sliding TTL). # on thanks to its sliding TTL).
@@ -119,7 +148,7 @@ def register(app: KayaApp, platform: "Platform") -> None:
async with platform.game_store.subscribe(game_id) as events: async with platform.game_store.subscribe(game_id) as events:
forward = asyncio.create_task( forward = asyncio.create_task(
_forward(platform, events, game_id, user.sub, send) _forward(platform, events, game_id, sub, send)
) )
try: try:
async for message in ws: async for message in ws:
@@ -128,19 +157,19 @@ def register(app: KayaApp, platform: "Platform") -> None:
if message.kind != "text" or not isinstance(message.data, str): if message.kind != "text" or not isinstance(message.data, str):
await send(_error("expected a text frame with a JSON object")) await send(_error("expected a text frame with a JSON object"))
continue continue
await _handle_message(platform, send, game_id, user.sub, message.data) await _handle_message(platform, send, game_id, sub, message.data)
finally: finally:
forward.cancel() forward.cancel()
with suppress(asyncio.CancelledError): with suppress(asyncio.CancelledError):
await forward await forward
log.debug("%s disconnected from game %s", user.sub, game_id) log.debug("%s disconnected from game %s", sub or "spectator", game_id)
async def _forward( async def _forward(
platform: "Platform", platform: "Platform",
events, events,
game_id: str, game_id: str,
sub: str, sub: Optional[str],
send: Send, send: Send,
) -> None: ) -> None:
async for _ in events: async for _ in events:
@@ -160,7 +189,7 @@ async def _forward(
async def _handle_message( async def _handle_message(
platform: "Platform", send: Send, game_id: str, sub: str, raw: str platform: "Platform", send: Send, game_id: str, sub: Optional[str], raw: str
) -> None: ) -> None:
try: try:
data = json.loads(raw) data = json.loads(raw)
@@ -189,7 +218,7 @@ async def _handle_action(
platform: "Platform", platform: "Platform",
send: Send, send: Send,
game_id: str, game_id: str,
sub: str, sub: Optional[str],
action: str, action: str,
data: Dict[str, Any], data: Dict[str, Any],
) -> None: ) -> None:
@@ -198,6 +227,13 @@ async def _handle_action(
if session is None: if session is None:
await send(_error("game not found", code="not_found")) await send(_error("game not found", code="not_found"))
return return
# Re-check seating under the lock: an anonymous spectator stays a
# spectator forever, while a logged-in viewer who joins the game
# mid-connection becomes a player and may act from then on.
if sub is None or not session.seated(sub):
log.debug("game %s: rejected %r from a spectator", game_id, action)
await send(_error("spectators cannot act on the game", code="forbidden"))
return
engine = platform.registry.require(session.game_type) engine = platform.registry.require(session.game_type)
payload = {key: value for key, value in data.items() if key != "action"} payload = {key: value for key, value in data.items() if key != "action"}
try: try:
@@ -8,7 +8,7 @@ without importing any real game. Its rules: the first player to reach
``target`` plays wins the match. ``target`` plays wins the match.
:func:`make_platform` builds a throwaway :class:`~kaya.core.KayaApp` :func:`make_platform` builds a throwaway :class:`~kaya.core.KayaApp`
wired with in-memory stores, a dummy OIDC mixin (patched per test) and wired with in-memory stores, dummy OIDC providers (patched per test) and
a sqlite :class:`~tavolo.platform.tortoise_mixin.TortoiseMixin`, so a sqlite :class:`~tavolo.platform.tortoise_mixin.TortoiseMixin`, so
tests construct their own app instead of importing a global one. tests construct their own app instead of importing a global one.
""" """
@@ -22,7 +22,7 @@ from functools import wraps
from typing import Any, Callable, Coroutine, Dict, Iterator, List, Mapping, Optional, Sequence, Tuple from typing import Any, Callable, Coroutine, Dict, Iterator, List, Mapping, Optional, Sequence, Tuple
from kaya.core import KayaApp from kaya.core import KayaApp
from kaya.oidc import OIDCConfig, OIDCMixin, OIDCUser from kaya.oidc import OIDCConfig, OIDCProviders, OIDCUser
from kaya.openapi import OpenAPIMixin from kaya.openapi import OpenAPIMixin
from kaya.session import InMemorySessionStore, SessionMixin from kaya.session import InMemorySessionStore, SessionMixin
@@ -138,7 +138,7 @@ class DummyEngine(GameEngine):
state["finished"] = True state["finished"] = True
state["winner"] = user_sub state["winner"] = user_sub
def view_for(self, session: GameSession, user_sub: str) -> Dict[str, Any]: def view_for(self, session: GameSession, user_sub: Optional[str]) -> Dict[str, Any]:
state = session.state state = session.state
return { return {
"started": state["started"], "started": state["started"],
@@ -146,7 +146,7 @@ class DummyEngine(GameEngine):
"winner": state["winner"], "winner": state["winner"],
"plays": len(state["plays"]), "plays": len(state["plays"]),
"target": state["target"], "target": state["target"],
"viewer_seated": session.seated(user_sub), "viewer_seated": user_sub is not None and session.seated(user_sub),
} }
def lobby_view(self, session: GameSession) -> Dict[str, Any]: def lobby_view(self, session: GameSession) -> Dict[str, Any]:
@@ -220,20 +220,26 @@ class DummyEngine(GameEngine):
def make_platform( def make_platform(
engines: Sequence[GameEngine] = (DummyEngine(),), engines: Sequence[GameEngine] = (DummyEngine(),),
heartbeat_ms: int = 50,
oidc_providers: Sequence[str] = ("default",),
) -> Tuple[KayaApp, Platform, TortoiseMixin]: ) -> Tuple[KayaApp, Platform, TortoiseMixin]:
"""Build a throwaway app + platform wired with in-memory stores.""" """Build a throwaway app + platform wired with in-memory stores."""
registry = GameRegistry(engines) registry = GameRegistry(engines)
game_store = InMemoryGameStore(registry) game_store = InMemoryGameStore(registry)
session_mixin = SessionMixin(InMemorySessionStore()) session_mixin = SessionMixin(InMemorySessionStore())
oidc_mixin = OIDCMixin( oidc = OIDCProviders(
OIDCConfig( [
issuer="http://localhost:8180/tavolo", OIDCConfig(
client_id="tavolo", provider=name,
client_secret=None, issuer=f"http://localhost:8180/{name}",
redirect_uri="http://localhost:8080/auth/callback", client_id="tavolo",
post_login_redirect="/", client_secret=None,
post_logout_redirect="/", redirect_uri=f"http://localhost:8080/auth/{name}/callback",
), post_login_redirect="/",
post_logout_redirect="/",
)
for name in oidc_providers
],
session=session_mixin, session=session_mixin,
) )
tortoise_mixin = TortoiseMixin( tortoise_mixin = TortoiseMixin(
@@ -251,19 +257,19 @@ def make_platform(
scheduler = DeadlineScheduler( scheduler = DeadlineScheduler(
game_store, game_store,
registry, registry,
heartbeat_ms=50, heartbeat_ms=heartbeat_ms,
context_binder=tortoise_mixin.ensure_context, context_binder=tortoise_mixin.ensure_context,
) )
platform = Platform( platform = Platform(
registry=registry, registry=registry,
game_store=game_store, game_store=game_store,
scheduler=scheduler, scheduler=scheduler,
oidc=oidc_mixin, oidc=oidc,
) )
app = KayaApp( app = KayaApp(
mixins=[ mixins=[
session_mixin, session_mixin,
oidc_mixin, oidc,
tortoise_mixin, tortoise_mixin,
openapi_mixin, openapi_mixin,
PlatformMixin(platform), PlatformMixin(platform),
@@ -319,10 +325,12 @@ def make_user(sub: str, name: Optional[str] = None) -> OIDCUser:
@contextlib.contextmanager @contextlib.contextmanager
def oidc_user(oidc: OIDCMixin, sub: str, name: Optional[str] = None) -> Iterator[OIDCUser]: def oidc_user(oidc: OIDCProviders, sub: str, name: Optional[str] = None) -> Iterator[OIDCUser]:
"""Context manager: patch ``oidc.get_user`` to return this user.""" """Context manager: make ``oidc.get_users`` return this user for the
first configured provider."""
user = make_user(sub, name) user = make_user(sub, name)
patcher = _mock.patch.object(oidc, "get_user", return_value=user) provider = next(iter(oidc.providers))
patcher = _mock.patch.object(oidc, "get_users", return_value={provider: user})
patcher.start() patcher.start()
try: try:
yield user yield user
@@ -0,0 +1,152 @@
"""Tests for the auth helpers' OIDC user lookup and gating.
kaya-oidc (0.0.6) namespaces session state per provider as
``oidc:{provider}:user`` and tavolo wires an :class:`OIDCProviders`; these
tests pin both contracts so a kaya upgrade or a provider change cannot
silently break HTTP or websocket authentication.
"""
from __future__ import annotations
import unittest
from types import SimpleNamespace
from typing import Any
from kaya.oidc import OIDCConfig, OIDCProviders
from kaya.session import InMemorySessionStore, Session, SessionMixin
from helpers import async_test
from tavolo.platform.auth import current_user, get_ws_user, require_auth, session_user
def _providers(*names: str) -> OIDCProviders:
session = SessionMixin(InMemorySessionStore())
return OIDCProviders(
[
OIDCConfig(
provider=name,
issuer=f"http://localhost:8180/{name}",
client_id="tavolo",
redirect_uri=f"http://localhost:8080/auth/{name}/callback",
)
for name in names
],
session=session,
)
def _ctx(data: Any) -> Any:
return SimpleNamespace(session=Session("sid", data))
def _ws(session: Any) -> Any:
return SimpleNamespace(session=session)
class SessionUserTest(unittest.TestCase):
def test_finds_default_provider_user(self) -> None:
user = session_user(Session("sid", {"oidc:default:user": {"sub": "alice"}}))
assert user is not None
self.assertEqual("alice", user.sub)
def test_finds_user_of_any_provider(self) -> None:
user = session_user(Session("sid", {"oidc:google:user": {"sub": "alice"}}))
assert user is not None
self.assertEqual("alice", user.sub)
def test_ignores_flow_state_and_tokens(self) -> None:
session = Session(
"sid",
{
"oidc:default:state": "state",
"oidc:default:id_token": "token",
"oidc:default:access_token": "token",
},
)
self.assertIsNone(session_user(session))
def test_ignores_legacy_unprefixed_user_key(self) -> None:
session = Session("sid", {"oidc_user": {"sub": "alice"}})
self.assertIsNone(session_user(session))
def test_ignores_malformed_user_claims(self) -> None:
session = Session("sid", {"oidc:default:user": "alice"})
self.assertIsNone(session_user(session))
class GetWsUserTest(unittest.TestCase):
def test_returns_user_from_provider_namespaced_session(self) -> None:
session = Session(
"sid",
{"oidc:default:user": {"sub": "alice", "preferred_username": "Alice"}},
)
user = get_ws_user(_ws(session))
assert user is not None
self.assertEqual("alice", user.sub)
def test_missing_session_returns_none(self) -> None:
self.assertIsNone(get_ws_user(SimpleNamespace()))
class CurrentUserTest(unittest.TestCase):
def test_anonymous_session_has_no_user(self) -> None:
self.assertIsNone(current_user(_providers("default"), _ctx({})))
def test_returns_the_authenticated_provider_user(self) -> None:
user = current_user(
_providers("default", "google"),
_ctx({"oidc:google:user": {"sub": "alice"}}),
)
assert user is not None
self.assertEqual("alice", user.sub)
def test_first_configured_provider_wins(self) -> None:
# Single-identity sessions never hold two users; this only pins the
# tie-break should they ever coexist.
user = current_user(
_providers("default", "google"),
_ctx(
{
"oidc:default:user": {"sub": "alice"},
"oidc:google:user": {"sub": "bob"},
}
),
)
assert user is not None
self.assertEqual("alice", user.sub)
class RequireAuthTest(unittest.TestCase):
@async_test
async def test_rejects_anonymous_requests(self) -> None:
oidc = _providers("default")
ctx = _ctx({})
sent = {}
async def send_bytes(*args: Any) -> None:
sent["status"] = args[0]
ctx.send_bytes = send_bytes
handled = []
@require_auth(oidc)
async def handler(_ctx: Any) -> None:
handled.append(True)
await handler(ctx)
self.assertEqual(401, sent["status"])
self.assertEqual([], handled)
@async_test
async def test_allows_authenticated_requests(self) -> None:
oidc = _providers("default")
ctx = _ctx({"oidc:default:user": {"sub": "alice"}})
handled = []
@require_auth(oidc)
async def handler(request_ctx: Any) -> None:
user = current_user(oidc, request_ctx)
assert user is not None
handled.append(user.sub)
await handler(ctx)
self.assertEqual(["alice"], handled)
@@ -112,6 +112,38 @@ async def _finished(store, game_id: str) -> Optional[GameSession]:
return None return None
class NotificationTest(unittest.TestCase):
"""Deadline-enqueue notifications wake consumers on other workers."""
@async_test
async def test_notification_wakes_a_long_heartbeat_consumer(self) -> None:
# The consumer's heartbeat is a minute away and nothing sets its
# local wake event: only the pub/sub notification can deliver the
# deadline in time. This is the cross-worker path.
_, platform, _ = make_platform(heartbeat_ms=60_000)
store = platform.game_store
scheduler = platform.scheduler
session = _started_session(deadline_in_seconds=0.1)
await store.save(session)
scheduler.ensure_consumer()
# Let the listener task subscribe before announcing.
await asyncio.sleep(0.05)
engine = platform.registry.require(session.game_type)
deadline = engine.next_deadline(session)
assert deadline is not None
member = encode({
"game_id": session.id,
"kind": deadline.kind,
"token": deadline.token,
})
await store.add_deadline(member, deadline.due_at.timestamp())
await store.notify_deadline()
result = await _wait_for(lambda: _plays_is(store, session.id, 1))
self.assertIsNotNone(result, "notification did not wake the consumer")
class ProcessDueTest(unittest.TestCase): class ProcessDueTest(unittest.TestCase):
"""Direct ``process_due`` behaviour: engine revalidation and idempotency.""" """Direct ``process_due`` behaviour: engine revalidation and idempotency."""
@@ -0,0 +1,81 @@
"""Keyset pagination tests: cursor values are restored to Python types.
Cursors are JSON, so a cursor carrying a datetime reaches the next
request as a string. Postgres/asyncpg rejects strings for timestamp
parameters (sqlite silently coerces them), so ``paginate`` must run the
decoded values through their model fields before building the WHERE
clause.
"""
from __future__ import annotations
import unittest
import uuid
from datetime import datetime, timezone
from typing import Any, List
from tavolo.platform.models import Match
from tavolo.platform.pagination import Cursor, decode_cursor, encode_cursor, paginate
from helpers import async_test
SORT = [("finished_at", "DESC"), ("id", "ASC")]
class _CapturingQuerySet:
"""Minimal stand-in for a Tortoise QuerySet, recording its filter."""
model = Match
def __init__(self) -> None:
self.queryset_filter: Any = None
def order_by(self, *orderings: str) -> "_CapturingQuerySet":
return self
def filter(self, q: Any) -> "_CapturingQuerySet":
self.queryset_filter = q
return self
async def limit(self, count: int) -> List[Any]:
return []
def _flatten_q(q: Any) -> List[Any]:
nodes = [q]
for child in q.children:
nodes.extend(_flatten_q(child))
return nodes
def _filter_values(q: Any, key: str) -> List[Any]:
return [node.filters[key] for node in _flatten_q(q) if key in node.filters]
class PaginateCursorTypesTest(unittest.TestCase):
@async_test
async def test_cursor_values_are_restored_before_filtering(self) -> None:
match_id = uuid.uuid4()
finished_at = datetime(2026, 9, 19, 5, 2, 50, 547904, tzinfo=timezone.utc)
# Round-trip through the wire format, exactly as a request does.
after = decode_cursor(
encode_cursor({"finished_at": finished_at, "id": match_id})
)
assert after is not None
qs = _CapturingQuerySet()
await paginate(qs, SORT, Cursor(limit=1, after=after))
assert qs.queryset_filter is not None
for key in ("finished_at", "finished_at__lt"):
values = _filter_values(qs.queryset_filter, key)
self.assertEqual(1, len(values))
self.assertIsInstance(values[0], datetime)
self.assertEqual(finished_at, values[0])
id_values = _filter_values(qs.queryset_filter, "id__gt")
self.assertEqual(1, len(id_values))
self.assertIsInstance(id_values[0], uuid.UUID)
self.assertEqual(match_id, id_values[0])
if __name__ == "__main__":
unittest.main()
@@ -44,10 +44,24 @@ class GamesRouteTest(unittest.TestCase):
self.assertEqual(game_id, snap["id"]) self.assertEqual(game_id, snap["id"])
self.assertEqual("dummy", snap["game_type"]) self.assertEqual("dummy", snap["game_type"])
self.assertTrue(snap["viewer_seated"]) self.assertTrue(snap["viewer_seated"])
self.assertFalse(snap["spectator"])
self.assertIn("join_code", snap)
# A logged-in non-player is a read-only spectator: the view
# marks it and the join code is withheld.
with oidc_user(platform.oidc, "mallory"): with oidc_user(platform.oidc, "mallory"):
forbidden = await client.get(f"/api/games/{game_id}") watched = await client.get(f"/api/games/{game_id}")
self.assertEqual(403, forbidden.status_code) self.assertEqual(200, watched.status_code)
spect = watched.json()
self.assertFalse(spect["viewer_seated"])
self.assertTrue(spect["spectator"])
self.assertNotIn("join_code", spect)
# Anonymous users may watch too.
anonymous = await client.get(f"/api/games/{game_id}")
self.assertEqual(200, anonymous.status_code)
self.assertTrue(anonymous.json()["spectator"])
self.assertNotIn("join_code", anonymous.json())
@async_test @async_test
async def test_join_starts_game(self) -> None: async def test_join_starts_game(self) -> None:
@@ -0,0 +1,67 @@
"""Tests for the provider-agnostic auth routes.
``GET /api/auth/providers`` feeds the SPA's login buttons; ``/auth/login``
and ``/auth/logout`` are the generic aliases that redirect to the right
per-provider kaya route.
"""
from __future__ import annotations
import unittest
from unittest import mock
from httpx import ASGITransport, AsyncClient
from helpers import async_test, make_platform
class AuthProvidersRouteTest(unittest.TestCase):
@async_test
async def test_lists_providers_in_configuration_order(self) -> None:
app, _platform, _tortoise = make_platform(oidc_providers=("default", "google"))
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://127.0.0.1") as client:
response = await client.get("/api/auth/providers")
self.assertEqual(200, response.status_code)
self.assertEqual(
{
"providers": [
{"name": "default", "login_url": "/auth/default/login"},
{"name": "google", "login_url": "/auth/google/login"},
]
},
response.json(),
)
class GenericAuthRedirectTest(unittest.TestCase):
@async_test
async def test_login_redirects_to_the_first_provider(self) -> None:
app, _platform, _tortoise = make_platform(oidc_providers=("google", "github"))
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://127.0.0.1") as client:
response = await client.get("/auth/login", follow_redirects=False)
self.assertEqual(302, response.status_code)
self.assertEqual("/auth/google/login", response.headers["location"])
@async_test
async def test_logout_redirects_to_the_logged_in_provider(self) -> None:
app, platform, _tortoise = make_platform()
with mock.patch.object(
platform.oidc, "logged_in_providers", return_value=["default"]
):
transport = ASGITransport(app=app)
async with AsyncClient(
transport=transport, base_url="http://127.0.0.1"
) as client:
response = await client.get("/auth/logout", follow_redirects=False)
self.assertEqual(302, response.status_code)
self.assertEqual("/auth/default/logout", response.headers["location"])
@async_test
async def test_logout_without_a_session_redirects_home(self) -> None:
app, _platform, _tortoise = make_platform()
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://127.0.0.1") as client:
response = await client.get("/auth/logout", follow_redirects=False)
self.assertEqual(302, response.status_code)
self.assertEqual("/", response.headers["location"])
@@ -4,6 +4,7 @@ from __future__ import annotations
import unittest import unittest
import uuid import uuid
from datetime import datetime, timezone from datetime import datetime, timezone
from unittest import mock
from httpx import ASGITransport, AsyncClient from httpx import ASGITransport, AsyncClient
@@ -197,6 +198,28 @@ class StatsRouteTest(unittest.TestCase):
second.json()["results"][0]["id"], second.json()["results"][0]["id"],
) )
@async_test
async def test_my_matches_fetches_participants_in_one_query(self) -> None:
app, platform, tortoise_mixin = make_platform()
await _seed_two_matches(tortoise_mixin)
transport = ASGITransport(app=app)
with mock.patch.object(
MatchPlayer, "filter", wraps=MatchPlayer.filter
) as spy:
async with AsyncClient(
transport=transport, base_url="http://127.0.0.1"
) as client:
with oidc_user(platform.oidc, "alice"):
response = await client.get("/api/me/matches")
self.assertEqual(200, response.status_code)
# One query for all matches, not one per match row.
self.assertEqual(1, spy.call_count)
results = response.json()["results"]
self.assertEqual(2, len(results))
for result in results:
seats = [player["seat"] for player in result["players"]]
self.assertEqual([0, 1], seats)
@async_test @async_test
async def test_my_matches_requires_auth(self) -> None: async def test_my_matches_requires_auth(self) -> None:
app, _, _ = make_platform() app, _, _ = make_platform()
@@ -2,11 +2,18 @@
from __future__ import annotations from __future__ import annotations
import asyncio import asyncio
import time
import unittest import unittest
from typing import Any, Optional, cast
from tavolo.platform import GameSession, Seat from tavolo.platform import GameSession, Seat
from tavolo.platform.registry import GameRegistry from tavolo.platform.registry import GameRegistry
from tavolo.platform.store import InMemoryGameStore from tavolo.platform.store import (
DEADLINES_KEY,
ONGOING_GAMES_KEY,
InMemoryGameStore,
RedisGameStore,
)
from helpers import DummyEngine, async_test from helpers import DummyEngine, async_test
@@ -27,6 +34,110 @@ def _session(game_id: str = "g1", code: str = "CODE01") -> GameSession:
return session 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 = {}
self.zrange_calls: list = []
def pipeline(self, transaction: bool = True) -> _FakePipeline:
return _FakePipeline(self)
async def zadd(self, key: str, mapping: dict) -> int:
return cast(int, self.apply(("zadd", key, mapping)))
async def zrem(self, key: str, member: str) -> int:
return cast(int, self.apply(("zrem", key, member)))
async def zrange(
self,
name: str,
start: Any,
end: Any,
byscore: bool = False,
offset: Optional[int] = None,
num: Optional[int] = None,
**kwargs: object,
) -> list:
self.zrange_calls.append((name, start, end, byscore, offset, num))
members = sorted(
self.zsets.get(name, {}).items(), key=lambda item: item[1]
)
if byscore:
low = float("-inf") if start == "-inf" else float(start)
high = float("inf") if end == "+inf" else float(end)
members = [item for item in members if low <= item[1] <= high]
if offset is not None and num is not None:
members = members[offset : offset + num]
return [member for member, _ in members]
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): class InMemoryGameStoreTest(unittest.TestCase):
@async_test @async_test
async def test_save_load_roundtrip(self) -> None: async def test_save_load_roundtrip(self) -> None:
@@ -114,6 +225,28 @@ class InMemoryGameStoreTest(unittest.TestCase):
["holder-enter", "holder-exit", "contender"], order ["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_test
async def test_deadline_queue(self) -> None: async def test_deadline_queue(self) -> None:
store = InMemoryGameStore(_registry()) store = InMemoryGameStore(_registry())
@@ -137,6 +270,65 @@ class InMemoryGameStoreTest(unittest.TestCase):
self.assertEqual(60.0, await store.next_deadline()) self.assertEqual(60.0, await store.next_deadline())
self.assertEqual(["b"], await store.due_deadlines(now=100.0)) self.assertEqual(["b"], await store.due_deadlines(now=100.0))
@async_test
async def test_deadline_notifications_reach_subscribers(self) -> None:
store = InMemoryGameStore(_registry())
async with store.subscribe_deadlines() as events:
await store.notify_deadline()
signal = await asyncio.wait_for(anext(events), timeout=1.0)
self.assertIsNone(signal)
# The subscription is gone after the context exits: notifying
# again is a no-op rather than an error.
await store.notify_deadline()
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])
@async_test
async def test_due_deadlines_uses_zrange_byscore(self) -> None:
redis = _FakeRedis()
store = RedisGameStore(redis, _registry()) # type: ignore[arg-type]
await store.add_deadline("b", due_at=50.0)
await store.add_deadline("a", due_at=10.0)
await store.add_deadline("c", due_at=200.0)
# Re-adding an existing member only updates its due time.
await store.add_deadline("b", due_at=60.0)
self.assertEqual(["a"], await store.due_deadlines(now=20.0))
self.assertEqual(
[(DEADLINES_KEY, "-inf", 20.0, True, 0, 32)], redis.zrange_calls
)
# Matches come back in due-time order and honour the limit.
self.assertEqual(["a"], await store.due_deadlines(now=100.0, limit=1))
self.assertEqual(["a", "b"], await store.due_deadlines(now=100.0))
if __name__ == "__main__": if __name__ == "__main__":
unittest.main() unittest.main()
+169 -16
View File
@@ -1,15 +1,92 @@
"""WebSocket live-play tests via kaya's ASGI websocket transport.""" """WebSocket live-play tests via kaya's ASGI websocket transport."""
from __future__ import annotations from __future__ import annotations
import json
import unittest import unittest
from typing import Any, Dict, Mapping, Sequence
from httpx import ASGITransport, AsyncClient from httpx import ASGITransport, AsyncClient
from httpx_ws import WebSocketDisconnect, aconnect_ws from httpx_ws import WebSocketDisconnect as HttpxWebSocketDisconnect, aconnect_ws
from httpx_ws.transport import ASGIWebSocketTransport from httpx_ws.transport import ASGIWebSocketTransport
from kaya.core import WebSocket, WebSocketDisconnect, WebSocketMessage
from helpers import async_test, make_platform, make_user, oidc_user, ws_users from helpers import async_test, make_platform, make_user, oidc_user, ws_users
class AbruptCloseSocket(WebSocket):
"""A socket whose peer vanished without a close handshake.
kaya surfaces this from the transport: ``receive()`` reports a
``close`` message so iteration ends normally, while sends raise
``WebSocketDisconnect``.
"""
def __init__(self, path: str, gone: bool = False) -> None:
self.path = path
self.query_string = ""
self.scheme = "ws"
self.headers: Mapping[str, Sequence[str]] = {}
self.client = ("127.0.0.1", 12345)
self.server = ("127.0.0.1", 80)
self.sent: list[Dict[str, Any]] = []
self.close_codes: list[int] = []
self.gone = gone
async def accept(self, headers: Any = None) -> None:
pass
async def receive(self) -> WebSocketMessage:
self.gone = True
return WebSocketMessage(kind="close")
async def send_text(self, data: str) -> None:
if self.gone:
raise WebSocketDisconnect(code=1006)
self.sent.append(json.loads(data))
async def send_bytes(self, data: bytes) -> None:
if self.gone:
raise WebSocketDisconnect(code=1006)
async def close(self, code: int = 1000) -> None:
self.close_codes.append(code)
async def __anext__(self) -> WebSocketMessage:
await self.receive()
raise StopAsyncIteration
class RecordingSocket(WebSocket):
"""A socket that only records pre-accept rejections."""
def __init__(self, path: str) -> None:
self.path = path
self.query_string = ""
self.scheme = "ws"
self.headers: Mapping[str, Sequence[str]] = {}
self.client = ("127.0.0.1", 12345)
self.server = ("127.0.0.1", 80)
self.close_codes: list[int] = []
async def accept(self, headers: Any = None) -> None:
raise AssertionError("a rejected socket must not be accepted")
async def receive(self) -> WebSocketMessage:
raise AssertionError("a rejected socket must not receive")
async def send_text(self, data: str) -> None:
raise AssertionError("a rejected socket must not send")
async def send_bytes(self, data: bytes) -> None:
raise AssertionError("a rejected socket must not send")
async def close(self, code: int = 1000) -> None:
self.close_codes.append(code)
async def __anext__(self) -> WebSocketMessage:
raise AssertionError("a rejected socket must not receive")
class WebSocketTest(unittest.TestCase): class WebSocketTest(unittest.TestCase):
async def _started_game(self, client: AsyncClient, oidc, target: int = 3) -> dict: async def _started_game(self, client: AsyncClient, oidc, target: int = 3) -> dict:
"""Create a game and seat both players; return the started state.""" """Create a game and seat both players; return the started state."""
@@ -39,6 +116,9 @@ class WebSocketTest(unittest.TestCase):
self.assertEqual("dummy", first["game"]["game_type"]) self.assertEqual("dummy", first["game"]["game_type"])
self.assertEqual(0, first["game"]["plays"]) self.assertEqual(0, first["game"]["plays"])
self.assertEqual(game_id, first["game"]["id"]) self.assertEqual(game_id, first["game"]["id"])
# Seated players get the join code; spectators don't.
self.assertIn("join_code", first["game"])
self.assertFalse(first["game"]["spectator"])
async with aconnect_ws(f"/ws/games/{game_id}", ws_client) as alice_ws: async with aconnect_ws(f"/ws/games/{game_id}", ws_client) as alice_ws:
alice_first = await alice_ws.receive_json() alice_first = await alice_ws.receive_json()
@@ -87,6 +167,38 @@ class WebSocketTest(unittest.TestCase):
resent = await bob_ws.receive_json() resent = await bob_ws.receive_json()
self.assertEqual("state", resent["type"]) self.assertEqual("state", resent["type"])
@async_test
async def test_abrupt_disconnect_is_not_an_error(self) -> None:
app, platform, _ = make_platform()
api_transport = ASGITransport(app=app)
async with AsyncClient(transport=api_transport, base_url="http://127.0.0.1") as client:
state = await self._started_game(client, platform.oidc)
game_id = state["id"]
ws = AbruptCloseSocket(f"/ws/games/{game_id}")
with ws_users([make_user("bob")]):
# kaya reports a peer that vanished without a close handshake as
# a close message; the handler must exit cleanly.
await app.handle_websocket(ws)
self.assertEqual("state", ws.sent[0]["type"])
self.assertEqual([], ws.close_codes)
@async_test
async def test_send_on_vanished_peer_is_ignored(self) -> None:
app, platform, _ = make_platform()
api_transport = ASGITransport(app=app)
async with AsyncClient(transport=api_transport, base_url="http://127.0.0.1") as client:
state = await self._started_game(client, platform.oidc)
game_id = state["id"]
ws = AbruptCloseSocket(f"/ws/games/{game_id}", gone=True)
with ws_users([make_user("bob")]):
# Sending raises WebSocketDisconnect; the handler swallows it
# rather than letting it escape the forward task.
await app.handle_websocket(ws)
self.assertEqual([], ws.sent)
self.assertEqual([], ws.close_codes)
@async_test @async_test
async def test_game_over_broadcast(self) -> None: async def test_game_over_broadcast(self) -> None:
app, platform, _ = make_platform() app, platform, _ = make_platform()
@@ -121,13 +233,13 @@ class WebSocketTest(unittest.TestCase):
ws_transport = ASGIWebSocketTransport(app=app) ws_transport = ASGIWebSocketTransport(app=app)
async with AsyncClient(transport=ws_transport, base_url="http://testserver") as ws_client: async with AsyncClient(transport=ws_transport, base_url="http://testserver") as ws_client:
with ws_users([make_user("alice")]): with ws_users([make_user("alice")]):
with self.assertRaises(WebSocketDisconnect) as caught: with self.assertRaises(HttpxWebSocketDisconnect) as caught:
async with aconnect_ws("/ws/games/no-such-game", ws_client): async with aconnect_ws("/ws/games/no-such-game", ws_client):
pass pass
self.assertEqual(4404, caught.exception.code) self.assertEqual(404, caught.exception.code)
@async_test @async_test
async def test_non_player_is_closed(self) -> None: async def test_non_player_watches_as_spectator(self) -> None:
app, platform, _ = make_platform() app, platform, _ = make_platform()
api_transport = ASGITransport(app=app) api_transport = ASGITransport(app=app)
async with AsyncClient(transport=api_transport, base_url="http://127.0.0.1") as client: async with AsyncClient(transport=api_transport, base_url="http://127.0.0.1") as client:
@@ -137,21 +249,62 @@ class WebSocketTest(unittest.TestCase):
ws_transport = ASGIWebSocketTransport(app=app) ws_transport = ASGIWebSocketTransport(app=app)
async with AsyncClient(transport=ws_transport, base_url="http://testserver") as ws_client: async with AsyncClient(transport=ws_transport, base_url="http://testserver") as ws_client:
with ws_users([make_user("mallory")]): with ws_users([make_user("mallory")]):
with self.assertRaises(WebSocketDisconnect) as caught: async with aconnect_ws(f"/ws/games/{game_id}", ws_client) as ws:
async with aconnect_ws(f"/ws/games/{game_id}", ws_client): first = await ws.receive_json()
pass self.assertEqual("state", first["type"])
self.assertEqual(4403, caught.exception.code) self.assertTrue(first["game"]["spectator"])
self.assertFalse(first["game"]["viewer_seated"])
self.assertNotIn("join_code", first["game"])
# Spectators may not act.
await ws.send_json({"action": "play"})
error = await ws.receive_json()
self.assertEqual("error", error["type"])
self.assertEqual("forbidden", error["code"])
@async_test @async_test
async def test_unauthenticated_is_closed(self) -> None: async def test_anonymous_user_watches_as_spectator(self) -> None:
app, platform, _ = make_platform()
api_transport = ASGITransport(app=app)
async with AsyncClient(transport=api_transport, base_url="http://127.0.0.1") as client:
state = await self._started_game(client, platform.oidc)
game_id = state["id"]
ws_transport = ASGIWebSocketTransport(app=app)
async with AsyncClient(transport=ws_transport, base_url="http://testserver") as ws_client:
with ws_users([]):
async with aconnect_ws(f"/ws/games/{game_id}", ws_client) as ws:
first = await ws.receive_json()
self.assertEqual("state", first["type"])
self.assertTrue(first["game"]["spectator"])
self.assertNotIn("join_code", first["game"])
await ws.send_json({"action": "play"})
error = await ws.receive_json()
self.assertEqual("error", error["type"])
self.assertEqual("forbidden", error["code"])
class RejectionStatusTest(unittest.TestCase):
"""Pre-accept rejections must carry HTTP statuses, not close codes.
Granian's RSGI layer uses the value passed to ``close()`` before
``accept()`` as the HTTP status of the refused upgrade; a WebSocket
close code like 4401 is not a valid status and panics the worker
(``InvalidStatusCode``, and with ``panic = "abort"`` the whole app
dies). Drive the rejection through the app with a recording socket
so the invariant is checked without a real server. Spectators are
accepted, so the only pre-accept rejection left is an unknown game.
"""
@async_test
async def test_unknown_game_rejection_is_an_http_status(self) -> None:
app, _, _ = make_platform() app, _, _ = make_platform()
ws_transport = ASGIWebSocketTransport(app=app) for label, users in (("anonymous", []), ("logged in", [make_user("alice")])):
async with AsyncClient(transport=ws_transport, base_url="http://testserver") as ws_client: ws = RecordingSocket("/ws/games/no-such-game")
with ws_users([]): with ws_users(users):
with self.assertRaises(WebSocketDisconnect) as caught: await app.handle_websocket(ws)
async with aconnect_ws("/ws/games/whatever", ws_client): self.assertEqual([404], ws.close_codes, label)
pass for code in ws.close_codes:
self.assertEqual(4401, caught.exception.code) self.assertTrue(100 <= code <= 599, f"{label}: {code} is not an HTTP status")
if __name__ == "__main__": if __name__ == "__main__":
@@ -528,13 +528,14 @@ def hand_points(state: ScoponeState) -> Tuple[List[int], Dict[str, object]]:
return points, details return points, details
def state_for_player(state: ScoponeState, sub: str) -> Dict[str, object]: def state_for_player(state: ScoponeState, sub: Optional[str]) -> Dict[str, object]:
"""Serialize ``state`` hiding other players' hands. """Serialize ``state`` hiding other players' hands.
Hands are reduced to a count, except for the requesting player's own Hands are reduced to a count, except for the requesting player's own
hand. Non-seated viewers simply see no hand at all. The platform hand. Non-seated viewers — including spectators, for whom ``sub`` is
merges the session envelope (``id``, ``join_code``, ``game_type``) ``None`` — simply see no hand at all. The platform merges the session
into the view itself. envelope (``id``, ``join_code`` for seated players, ``game_type``,
``spectator``) into the view itself.
""" """
viewer = state.player_for(sub) viewer = state.player_for(sub)
players: List[Dict[str, object]] = [] players: List[Dict[str, object]] = []
@@ -171,13 +171,13 @@ class ScoponeEngine(GameEngine):
else: else:
raise IllegalMove(f"unknown action: {action!r}") raise IllegalMove(f"unknown action: {action!r}")
def view_for(self, session: GameSession, user_sub: str) -> Dict[str, Any]: def view_for(self, session: GameSession, user_sub: Optional[str]) -> Dict[str, Any]:
return engine.state_for_player(session.state, user_sub) return engine.state_for_player(session.state, user_sub)
def is_finished(self, session: GameSession) -> bool: def is_finished(self, session: GameSession) -> bool:
return session.state.phase == PHASE_FINISHED return session.state.phase == PHASE_FINISHED
def game_over_view(self, session: GameSession, user_sub: str) -> Dict[str, Any]: def game_over_view(self, session: GameSession, user_sub: Optional[str]) -> Dict[str, Any]:
state: ScoponeState = session.state state: ScoponeState = session.state
return { return {
"scores": {"A": state.scores[0], "B": state.scores[1]}, "scores": {"A": state.scores[0], "B": state.scores[1]},
@@ -239,7 +239,7 @@ class ScoponeState:
# -- helpers ---------------------------------------------------------- # -- helpers ----------------------------------------------------------
def player_for(self, sub: str) -> Optional[PlayerState]: def player_for(self, sub: Optional[str]) -> Optional[PlayerState]:
for player in self.players: for player in self.players:
if player.sub == sub: if player.sub == sub:
return player return player
@@ -574,6 +574,27 @@ class MatchFlowTest(unittest.TestCase):
self.assertEqual(["07C"], view["table"]) self.assertEqual(["07C"], view["table"])
self.assertTrue(view.get("your_turn")) self.assertTrue(view.get("your_turn"))
def test_state_for_player_spectator_sees_public_info_only(self) -> None:
state = make_state(
[["02D", "03C"], ["04D"], ["05D"], ["06D"]],
table=["07C"],
captured=[["01D"], [], ["02C"], ["03B"]],
scope=[1, 0, 2, 0],
)
view = engine.state_for_player(state, None)
for player in view["players"]:
self.assertNotIn("hand", player)
self.assertNotIn("your_turn", view)
self.assertNotIn("legal_moves", view)
# The public table state is still fully visible.
self.assertEqual(["07C"], view["table"])
self.assertEqual({"A": 0, "B": 0}, view["scores"])
players = {p["seat"]: p for p in view["players"]}
self.assertEqual(1, players[0]["captured_count"])
self.assertEqual(1, players[0]["scope"])
self.assertEqual(2, players[2]["scope"])
self.assertEqual(1, players[1]["cards_left"])
def test_legal_moves_only_for_player_on_turn(self) -> None: def test_legal_moves_only_for_player_on_turn(self) -> None:
state = make_state([["02D", "03C"], ["04D"], ["05D"], ["06D"]], state = make_state([["02D", "03C"], ["04D"], ["05D"], ["06D"]],
table=["07C"]) table=["07C"])
@@ -160,6 +160,17 @@ class ViewTest(unittest.TestCase):
self.assertNotIn("join_code", view) self.assertNotIn("join_code", view)
self.assertNotIn("game_type", view) self.assertNotIn("game_type", view)
def test_view_for_spectator_hides_all_hands(self) -> None:
engine = ScoponeEngine()
session = _started(engine)
view = engine.view_for(session, None)
for player in view["players"]:
self.assertNotIn("hand", player)
self.assertNotIn("your_turn", view)
self.assertNotIn("legal_moves", view)
self.assertIn("table", view)
self.assertEqual({"A": 0, "B": 0}, view["scores"])
def test_lobby_view(self) -> None: def test_lobby_view(self) -> None:
engine = ScoponeEngine() engine = ScoponeEngine()
session = _session(engine, target_score=16) session = _session(engine, target_score=16)
+83 -11
View File
@@ -1,15 +1,19 @@
# #
# This file is autogenerated by pip-compile with Python 3.14 # This file is autogenerated by pip-compile with Python 3.14
# by the following command: # by the following command (run from server/packages/tavolo-app/):
# #
# pip-compile --allow-unsafe --extra-index-url=https://pypi.org/simple --index-url=https://gitea.woggioni.net/api/packages/woggioni/pypi/simple --output-file=requirements.txt pyproject.toml # pip-compile --allow-unsafe --extra otel --extra-index-url=https://pypi.org/simple --index-url=https://gitea.woggioni.net/api/packages/woggioni/pypi/simple --output-file=../../requirements.txt pyproject.toml
# #
# NOTE: the tavolo-platform, tavolo-scopone and tavolo-app sibling packages # NOTE: the tavolo-platform, tavolo-scopone and tavolo-app sibling packages
# are NOT pinned here: they are installed from ./packages in the Dockerfile # are NOT pinned here: they are installed from ./packages in the Dockerfile
# and in development (pip install -e ./packages/tavolo-platform # and in development (pip install -e ./packages/tavolo-platform
# -e ./packages/tavolo-scopone -e ./packages/tavolo-app). All of their # -e ./packages/tavolo-scopone -e ./packages/tavolo-app). All of their
# third-party dependencies are direct dependencies of tavolo-app, so this # third-party dependencies are direct dependencies of tavolo-app, so this
# lock covers the full closure. # lock covers the full closure. The otel extra (kaya-otel and its
# OpenTelemetry dependencies) IS included, so OTEL_ENABLED works in the
# Docker image; pip-compile cannot resolve the unpublishable sibling
# packages, so regeneration drops them from a copy of the input file (the
# existing pins constrain the rest).
# #
--index-url https://gitea.woggioni.net/api/packages/woggioni/pypi/simple --index-url https://gitea.woggioni.net/api/packages/woggioni/pypi/simple
--extra-index-url https://pypi.org/simple --extra-index-url https://pypi.org/simple
@@ -39,6 +43,8 @@ cryptography==50.0.1
# via pyjwt # via pyjwt
dictdiffer==0.10.0 dictdiffer==0.10.0
# via aerich # via aerich
googleapis-common-protos==1.75.4
# via opentelemetry-exporter-otlp-proto-http
granian==2.8.3 granian==2.8.3
# via # via
# kaya-rsgi # kaya-rsgi
@@ -57,29 +63,85 @@ idna==3.20
# httpx # httpx
iso8601==2.1.0 iso8601==2.1.0
# via tortoise-orm # via tortoise-orm
kaya-core==0.0.4 kaya-core==0.0.7
# via # via
# kaya-cors # kaya-cors
# kaya-oidc # kaya-oidc
# kaya-openapi # kaya-openapi
# kaya-otel
# kaya-rsgi # kaya-rsgi
# kaya-session # kaya-session
# tavolo-app (pyproject.toml) # tavolo-app (pyproject.toml)
kaya-cors==0.0.4 kaya-cors==0.0.7
# via tavolo-app (pyproject.toml) # via tavolo-app (pyproject.toml)
kaya-oidc==0.0.4 kaya-oidc==0.0.7
# via tavolo-app (pyproject.toml) # via tavolo-app (pyproject.toml)
kaya-openapi==0.0.4 kaya-openapi==0.0.7
# via tavolo-app (pyproject.toml) # via tavolo-app (pyproject.toml)
kaya-rsgi==0.0.4 kaya-otel==0.0.7
# via tavolo-app (pyproject.toml) # via tavolo-app (pyproject.toml)
kaya-session==0.0.4 kaya-rsgi==0.0.7
# via tavolo-app (pyproject.toml)
kaya-session==0.0.7
# via # via
# kaya-oidc # kaya-oidc
# kaya-session-redis # kaya-session-redis
# tavolo-app (pyproject.toml) # tavolo-app (pyproject.toml)
kaya-session-redis==0.0.4 kaya-session-redis==0.0.7
# via tavolo-app (pyproject.toml) # via tavolo-app (pyproject.toml)
opentelemetry-api==1.45.0
# via
# opentelemetry-exporter-http-transport
# opentelemetry-exporter-otlp-proto-http
# opentelemetry-instrumentation
# opentelemetry-instrumentation-asyncpg
# opentelemetry-instrumentation-redis
# opentelemetry-instrumentation-system-metrics
# opentelemetry-sdk
# opentelemetry-semantic-conventions
opentelemetry-exporter-http-transport[urllib3]==0.66b0
# via opentelemetry-exporter-otlp-proto-http
opentelemetry-exporter-otlp-common==0.66b0
# via opentelemetry-exporter-otlp-proto-http
opentelemetry-exporter-otlp-proto-common==1.45.0
# via opentelemetry-exporter-otlp-proto-http
opentelemetry-exporter-otlp-proto-http==1.45.0
# via kaya-otel
opentelemetry-instrumentation==0.66b0
# via
# opentelemetry-instrumentation-asyncpg
# opentelemetry-instrumentation-redis
# opentelemetry-instrumentation-system-metrics
opentelemetry-instrumentation-asyncpg==0.66b0
# via tavolo-app (pyproject.toml)
opentelemetry-instrumentation-redis==0.66b0
# via tavolo-app (pyproject.toml)
opentelemetry-instrumentation-system-metrics==0.66b0
# via tavolo-app (pyproject.toml)
opentelemetry-proto==1.45.0
# via
# opentelemetry-exporter-otlp-proto-common
# opentelemetry-exporter-otlp-proto-http
opentelemetry-sdk==1.45.0
# via
# kaya-otel
# opentelemetry-exporter-otlp-common
# opentelemetry-exporter-otlp-proto-http
opentelemetry-semantic-conventions==0.66b0
# via
# opentelemetry-instrumentation
# opentelemetry-instrumentation-asyncpg
# opentelemetry-instrumentation-redis
# opentelemetry-instrumentation-system-metrics
# opentelemetry-sdk
packaging==26.3
# via opentelemetry-instrumentation
protobuf==7.36.2
# via
# googleapis-common-protos
# opentelemetry-proto
psutil==7.2.2
# via opentelemetry-instrumentation-system-metrics
pwo==0.1.2 pwo==0.1.2
# via # via
# kaya-core # kaya-core
@@ -88,7 +150,7 @@ pwo==0.1.2
# tavolo-app (pyproject.toml) # tavolo-app (pyproject.toml)
pycparser==3.0 pycparser==3.0
# via cffi # via cffi
pyjwt[crypto]==2.14.0 pyjwt[crypto]==2.15.1
# via # via
# kaya-oidc # kaya-oidc
# tavolo-app (pyproject.toml) # tavolo-app (pyproject.toml)
@@ -106,4 +168,14 @@ typing-extensions==4.16.0
# via # via
# anyio # anyio
# kaya-core # kaya-core
# opentelemetry-api
# opentelemetry-exporter-otlp-proto-http
# opentelemetry-sdk
# opentelemetry-semantic-conventions
# pwo # pwo
urllib3==2.8.0
# via opentelemetry-exporter-http-transport
wrapt==2.5.0
# via
# opentelemetry-instrumentation
# opentelemetry-instrumentation-redis
+4 -3
View File
@@ -10,8 +10,9 @@ from __future__ import annotations
import os import os
os.environ.setdefault("DATABASE_URL", "sqlite://:memory:") os.environ.setdefault("DATABASE_URL", "sqlite://:memory:")
os.environ.setdefault("OIDC_ISSUER", "http://localhost:8180/tavolo") os.environ.setdefault("OIDC_0_PROVIDER", "default")
os.environ.setdefault("OIDC_CLIENT_ID", "tavolo") os.environ.setdefault("OIDC_0_ISSUER", "http://localhost:8180/tavolo")
os.environ.setdefault("OIDC_REDIRECT_URI", "http://localhost:8080/auth/callback") os.environ.setdefault("OIDC_0_CLIENT_ID", "tavolo")
os.environ.setdefault("OIDC_0_REDIRECT_URI", "http://localhost:8080/auth/default/callback")
# Unset REDIS_URL: sessions and live games use the in-memory fallbacks. # Unset REDIS_URL: sessions and live games use the in-memory fallbacks.
os.environ.pop("REDIS_URL", None) os.environ.pop("REDIS_URL", None)
+11 -6
View File
@@ -1,8 +1,9 @@
"""Helpers for faking the OIDC authenticated user during tests. """Helpers for faking the OIDC authenticated user during tests.
HTTP handlers funnel through ``oidc_mixin.get_user``; websocket handlers HTTP handlers funnel through ``OIDCProviders.get_users`` (wrapped by
through :func:`tavolo.platform.auth.get_ws_user`. Patching those two entry ``tavolo.platform.auth.current_user``); websocket handlers through
points lets route and websocket tests run entirely in-process with no IdP. :func:`tavolo.platform.auth.get_ws_user`. Patching those two entry points
lets route and websocket tests run entirely in-process with no IdP.
""" """
from __future__ import annotations from __future__ import annotations
@@ -14,9 +15,13 @@ from kaya.oidc import OIDCUser
# Import the app first: it pulls in the route modules, which import # Import the app first: it pulls in the route modules, which import
# ``tavolo.platform.auth`` themselves. # ``tavolo.platform.auth`` themselves.
from tavolo.app import oidc_mixin from tavolo.app import oidc
from tavolo.platform import auth from tavolo.platform import auth
# The app's first configured provider (the test environment configures one
# named "default").
_PROVIDER = next(iter(oidc.providers))
def make_user(sub: str, name: Optional[str] = None) -> OIDCUser: def make_user(sub: str, name: Optional[str] = None) -> OIDCUser:
return OIDCUser({"sub": sub, "preferred_username": name or sub}) return OIDCUser({"sub": sub, "preferred_username": name or sub})
@@ -24,9 +29,9 @@ def make_user(sub: str, name: Optional[str] = None) -> OIDCUser:
@contextlib.contextmanager @contextlib.contextmanager
def oidc_user(sub: str, name: Optional[str] = None) -> Iterator[OIDCUser]: def oidc_user(sub: str, name: Optional[str] = None) -> Iterator[OIDCUser]:
"""Context manager: patch ``oidc_mixin.get_user`` to return this user.""" """Context manager: make ``oidc.get_users`` return this user."""
user = make_user(sub, name) user = make_user(sub, name)
patcher = _mock.patch.object(oidc_mixin, "get_user", return_value=user) patcher = _mock.patch.object(oidc, "get_users", return_value={_PROVIDER: user})
patcher.start() patcher.start()
try: try:
yield user yield user
+148
View File
@@ -164,6 +164,154 @@ class OTelSettingsTests(unittest.TestCase):
settings = _settings({"OTEL_EXCLUDED_PATHS": "/api/health, /metrics"}) settings = _settings({"OTEL_EXCLUDED_PATHS": "/api/health, /metrics"})
self.assertEqual(("/api/health", "/metrics"), settings.otel_excluded_paths) self.assertEqual(("/api/health", "/metrics"), settings.otel_excluded_paths)
def test_otel_interval_defaults(self):
settings = _settings({})
self.assertEqual(60000, settings.otel_metric_export_interval_millis)
self.assertEqual(15, settings.otel_metrics_poll_interval_seconds)
def test_otel_intervals_parse(self):
settings = _settings({
"OTEL_METRIC_EXPORT_INTERVAL_MILLIS": "5000",
"OTEL_METRICS_POLL_INTERVAL_SECONDS": "3",
})
self.assertEqual(5000, settings.otel_metric_export_interval_millis)
self.assertEqual(3, settings.otel_metrics_poll_interval_seconds)
class OidcSettingsTests(unittest.TestCase):
def test_defaults_to_the_dev_mock_provider(self):
settings = _settings({})
self.assertEqual(1, len(settings.oidc_providers))
provider = settings.oidc_providers[0]
self.assertEqual("default", provider.provider)
self.assertEqual("http://localhost:8180/tavolo", provider.issuer)
self.assertEqual("tavolo", provider.client_id)
self.assertIsNone(provider.client_secret)
self.assertEqual(
"http://localhost:8080/auth/default/callback", provider.redirect_uri
)
self.assertEqual("/", provider.post_login_redirect)
self.assertEqual("/", provider.post_logout_redirect)
def test_single_indexed_provider_defaults(self):
settings = _settings({
"OIDC_0_PROVIDER": "google",
"OIDC_0_ISSUER": "https://accounts.google.com",
"OIDC_0_CLIENT_ID": "client-id",
"OIDC_0_CLIENT_SECRET": "client-secret",
})
self.assertEqual(1, len(settings.oidc_providers))
provider = settings.oidc_providers[0]
self.assertEqual("google", provider.provider)
self.assertEqual("https://accounts.google.com", provider.issuer)
self.assertEqual("client-id", provider.client_id)
self.assertEqual("client-secret", provider.client_secret)
# The callback default mirrors kaya's /auth/<provider>/callback.
self.assertEqual(
"http://localhost:8080/auth/google/callback", provider.redirect_uri
)
self.assertEqual("/", provider.post_login_redirect)
self.assertEqual("/", provider.post_logout_redirect)
def test_multiple_providers_are_ordered_by_index(self):
settings = _settings({
"OIDC_0_PROVIDER": "google",
"OIDC_0_ISSUER": "https://accounts.google.com",
"OIDC_0_CLIENT_ID": "g",
"OIDC_1_PROVIDER": "github",
"OIDC_1_ISSUER": "https://github.com/login/oauth",
"OIDC_1_CLIENT_ID": "h",
"OIDC_1_REDIRECT_URI": "https://tavolo.example/auth/github/callback",
"OIDC_1_POST_LOGIN_REDIRECT": "/lobby",
"OIDC_1_POST_LOGOUT_REDIRECT": "/bye",
})
self.assertEqual(
["google", "github"], [p.provider for p in settings.oidc_providers]
)
github = settings.oidc_providers[1]
self.assertEqual(
"https://tavolo.example/auth/github/callback", github.redirect_uri
)
self.assertEqual("/lobby", github.post_login_redirect)
self.assertEqual("/bye", github.post_logout_redirect)
def test_empty_optional_values_fall_back_to_defaults(self):
settings = _settings({
"OIDC_0_PROVIDER": "google",
"OIDC_0_ISSUER": "https://accounts.google.com",
"OIDC_0_CLIENT_ID": "g",
"OIDC_0_CLIENT_SECRET": "",
"OIDC_0_REDIRECT_URI": "",
})
provider = settings.oidc_providers[0]
self.assertIsNone(provider.client_secret)
self.assertEqual(
"http://localhost:8080/auth/google/callback", provider.redirect_uri
)
def test_missing_provider_name(self):
with self.assertRaisesRegex(RuntimeError, "OIDC_0_PROVIDER"):
_settings({"OIDC_0_ISSUER": "https://idp", "OIDC_0_CLIENT_ID": "c"})
def test_missing_issuer(self):
with self.assertRaisesRegex(RuntimeError, "OIDC_0_ISSUER"):
_settings({"OIDC_0_PROVIDER": "google", "OIDC_0_CLIENT_ID": "c"})
def test_missing_client_id(self):
with self.assertRaisesRegex(RuntimeError, "OIDC_0_CLIENT_ID"):
_settings({"OIDC_0_PROVIDER": "google", "OIDC_0_ISSUER": "https://idp"})
def test_indices_must_start_at_zero(self):
with self.assertRaisesRegex(RuntimeError, "contiguous"):
_settings({
"OIDC_3_PROVIDER": "google",
"OIDC_3_ISSUER": "https://idp",
"OIDC_3_CLIENT_ID": "c",
})
def test_indices_must_be_contiguous(self):
with self.assertRaisesRegex(RuntimeError, "OIDC_1_"):
_settings({
"OIDC_0_PROVIDER": "google",
"OIDC_0_ISSUER": "https://idp",
"OIDC_0_CLIENT_ID": "c",
"OIDC_2_PROVIDER": "github",
"OIDC_2_ISSUER": "https://idp2",
"OIDC_2_CLIENT_ID": "c2",
})
def test_duplicate_provider_names_are_rejected(self):
with self.assertRaisesRegex(RuntimeError, "duplicate"):
_settings({
"OIDC_0_PROVIDER": "google",
"OIDC_0_ISSUER": "https://idp",
"OIDC_0_CLIENT_ID": "c",
"OIDC_1_PROVIDER": "google",
"OIDC_1_ISSUER": "https://idp2",
"OIDC_1_CLIENT_ID": "c2",
})
def test_invalid_provider_name_is_rejected(self):
with self.assertRaisesRegex(RuntimeError, "OIDC_0_PROVIDER"):
_settings({
"OIDC_0_PROVIDER": "not a name",
"OIDC_0_ISSUER": "https://idp",
"OIDC_0_CLIENT_ID": "c",
})
def test_unknown_indexed_variable_is_rejected(self):
with self.assertRaisesRegex(RuntimeError, "OIDC_0_ISUER"):
_settings({
"OIDC_0_PROVIDER": "google",
"OIDC_0_ISSUER": "https://idp",
"OIDC_0_CLIENT_ID": "c",
"OIDC_0_ISUER": "typo",
})
def test_legacy_unindexed_variables_get_a_migration_hint(self):
with self.assertRaisesRegex(RuntimeError, "OIDC_ISSUER -> OIDC_0_ISSUER"):
_settings({"OIDC_ISSUER": "https://idp", "OIDC_CLIENT_ID": "c"})
if __name__ == "__main__": if __name__ == "__main__":
unittest.main() unittest.main()
+9
View File
@@ -18,12 +18,15 @@ class LoggingConfigTest(unittest.TestCase):
self._root_level = root.level self._root_level = root.level
tavolo = logging.getLogger("tavolo") tavolo = logging.getLogger("tavolo")
self._tavolo_level = tavolo.level self._tavolo_level = tavolo.level
pool = logging.getLogger("urllib3.connectionpool")
self._pool_level = pool.level
def tearDown(self) -> None: def tearDown(self) -> None:
root = logging.getLogger() root = logging.getLogger()
root.handlers = self._root_handlers root.handlers = self._root_handlers
root.level = self._root_level root.level = self._root_level
logging.getLogger("tavolo").level = self._tavolo_level logging.getLogger("tavolo").level = self._tavolo_level
logging.getLogger("urllib3.connectionpool").level = self._pool_level
def test_default_config_when_unset(self) -> None: def test_default_config_when_unset(self) -> None:
configure_logging(None) configure_logging(None)
@@ -33,6 +36,12 @@ class LoggingConfigTest(unittest.TestCase):
any(isinstance(h, logging.StreamHandler) for h in root.handlers), any(isinstance(h, logging.StreamHandler) for h in root.handlers),
"default config installs a console stream handler", "default config installs a console stream handler",
) )
# The noisy per-request HTTP logger is capped at INFO while the
# root stays at DEBUG.
self.assertEqual(
logging.INFO,
logging.getLogger("urllib3.connectionpool").getEffectiveLevel(),
)
def test_yaml_config_is_applied(self) -> None: def test_yaml_config_is_applied(self) -> None:
with tempfile.TemporaryDirectory() as tmp: with tempfile.TemporaryDirectory() as tmp:
+28
View File
@@ -16,6 +16,7 @@ from unittest.mock import patch
from tavolo.app import otel_mixin_from_settings from tavolo.app import otel_mixin_from_settings
from tavolo.config import Settings from tavolo.config import Settings
from tavolo.telemetry import session_tag_hook
def _settings(env: dict) -> Settings: def _settings(env: dict) -> Settings:
@@ -58,8 +59,17 @@ class OTelMixinFromSettingsTests(unittest.TestCase):
"endpoint": "http://collector:4318", "endpoint": "http://collector:4318",
"headers": {"Authorization": "Bearer t", "X-Tenant": "one"}, "headers": {"Authorization": "Bearer t", "X-Tenant": "one"},
"excluded_paths": ("/api/health",), "excluded_paths": ("/api/health",),
"server_response_hook": session_tag_hook,
"websocket_close_hook": session_tag_hook,
}, mixin.kwargs) }, mixin.kwargs)
def test_session_tag_hooks_are_wired(self):
with _stub_kaya_otel():
mixin = otel_mixin_from_settings(_settings({"OTEL_ENABLED": "1"}))
assert isinstance(mixin, _StubOTelMixin)
self.assertIs(session_tag_hook, mixin.kwargs["server_response_hook"])
self.assertIs(session_tag_hook, mixin.kwargs["websocket_close_hook"])
def test_defaults_when_only_enabled(self): def test_defaults_when_only_enabled(self):
with _stub_kaya_otel(): with _stub_kaya_otel():
mixin = otel_mixin_from_settings(_settings({"OTEL_ENABLED": "on"})) mixin = otel_mixin_from_settings(_settings({"OTEL_ENABLED": "on"}))
@@ -69,6 +79,24 @@ class OTelMixinFromSettingsTests(unittest.TestCase):
self.assertIsNone(mixin.kwargs["headers"]) self.assertIsNone(mixin.kwargs["headers"])
self.assertEqual(("/api/health",), mixin.kwargs["excluded_paths"]) self.assertEqual(("/api/health",), mixin.kwargs["excluded_paths"])
def test_meter_provider_is_passed_through(self):
with _stub_kaya_otel():
provider = object()
mixin = otel_mixin_from_settings(
_settings({"OTEL_ENABLED": "1"}), meter_provider=provider
)
assert isinstance(mixin, _StubOTelMixin)
self.assertIs(provider, mixin.kwargs["meter_provider"])
def test_tracer_provider_is_passed_through(self):
with _stub_kaya_otel():
provider = object()
mixin = otel_mixin_from_settings(
_settings({"OTEL_ENABLED": "1"}), tracer_provider=provider
)
assert isinstance(mixin, _StubOTelMixin)
self.assertIs(provider, mixin.kwargs["tracer_provider"])
def test_excluded_paths_are_passed_through(self): def test_excluded_paths_are_passed_through(self):
with _stub_kaya_otel(): with _stub_kaya_otel():
mixin = otel_mixin_from_settings(_settings({ mixin = otel_mixin_from_settings(_settings({
+35
View File
@@ -0,0 +1,35 @@
"""Auth routes on the fully wired application.
The integration suite's app is built from the indexed test environment
(one provider named ``default``); these tests pin the public surface the
SPA and bookmarks rely on.
"""
from __future__ import annotations
import unittest
from httpx import ASGITransport, AsyncClient
from tavolo.app import app
from tests.helpers import async_test
class AuthRoutesTest(unittest.TestCase):
@async_test
async def test_providers_endpoint_reports_the_configured_provider(self) -> None:
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://127.0.0.1") as client:
response = await client.get("/api/auth/providers")
self.assertEqual(200, response.status_code)
self.assertEqual(
{"providers": [{"name": "default", "login_url": "/auth/default/login"}]},
response.json(),
)
@async_test
async def test_generic_login_redirects_to_the_provider(self) -> None:
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://127.0.0.1") as client:
response = await client.get("/auth/login", follow_redirects=False)
self.assertEqual(302, response.status_code)
self.assertEqual("/auth/default/login", response.headers["location"])
+28 -2
View File
@@ -39,9 +39,19 @@ class GamesRouteTest(unittest.TestCase):
self.assertEqual(200, snapshot.status_code) self.assertEqual(200, snapshot.status_code)
self.assertEqual("alice", snapshot.json()["players"][0]["sub"]) self.assertEqual("alice", snapshot.json()["players"][0]["sub"])
# A logged-in non-player is a read-only spectator: no join
# code, and the view is marked as a spectator view.
with oidc_user("mallory"): with oidc_user("mallory"):
forbidden = await client.get(f"/api/games/{game_id}") watched = await client.get(f"/api/games/{game_id}")
self.assertEqual(403, forbidden.status_code) self.assertEqual(200, watched.status_code)
self.assertTrue(watched.json()["spectator"])
self.assertNotIn("join_code", watched.json())
# Anonymous users may watch too.
anonymous = await client.get(f"/api/games/{game_id}")
self.assertEqual(200, anonymous.status_code)
self.assertTrue(anonymous.json()["spectator"])
self.assertNotIn("join_code", anonymous.json())
@async_test @async_test
async def test_join_fills_seats_and_starts_game(self) -> None: async def test_join_fills_seats_and_starts_game(self) -> None:
@@ -61,6 +71,7 @@ class GamesRouteTest(unittest.TestCase):
started = await client.post("/api/games/join", json={"code": code}) started = await client.post("/api/games/join", json={"code": code})
self.assertEqual(200, started.status_code) self.assertEqual(200, started.status_code)
state = started.json() state = started.json()
game_id = state["id"]
self.assertEqual("playing", state["phase"]) self.assertEqual("playing", state["phase"])
self.assertEqual(4, len(state["players"])) self.assertEqual(4, len(state["players"]))
self.assertEqual([], state["table"]) self.assertEqual([], state["table"])
@@ -72,6 +83,21 @@ class GamesRouteTest(unittest.TestCase):
self.assertNotIn("hand", state["players"][0]) self.assertNotIn("hand", state["players"][0])
self.assertEqual(1, state["turn"]) self.assertEqual(1, state["turn"])
# A spectator sees the table, the score and each player's
# captured count and scope — but no hand at all.
watched = await client.get(f"/api/games/{game_id}")
self.assertEqual(200, watched.status_code)
spect = watched.json()
self.assertTrue(spect["spectator"])
self.assertNotIn("join_code", spect)
self.assertEqual([], spect["table"])
self.assertEqual({"A": 0, "B": 0}, spect["scores"])
for participant in spect["players"]:
self.assertNotIn("hand", participant)
self.assertEqual(10, participant["cards_left"])
self.assertEqual(0, participant["captured_count"])
self.assertEqual(0, participant["scope"])
@async_test @async_test
async def test_create_napola_option(self) -> None: async def test_create_napola_option(self) -> None:
transport = ASGITransport(app=app) transport = ASGITransport(app=app)
+427
View File
@@ -0,0 +1,427 @@
"""Telemetry unit tests: app gauges, process metrics, log export and counts.
The OpenTelemetry packages are an optional extra (``tavolo-app[otel]``),
so the provider-dependent tests skip when they are not installed.
"""
from __future__ import annotations
import hashlib
import logging
import os
import sys
import unittest
from types import SimpleNamespace
from typing import Optional
from unittest.mock import patch
from kaya.session import Session
from tavolo.telemetry import (
CountingInMemorySessionStore,
CountingRedisSessionStore,
TelemetryMixin,
_NoOtelLogs,
session_tag_hook,
telemetry_from_settings,
)
from tavolo.config import Settings
from tests.helpers import async_test
try:
from opentelemetry.sdk._logs import LoggingHandler, LoggerProvider
from opentelemetry.sdk._logs.export import (
InMemoryLogRecordExporter,
SimpleLogRecordProcessor,
)
from opentelemetry.sdk.metrics import MeterProvider
from opentelemetry.sdk.metrics.export import InMemoryMetricReader
from opentelemetry.sdk.resources import Resource
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import SimpleSpanProcessor
from opentelemetry.sdk.trace.export.in_memory_span_exporter import (
InMemorySpanExporter,
)
HAVE_OTEL = True
except ImportError: # pragma: no cover - installs without the otel extra
HAVE_OTEL = False
def _settings(env: dict) -> Settings:
with patch.dict(os.environ, env, clear=True):
return Settings.from_env()
class _FakeSessions:
def __init__(self, count: int = 0) -> None:
self.value: Optional[int] = count
self.error: Optional[Exception] = None
async def count(self) -> int:
if self.error is not None:
raise self.error
assert self.value is not None
return self.value
class _FakeGames:
def __init__(self, count: int = 0) -> None:
self.value: Optional[int] = count
async def count_ongoing(self) -> int:
assert self.value is not None
return self.value
class _FakeRedis:
"""Minimal async Redis double covering the session-count scan."""
def __init__(self, keys: list) -> None:
self.keys = keys
async def scan_iter(self, match: str = "", count: int = 0):
prefix = match.rstrip("*")
for key in self.keys:
if key.startswith(prefix):
yield key
class TelemetrySetupTest(unittest.TestCase):
def test_disabled_returns_none(self) -> None:
self.assertIsNone(
telemetry_from_settings(
_settings({}), _FakeSessions(), _FakeGames()
)
)
def test_missing_otel_dependencies_raise(self) -> None:
settings = _settings({"OTEL_ENABLED": "true"})
missing = "opentelemetry.exporter.otlp.proto.http._log_exporter"
with patch.dict(sys.modules, {missing: None}):
with self.assertRaises(RuntimeError):
telemetry_from_settings(settings, _FakeSessions(), _FakeGames())
class SessionCountTest(unittest.TestCase):
@async_test
async def test_in_memory_counts_live_sessions(self) -> None:
now = [1000.0]
store = CountingInMemorySessionStore(clock=lambda: now[0])
await store.save("a", Session("a", {"user": "alice"}), max_age=10)
await store.save("b", Session("b", {"user": "bob"}), max_age=10)
await store.save("c", Session("c", {"user": "carol"})) # no expiry
self.assertEqual(3, await store.count())
now[0] += 11 # a and b have expired
self.assertEqual(1, await store.count())
await store.delete("c")
self.assertEqual(0, await store.count())
@async_test
async def test_redis_counts_only_session_keys(self) -> None:
store = CountingRedisSessionStore(
_FakeRedis(["kaya:session:a", "kaya:session:b", "tavolo:code:X"])
)
self.assertEqual(2, await store.count())
class NoOtelLogsFilterTest(unittest.TestCase):
def test_filters_opentelemetry_records_only(self) -> None:
filter_ = _NoOtelLogs()
self.assertFalse(
filter_.filter(logging.makeLogRecord({"name": "opentelemetry.sdk"}))
)
self.assertTrue(
filter_.filter(logging.makeLogRecord({"name": "tavolo.app"}))
)
self.assertTrue(
filter_.filter(logging.makeLogRecord({"name": "granian.server"}))
)
class _RecordedSpan:
def __init__(self) -> None:
self.attributes: dict = {}
def set_attribute(self, key: str, value) -> None:
self.attributes[key] = value
class SessionTagHookTest(unittest.TestCase):
def test_tags_authenticated_session(self) -> None:
span = _RecordedSpan()
carrier = SimpleNamespace(
session=Session("sid", {"oidc:default:user": {"sub": "alice"}})
)
session_tag_hook(span, carrier)
self.assertEqual(
hashlib.sha256(b"sid").hexdigest()[:16],
span.attributes["session.id"],
)
self.assertEqual("alice", span.attributes["user.id"])
def test_tags_user_of_any_provider(self) -> None:
span = _RecordedSpan()
carrier = SimpleNamespace(
session=Session("sid", {"oidc:google:user": {"sub": "alice"}})
)
session_tag_hook(span, carrier)
self.assertEqual("alice", span.attributes["user.id"])
def test_ignores_legacy_unprefixed_user_key(self) -> None:
span = _RecordedSpan()
carrier = SimpleNamespace(
session=Session("sid", {"oidc_user": {"sub": "alice"}})
)
session_tag_hook(span, carrier)
self.assertIn("session.id", span.attributes)
self.assertNotIn("user.id", span.attributes)
def test_tags_session_without_user(self) -> None:
span = _RecordedSpan()
session_tag_hook(span, SimpleNamespace(session=Session("sid", {})))
self.assertIn("session.id", span.attributes)
self.assertNotIn("user.id", span.attributes)
def test_tags_user_when_session_id_not_yet_minted(self) -> None:
span = _RecordedSpan()
carrier = SimpleNamespace(
session=Session(None, {"oidc:default:user": {"sub": "alice"}})
)
session_tag_hook(span, carrier)
self.assertNotIn("session.id", span.attributes)
self.assertEqual("alice", span.attributes["user.id"])
def test_missing_session_is_a_noop(self) -> None:
span = _RecordedSpan()
session_tag_hook(span, SimpleNamespace())
self.assertEqual({}, span.attributes)
def test_malformed_user_is_ignored(self) -> None:
span = _RecordedSpan()
carrier = SimpleNamespace(session=Session("sid", {"oidc:default:user": "alice"}))
session_tag_hook(span, carrier)
self.assertIn("session.id", span.attributes)
self.assertNotIn("user.id", span.attributes)
@unittest.skipUnless(HAVE_OTEL, "tavolo-app[otel] not installed")
class SessionTagIntegrationTest(unittest.TestCase):
"""The hook is wired as kaya-otel's response hook, so it sees the
session after the session mixin persisted and minted it."""
@async_test
async def test_spans_carry_hashed_session_and_oidc_sub(self) -> None:
from httpx import ASGITransport, AsyncClient
from kaya.core import HttpContext, KayaApp
from kaya.otel import OTelMixin
from kaya.session import InMemorySessionStore, SessionMixin
span_exporter = InMemorySpanExporter()
tracer_provider = TracerProvider(
resource=Resource.create({"service.name": "test"})
)
tracer_provider.add_span_processor(SimpleSpanProcessor(span_exporter))
app = KayaApp(
mixins=[
OTelMixin(
service_name="test",
tracer_provider=tracer_provider,
meter_provider=MeterProvider(
metric_readers=[InMemoryMetricReader()]
),
server_response_hook=session_tag_hook,
websocket_close_hook=session_tag_hook,
),
SessionMixin(InMemorySessionStore()),
]
)
@app.GET("/login")
async def login(ctx: HttpContext) -> None:
session = ctx.session
assert session is not None
session["oidc:default:user"] = {"sub": "alice"}
await ctx.send_str(200, "ok")
@app.GET("/me")
async def me(ctx: HttpContext) -> None:
await ctx.send_str(200, "ok")
transport = ASGITransport(app=app) # type: ignore[arg-type]
async with AsyncClient(
transport=transport, base_url="http://127.0.0.1"
) as client:
first = await client.get("/login")
second = await client.get("/me")
self.assertEqual(200, first.status_code)
self.assertEqual(200, second.status_code)
# The cookie value is the session id; only its digest is exported.
cookie = first.cookies["session_id"]
digest = hashlib.sha256(cookie.encode()).hexdigest()[:16]
spans = {span.name: span for span in span_exporter.get_finished_spans()}
self.assertEqual({"GET /login", "GET /me"}, set(spans))
for name, span in spans.items():
attributes = span.attributes
assert attributes is not None
self.assertEqual(digest, attributes["session.id"], name)
self.assertEqual("alice", attributes["user.id"], name)
def _metric_points(reader) -> dict:
"""Map metric name -> list of (attributes, value) data points."""
data = reader.get_metrics_data()
assert data is not None
points = {}
for resource_metrics in data.resource_metrics:
for scope_metrics in resource_metrics.scope_metrics:
for metric in scope_metrics.metrics:
points[metric.name] = [
(dict(point.attributes), point.value)
for point in metric.data.data_points
]
return points
@unittest.skipUnless(HAVE_OTEL, "tavolo-app[otel] not installed")
class MetricExportTest(unittest.TestCase):
# The instrumentation packages are process-wide singletons whose
# callbacks cannot be re-registered, so the providers and mixin are
# built once for the whole class (as the application does).
sessions = _FakeSessions(0)
games = _FakeGames(0)
@classmethod
def setUpClass(cls) -> None:
resource = Resource.create({"service.name": "test"})
cls.reader = InMemoryMetricReader()
cls.provider = MeterProvider(resource=resource, metric_readers=[cls.reader])
cls.span_exporter = InMemorySpanExporter()
cls.tracer_provider = TracerProvider(resource=resource)
cls.tracer_provider.add_span_processor(
SimpleSpanProcessor(cls.span_exporter)
)
cls.mixin = TelemetryMixin(
meter_provider=cls.provider,
tracer_provider=cls.tracer_provider,
logger_provider=None,
log_handler=None,
session_store=cls.sessions,
game_store=cls.games,
poll_interval_seconds=1,
)
cls.mixin.apply(None) # type: ignore[arg-type]
@classmethod
def tearDownClass(cls) -> None:
cls.mixin.shutdown(None) # type: ignore[arg-type]
@async_test
async def test_app_gauges_report_polled_counts(self) -> None:
self.sessions.value = 3
self.sessions.error = None
self.games.value = 2
await self.mixin.refresh_counts()
points = _metric_points(self.reader)
self.assertEqual([({}, 3)], points["tavolo.active_users"])
self.assertEqual([({}, 2)], points["tavolo.ongoing_games"])
@async_test
async def test_failed_refresh_keeps_last_value(self) -> None:
self.sessions.value = 5
self.sessions.error = None
await self.mixin.refresh_counts()
self.sessions.error = RuntimeError("redis down")
await self.mixin.refresh_counts()
points = _metric_points(self.reader)
self.assertEqual([({}, 5)], points["tavolo.active_users"])
@async_test
async def test_process_and_gc_metrics_are_exported(self) -> None:
await self.mixin.refresh_counts()
points = _metric_points(self.reader)
for name in (
"process.cpu.time",
"process.cpu.utilization",
"process.memory.usage",
"process.memory.virtual",
"cpython.gc.collections",
"cpython.gc.collected_objects",
"cpython.gc.uncollectable_objects",
):
self.assertIn(name, points)
self.assertGreater(points["process.memory.usage"][0][1], 0)
modes = {attrs["type"] for attrs, _ in points["process.cpu.time"]}
self.assertEqual({"user", "system"}, modes)
generations = {
attrs["cpython.gc.generation"]
for attrs, _ in points["cpython.gc.collections"]
}
self.assertEqual({0, 1, 2}, generations)
# Host-level metrics are deliberately not configured.
self.assertNotIn("system.cpu.time", points)
@async_test
async def test_database_clients_are_instrumented(self) -> None:
import asyncpg.connection
import redis.asyncio
self.assertTrue(hasattr(redis.asyncio.Redis.execute_command, "__wrapped__"))
self.assertTrue(hasattr(asyncpg.connection.Connection.fetch, "__wrapped__"))
# Exercise each wrapper through stubs: the downstream call fails (no
# server), but the span is recorded before it does.
client = object.__new__(redis.asyncio.Redis)
client.connection_pool = SimpleNamespace(
connection_kwargs={"host": "127.0.0.1", "port": 6379, "db": 0}
)
with self.assertRaises(AttributeError):
await client.execute_command("GET", "key")
connection = object.__new__(asyncpg.connection.Connection)
connection._aborted = True # keep __del__ quiet on the stub
with self.assertRaises(Exception):
await connection.fetch("SELECT 1")
spans = {
span.name: span for span in self.span_exporter.get_finished_spans()
}
self.assertEqual("GET ?", spans["GET"].attributes["db.statement"])
self.assertEqual("redis", spans["GET"].attributes["db.system"])
self.assertEqual("SELECT 1", spans["SELECT"].attributes["db.statement"])
self.assertEqual("postgresql", spans["SELECT"].attributes["db.system"])
@unittest.skipUnless(HAVE_OTEL, "tavolo-app[otel] not installed")
class LogExportTest(unittest.TestCase):
def test_logs_are_exported_and_otel_records_filtered(self) -> None:
exporter = InMemoryLogRecordExporter()
resource = Resource.create({"service.name": "test"})
provider = LoggerProvider(resource=resource)
provider.add_log_record_processor(SimpleLogRecordProcessor(exporter))
handler = LoggingHandler(level=logging.NOTSET, logger_provider=provider)
handler.addFilter(_NoOtelLogs())
root = logging.getLogger()
previous_level = root.level
root.addHandler(handler)
root.setLevel(logging.DEBUG)
try:
logging.getLogger("tavolo.test").info("exported message")
logging.getLogger("opentelemetry.test").info("filtered message")
bodies = [r.log_record.body for r in exporter.get_finished_logs()]
self.assertIn("exported message", bodies)
self.assertNotIn("filtered message", bodies)
records = exporter.get_finished_logs()
self.assertEqual(
"test", records[0].resource.attributes["service.name"]
)
finally:
root.removeHandler(handler)
handler.close()
root.setLevel(previous_level)
provider.shutdown()
if __name__ == "__main__":
unittest.main()
+36 -14
View File
@@ -131,10 +131,10 @@ class WebSocketTest(unittest.TestCase):
with self.assertRaises(WebSocketDisconnect) as caught: with self.assertRaises(WebSocketDisconnect) as caught:
async with aconnect_ws("/ws/games/no-such-game", ws_client): async with aconnect_ws("/ws/games/no-such-game", ws_client):
pass pass
self.assertEqual(4404, caught.exception.code) self.assertEqual(404, caught.exception.code)
@async_test @async_test
async def test_non_player_is_closed(self) -> None: async def test_non_player_watches_as_spectator(self) -> None:
api_transport = ASGITransport(app=app) api_transport = ASGITransport(app=app)
async with AsyncClient(transport=api_transport, base_url="http://127.0.0.1") as client: async with AsyncClient(transport=api_transport, base_url="http://127.0.0.1") as client:
state = await self._started_game(client) state = await self._started_game(client)
@@ -143,20 +143,42 @@ class WebSocketTest(unittest.TestCase):
ws_transport = ASGIWebSocketTransport(app=app) ws_transport = ASGIWebSocketTransport(app=app)
async with AsyncClient(transport=ws_transport, base_url="http://testserver") as ws_client: async with AsyncClient(transport=ws_transport, base_url="http://testserver") as ws_client:
with ws_users([make_user("mallory")]): with ws_users([make_user("mallory")]):
with self.assertRaises(WebSocketDisconnect) as caught: async with aconnect_ws(f"/ws/games/{game_id}", ws_client) as ws:
async with aconnect_ws(f"/ws/games/{game_id}", ws_client): first = await ws.receive_json()
pass self.assertEqual("state", first["type"])
self.assertEqual(4403, caught.exception.code) self.assertTrue(first["game"]["spectator"])
self.assertNotIn("join_code", first["game"])
self.assertIn("table", first["game"])
self.assertEqual({"A": 0, "B": 0}, first["game"]["scores"])
for player in first["game"]["players"]:
self.assertNotIn("hand", player)
# Spectators may not act.
await ws.send_json({"action": "play", "card": "01D"})
error = await ws.receive_json()
self.assertEqual("error", error["type"])
self.assertEqual("forbidden", error["code"])
@async_test @async_test
async def test_unauthenticated_is_closed(self) -> None: async def test_anonymous_user_watches_as_spectator(self) -> None:
ws_transport = ASGIWebSocketTransport(app=app) api_transport = ASGITransport(app=app)
async with AsyncClient(transport=ws_transport, base_url="http://testserver") as ws_client: async with AsyncClient(transport=api_transport, base_url="http://127.0.0.1") as client:
with ws_users([]): state = await self._started_game(client)
with self.assertRaises(WebSocketDisconnect) as caught: game_id = state["id"]
async with aconnect_ws("/ws/games/whatever", ws_client):
pass ws_transport = ASGIWebSocketTransport(app=app)
self.assertEqual(4401, caught.exception.code) async with AsyncClient(transport=ws_transport, base_url="http://testserver") as ws_client:
with ws_users([]):
async with aconnect_ws(f"/ws/games/{game_id}", ws_client) as ws:
first = await ws.receive_json()
self.assertEqual("state", first["type"])
self.assertTrue(first["game"]["spectator"])
self.assertNotIn("join_code", first["game"])
for player in first["game"]["players"]:
self.assertNotIn("hand", player)
await ws.send_json({"action": "play", "card": "01D"})
error = await ws.receive_json()
self.assertEqual("error", error["type"])
self.assertEqual("forbidden", error["code"])
async def _seed_last_play_state(hand_ack_timeout: int = 30) -> str: async def _seed_last_play_state(hand_ack_timeout: int = 30) -> str:
+38 -9
View File
@@ -22,6 +22,19 @@ pub async fn me() -> Result<Option<User>, String> {
resp.json().await.map(Some).map_err(|e| e.to_string()) resp.json().await.map(Some).map_err(|e| e.to_string())
} }
/// Fetch the configured OIDC login providers (public endpoint).
pub async fn auth_providers() -> Result<Vec<AuthProvider>, String> {
let resp = Request::get("/api/auth/providers")
.send()
.await
.map_err(|e| e.to_string())?;
if !resp.ok() {
return Err(server_error(resp.status()));
}
let page: AuthProvidersPage = resp.json().await.map_err(|e| e.to_string())?;
Ok(page.providers)
}
/// Fetch the card games the platform can host (for the creation dropdown). /// Fetch the card games the platform can host (for the creation dropdown).
pub async fn game_types() -> Result<Vec<GameTypeInfo>, String> { pub async fn game_types() -> Result<Vec<GameTypeInfo>, String> {
let resp = Request::get("/api/game-types") let resp = Request::get("/api/game-types")
@@ -87,16 +100,32 @@ pub async fn join_preview(code: &str) -> Result<Option<JoinPreview>, String> {
resp.json().await.map(Some).map_err(|e| e.to_string()) resp.json().await.map(Some).map_err(|e| e.to_string())
} }
#[allow(dead_code)] /// Why a game connection failed before the first snapshot arrived.
pub async fn game_state(game_id: &str) -> Result<GameView, String> { ///
let resp = Request::get(&format!("/api/games/{game_id}")) /// A refused WebSocket handshake is invisible to the browser: the close
.send() /// event reports an abnormal closure (1006) with no HTTP status, so the
.await /// game page asks the REST API to tell a deliberate rejection from a
.map_err(|e| e.to_string())?; /// transient network problem. Anonymous users and non-players are
if !resp.ok() { /// welcome as spectators, so the only deliberate refusal is an unknown
return Err(server_error(resp.status())); /// game.
pub enum GameEntry {
/// Unknown game (`404`): the id is wrong or the game's TTL expired.
Unavailable,
/// The API could not be asked (or answered unexpectedly): treat the
/// connection loss as transient and retry.
Transient,
}
/// Classify a failed game connection via `GET /api/games/{game_id}`,
/// mirroring the status used to refuse the websocket upgrade.
pub async fn probe_game(game_id: &str) -> GameEntry {
match Request::get(&format!("/api/games/{game_id}")).send().await {
Ok(resp) => match resp.status() {
404 => GameEntry::Unavailable,
_ => GameEntry::Transient,
},
Err(_) => GameEntry::Transient,
} }
resp.json().await.map_err(|e| e.to_string())
} }
pub async fn my_matches(cursor: Option<&str>) -> Result<MatchesPage, String> { pub async fn my_matches(cursor: Option<&str>) -> Result<MatchesPage, String> {
+45
View File
@@ -0,0 +1,45 @@
//! Clipboard copy button, shared by the join-code and watch-link actions.
use std::cell::RefCell;
use std::rc::Rc;
use sycamore::prelude::*;
/// A button that copies `text` to the clipboard and shows "Copied!" for a
/// moment. `label` is the idle caption.
#[component(inline_props)]
pub fn CopyButton(text: String, label: String) -> View {
let copied = create_signal(false);
// Held so the reset timer is cancelled if the button is rebuilt (e.g.
// another player joins) while the "Copied!" feedback is showing:
// firing it afterwards would set a disposed signal and panic.
let reset_timer: Rc<RefCell<Option<gloo_timers::callback::Timeout>>> =
Rc::new(RefCell::new(None));
on_cleanup({
let reset_timer = reset_timer.clone();
move || {
reset_timer.borrow_mut().take();
}
});
let on_click = {
let text = text.clone();
move |_| {
if let Some(win) = web_sys::window() {
let _ = win.navigator().clipboard().write_text(&text);
}
copied.set(true);
*reset_timer.borrow_mut() = Some(gloo_timers::callback::Timeout::new(
1500,
move || {
copied.set(false);
},
));
}
};
view! {
button(class="btn btn-outline-light", on:click=on_click) {
(move || if copied.get() { "Copied!".to_string() } else { label.clone() })
}
}
}
+1
View File
@@ -1,4 +1,5 @@
pub mod card; pub mod card;
pub mod copy;
pub mod nav; pub mod nav;
pub mod summary; pub mod summary;
pub mod toast; pub mod toast;
+8 -1
View File
@@ -260,7 +260,14 @@ pub fn hand_summary_modal(
let totals_view = totals(&game); let totals_view = totals(&game);
let title = format!("Hand {} — results", summary.hand); let title = format!("Hand {} — results", summary.hand);
let action = if already_acked { let action = if game.spectator {
// Spectators cannot acknowledge; they just watch the summary.
view! {
p(class="text-secondary small mb-0") {
"Spectating — the next hand is dealt when every player is ready."
}
}
} else if already_acked {
let waiting_text = format!("Waiting for {}…", waiting.join(", ")); let waiting_text = format!("Waiting for {}…", waiting.join(", "));
view! { view! {
button(class="btn btn-warning", disabled=true) { (waiting_text) } button(class="btn btn-warning", disabled=true) { (waiting_text) }
+107 -1
View File
@@ -31,7 +31,7 @@ pub struct PlayerView {
pub hand: Option<Vec<String>>, pub hand: Option<Vec<String>>,
} }
#[derive(Debug, Clone, Copy, Deserialize)] #[derive(Debug, Clone, Copy, PartialEq, Deserialize)]
pub struct Scores { pub struct Scores {
#[serde(rename = "A")] #[serde(rename = "A")]
pub a: i32, pub a: i32,
@@ -112,11 +112,16 @@ pub struct HandSummary {
#[allow(dead_code)] #[allow(dead_code)]
pub struct GameView { pub struct GameView {
pub id: String, pub id: String,
/// The lobby join code; withheld by the server from spectators.
#[serde(default)] #[serde(default)]
pub join_code: String, pub join_code: String,
/// Which card game this match is (id from /api/game-types). /// Which card game this match is (id from /api/game-types).
#[serde(default)] #[serde(default)]
pub game_type: String, pub game_type: String,
/// Whether the viewer is watching without a seat: no hand, no join
/// code, and actions are rejected by the server.
#[serde(default)]
pub spectator: bool,
/// Whether the napola rule is scored in this match. /// Whether the napola rule is scored in this match.
#[serde(default = "default_true")] #[serde(default = "default_true")]
pub napola: bool, pub napola: bool,
@@ -291,6 +296,13 @@ pub fn fmt_points(points: f64) -> String {
} }
} }
/// Render a calendar instant as `YYYY-MM-DD HH:MM`. `month` is 0-based
/// (the JavaScript `Date` convention); the other components are as
/// returned by `Date`'s local-time getters.
pub fn fmt_datetime_parts(year: u32, month: u32, day: u32, hour: u32, minute: u32) -> String {
format!("{year:04}-{:02}-{day:02} {hour:02}:{minute:02}", month + 1)
}
/// The caller's Elo rating for one game type (GET /api/me/ratings). /// The caller's Elo rating for one game type (GET /api/me/ratings).
#[derive(Debug, Clone, Deserialize)] #[derive(Debug, Clone, Deserialize)]
#[allow(dead_code)] #[allow(dead_code)]
@@ -337,6 +349,19 @@ pub struct GameTypesPage {
pub results: Vec<GameTypeInfo>, pub results: Vec<GameTypeInfo>,
} }
/// One configured OIDC login provider (GET /api/auth/providers).
#[derive(Debug, Clone, PartialEq, Deserialize)]
pub struct AuthProvider {
pub name: String,
pub login_url: String,
}
#[derive(Debug, Clone, Deserialize)]
pub struct AuthProvidersPage {
#[serde(default)]
pub providers: Vec<AuthProvider>,
}
/// A join-code lookup result (GET /api/games/preview): the game's lobby /// A join-code lookup result (GET /api/games/preview): the game's lobby
/// payload plus its display name, for the pre-join summary. The options /// payload plus its display name, for the pre-join summary. The options
/// echoed by the engine land in `details` under their `options_schema` /// echoed by the engine land in `details` under their `options_schema`
@@ -544,6 +569,21 @@ pub fn card_asset(code: &str) -> String {
pub const CARD_BACK: &str = "/assets/cards/back.svg"; pub const CARD_BACK: &str = "/assets/cards/back.svg";
/// Display order key for a hand: rank ascending (01..10, i.e. 1..7 then
/// Fante, Cavallo, Re), then suit (denari, coppe, spade, bastoni).
/// Malformed codes sort last.
pub fn hand_sort_key(code: &str) -> (u8, u8) {
let rank = code.get(..2).and_then(|r| r.parse().ok()).unwrap_or(u8::MAX);
let suit = match code.get(2..3) {
Some("D") => 0,
Some("C") => 1,
Some("S") => 2,
Some("B") => 3,
_ => u8::MAX,
};
(rank, suit)
}
/// Human-friendly rank+suit label, e.g. `07D` -> "7 of denari". /// Human-friendly rank+suit label, e.g. `07D` -> "7 of denari".
pub fn card_label(code: &str) -> String { pub fn card_label(code: &str) -> String {
if code.len() != 3 { if code.len() != 3 {
@@ -593,6 +633,38 @@ mod tests {
assert_eq!("7 di denari", card_label("07D")); assert_eq!("7 di denari", card_label("07D"));
} }
#[test]
fn hand_sort_orders_ranks_ascending() {
let mut hand = vec![
"10B".to_string(),
"01S".to_string(),
"07D".to_string(),
"08C".to_string(),
];
hand.sort_by_key(|c| hand_sort_key(c));
assert_eq!(vec!["01S", "07D", "08C", "10B"], hand);
}
#[test]
fn hand_sort_breaks_ties_by_suit() {
let mut hand = vec![
"02B".to_string(),
"02S".to_string(),
"02C".to_string(),
"02D".to_string(),
"01S".to_string(),
];
hand.sort_by_key(|c| hand_sort_key(c));
assert_eq!(vec!["01S", "02D", "02C", "02S", "02B"], hand);
}
#[test]
fn hand_sort_puts_malformed_codes_last() {
let mut hand = vec!["07D".to_string(), "xx".to_string(), "AB1".to_string()];
hand.sort_by_key(|c| hand_sort_key(c));
assert_eq!(vec!["07D", "xx", "AB1"], hand);
}
fn scopone_match() -> MatchSummary { fn scopone_match() -> MatchSummary {
serde_json::from_value(json!({ serde_json::from_value(json!({
"id": "m1", "id": "m1",
@@ -658,6 +730,12 @@ mod tests {
assert_eq!("2.5", fmt_points(2.5)); assert_eq!("2.5", fmt_points(2.5));
} }
#[test]
fn fmt_datetime_parts_pads_and_offsets_month() {
assert_eq!("2026-09-19 05:02", fmt_datetime_parts(2026, 8, 19, 5, 2));
assert_eq!("2026-01-01 00:00", fmt_datetime_parts(2026, 0, 1, 0, 0));
}
#[test] #[test]
fn leaderboard_points_accept_floats() { fn leaderboard_points_accept_floats() {
let e: LeaderboardEntry = serde_json::from_value(json!({ let e: LeaderboardEntry = serde_json::from_value(json!({
@@ -807,11 +885,39 @@ mod tests {
.unwrap(); .unwrap();
assert_eq!("playing", g.phase); assert_eq!("playing", g.phase);
assert!(g.asso_pigliatutto); assert!(g.asso_pigliatutto);
assert!(!g.spectator);
assert_eq!(None, g.players[0].hand); assert_eq!(None, g.players[0].hand);
assert_eq!(Some(vec!["01D".to_string()]), g.players[1].hand); assert_eq!(Some(vec!["01D".to_string()]), g.players[1].hand);
assert_eq!(Some(true), g.your_turn); assert_eq!(Some(true), g.your_turn);
} }
#[test]
fn game_view_parses_spectator_snapshot() {
// A spectator payload: no hand anywhere, no join code, but the
// public table state is present.
let g: GameView = serde_json::from_value(json!({
"id": "11111111-2222-3333-4444-555555555555",
"game_type": "scopone_scientifico",
"spectator": true,
"phase": "playing",
"target_score": 11,
"scores": {"A": 2, "B": 1},
"table": ["07C"],
"players": [
{"sub": "alice", "name": "Alice", "seat": 0, "team": "A",
"cards_left": 5, "captured_count": 12, "scope": 1},
],
}))
.unwrap();
assert!(g.spectator);
assert_eq!("", g.join_code);
assert_eq!(Some(Scores { a: 2, b: 1 }), g.scores);
assert_eq!(vec!["07C".to_string()], g.table);
assert_eq!(None, g.players[0].hand);
assert_eq!(12, g.players[0].captured_count);
assert_eq!(1, g.players[0].scope);
}
#[test] #[test]
fn move_view_parses_pigliatutto() { fn move_view_parses_pigliatutto() {
let mv: MoveView = serde_json::from_value(json!({ let mv: MoveView = serde_json::from_value(json!({
+101 -58
View File
@@ -1,14 +1,16 @@
//! Live game page: table view over the websocket. //! Live game page: table view over the websocket.
use std::cell::Cell; use std::cell::Cell;
use std::cell::RefCell;
use std::rc::Rc; use std::rc::Rc;
use sycamore::prelude::*; use sycamore::prelude::*;
use wasm_bindgen_futures::spawn_local;
use crate::api;
use crate::components::card::{card_back, card_img}; use crate::components::card::{card_back, card_img};
use crate::components::copy::CopyButton;
use crate::components::summary::{hand_summary_modal, summary_rows}; use crate::components::summary::{hand_summary_modal, summary_rows};
use crate::components::toast::toast; use crate::components::toast::toast;
use crate::model::{card_label, GameView, MoveView, PlayerView, Scores, ServerMessage}; use crate::model::{card_label, hand_sort_key, GameView, MoveView, PlayerView, Scores, ServerMessage};
use crate::ws::{self, GameSocket}; use crate::ws::{self, GameSocket};
fn send_play(socket: Signal<Option<GameSocket>>, card: String, capture: Option<Vec<String>>) { fn send_play(socket: Signal<Option<GameSocket>>, card: String, capture: Option<Vec<String>>) {
@@ -17,6 +19,14 @@ fn send_play(socket: Signal<Option<GameSocket>>, card: String, capture: Option<V
} }
} }
/// The absolute, shareable URL of this game's page, for the clipboard.
fn watch_link(id: &str) -> String {
let origin = web_sys::window()
.and_then(|w| w.location().origin().ok())
.unwrap_or_default();
format!("{origin}/game/{id}")
}
/// Panel for one player seat (name, team, hidden card count, stats). /// Panel for one player seat (name, team, hidden card count, stats).
fn seat_panel(game: GameView, seat: usize, position: &'static str) -> View { fn seat_panel(game: GameView, seat: usize, position: &'static str) -> View {
let Some(player) = game.players.iter().find(|p| p.seat == seat).cloned() else { let Some(player) = game.players.iter().find(|p| p.seat == seat).cloned() else {
@@ -135,23 +145,33 @@ fn start_connect(id: Rc<String>, ctx: ConnCtx, alive: Rc<Cell<bool>>) {
let on_close = { let on_close = {
let id = id.clone(); let id = id.clone();
let alive = alive.clone(); let alive = alive.clone();
move |code: Option<u16>| { move |_code: Option<u16>| {
if !alive.get() { if !alive.get() {
return; return;
} }
ctx.closed.set(true); ctx.closed.set(true);
match code { if ctx.game.get_clone().is_none() && ctx.attempts.get() == 0 {
Some(4401) => { // The connection ended before any snapshot arrived. A
ctx.fatal.set(true); // refused handshake looks exactly like a network drop to
ctx.error // the browser (close code 1006, no HTTP status), so ask
.set(Some("Session expired — please log in again.".to_string())); // the API whether the server deliberately rejected us —
} // in which case retrying is pointless — or the loss was
Some(4403) | Some(4404) => { // transient. Only on the first failure: later attempts
ctx.fatal.set(true); // are governed by the retry backoff.
ctx.error let id = id.clone();
.set(Some("This game is no longer available.".to_string())); let alive = alive.clone();
} spawn_local(async move {
_ => schedule_retry(id.clone(), ctx, alive.clone()), match api::probe_game(&id).await {
api::GameEntry::Unavailable => {
ctx.fatal.set(true);
ctx.error
.set(Some("This game is no longer available.".to_string()));
}
api::GameEntry::Transient => schedule_retry(id, ctx, alive),
}
});
} else {
schedule_retry(id.clone(), ctx, alive.clone());
} }
} }
}; };
@@ -390,38 +410,16 @@ pub fn GamePage(id: String) -> View {
/// Lobby view while waiting for the remaining players. The seat count /// Lobby view while waiting for the remaining players. The seat count
/// comes from the payload (`seats_open`), so games with other player /// comes from the payload (`seats_open`), so games with other player
/// counts render correctly. /// counts render correctly. Spectators (no seat, no join code) get a
/// "not started yet" note and the watch link instead.
#[component(inline_props)] #[component(inline_props)]
fn LobbyView(game: GameView) -> View { fn LobbyView(game: GameView) -> View {
let seats_open = game let seats_open = game
.seats_open .seats_open
.unwrap_or_else(|| 4usize.saturating_sub(game.players.len())); .unwrap_or_else(|| 4usize.saturating_sub(game.players.len()));
let spectator = game.spectator;
let join_code = game.join_code.clone(); let join_code = game.join_code.clone();
let copied = create_signal(false); let watch_link = watch_link(&game.id);
// Held so the reset timer is cancelled if the waiting room is rebuilt
// (e.g. another player joins) while the "Copied!" feedback is showing:
// firing it afterwards would set a disposed signal and panic.
let reset_timer: Rc<RefCell<Option<gloo_timers::callback::Timeout>>> =
Rc::new(RefCell::new(None));
on_cleanup({
let reset_timer = reset_timer.clone();
move || {
reset_timer.borrow_mut().take();
}
});
let copy = {
let join_code = join_code.clone();
move |_| {
if let Some(win) = web_sys::window() {
let _ = win.navigator().clipboard().write_text(&join_code);
}
copied.set(true);
*reset_timer.borrow_mut() = Some(gloo_timers::callback::Timeout::new(1500, move || {
copied.set(false);
}));
}
};
let players = game let players = game
.players .players
@@ -455,18 +453,39 @@ fn LobbyView(game: GameView) -> View {
}; };
let percent_style = format!("width: {percent}%"); let percent_style = format!("width: {percent}%");
// The server withholds the join code from spectators; they get a
// "not started yet" note and can share the watch link instead.
let share = if spectator {
view! {
p(class="text-secondary") {
"This game hasn't started yet — waiting for the players to take their seats."
}
div(class="d-grid mb-4") {
CopyButton(text=watch_link, label="Copy watch link".to_string())
}
}
} else {
// Separate clones: each interpolation/prop captures its own copy.
let display_code = join_code.clone();
let copy_code = join_code.clone();
view! {
p(class="text-secondary") { "Share this join code with your friends:" }
p(class="code-display") { (display_code) }
div(class="d-grid gap-2 mb-4") {
CopyButton(text=copy_code, label="Copy code".to_string())
CopyButton(text=watch_link, label="Copy watch link".to_string())
}
}
};
view! { view! {
div(class="container page-shell py-4") { div(class="container page-shell py-4") {
div(class="card status-card") { div(class="card status-card") {
div(class="card-body p-4") { div(class="card-body p-4") {
h1(class="h4 mb-3") { "Waiting for players" } h1(class="h4 mb-3") {
p(class="text-secondary") { "Share this join code with your friends:" } (if spectator { "Game not started yet" } else { "Waiting for players" })
p(class="code-display") { (join_code) }
div(class="d-grid mb-4") {
button(class="btn btn-outline-light", on:click=copy) {
(move || if copied.get() { "Copied!" } else { "Copy code" })
}
} }
(share)
ul(class="list-group list-group-flush text-start mb-3") { ul(class="list-group list-group-flush text-start mb-3") {
(players) (players)
(open_seats) (open_seats)
@@ -519,7 +538,8 @@ fn table_view(
selected: Signal<Option<String>>, selected: Signal<Option<String>>,
now: Signal<f64>, now: Signal<f64>,
) -> View { ) -> View {
// Own seat: the only player entry carrying a hand. // Own seat: the only player entry carrying a hand. Spectators have no
// hand, so they watch from seat 0's perspective.
let viewer_seat = game let viewer_seat = game
.players .players
.iter() .iter()
@@ -576,7 +596,8 @@ fn table_view(
let viewer = player_for_seat(&game, viewer_seat); let viewer = player_for_seat(&game, viewer_seat);
let my_captured = viewer.as_ref().map(|p| p.captured_count).unwrap_or(0); let my_captured = viewer.as_ref().map(|p| p.captured_count).unwrap_or(0);
let my_scope = viewer.as_ref().map(|p| p.scope).unwrap_or(0); let my_scope = viewer.as_ref().map(|p| p.scope).unwrap_or(0);
let hand = viewer.and_then(|p| p.hand).unwrap_or_default(); let mut hand = viewer.and_then(|p| p.hand).unwrap_or_default();
hand.sort_by_key(|c| hand_sort_key(c));
let current_selection = selected.get_clone(); let current_selection = selected.get_clone();
let hand_cards = hand let hand_cards = hand
.into_iter() .into_iter()
@@ -604,18 +625,41 @@ fn table_view(
} }
}); });
// Spectators have no hand: render the pivot seat as a regular panel
// so its captured count and scope stay visible where the hand would be.
let bottom = if game.spectator {
seat_panel(game.clone(), viewer_seat, "bottom")
} else {
view! {
div(class="seat-bottom") {
div(class="seat-stats own-stats") {
(my_captured) " captured · " (my_scope) " scope"
}
div(class="hand") { (hand_cards) }
(hint)
}
}
};
let spectator_badge = game.spectator.then(|| view! {
span(class="spectator-badge") { "Spectating" }
});
let watch_link = watch_link(&game.id);
view! { view! {
div(class="table-wrap") { div(class="table-wrap") {
div(class="hud") { div(class="hud") {
a(href="/") { "← Lobby" } a(href="/") { "← Lobby" }
span { "Hand " (hand_number) } span { "Hand " (hand_number) }
(rule_badge) (rule_badge)
(spectator_badge)
span(class="hud-scores") { span(class="hud-scores") {
"Team A " (scores.a) " — " (scores.b) " Team B (target " "Team A " (scores.a) " — " (scores.b) " Team B (target "
(target_score) ")" (target_score) ")"
} }
span(class=turn_cls) { (turn_note) } span(class=turn_cls) { (turn_note) }
(countdown) (countdown)
CopyButton(text=watch_link, label="Copy watch link".to_string())
} }
div(class="table-grid") { div(class="table-grid") {
(top) (top)
@@ -628,13 +672,7 @@ fn table_view(
} }
} }
(right) (right)
div(class="seat-bottom") { (bottom)
div(class="seat-stats own-stats") {
(my_captured) " captured · " (my_scope) " scope"
}
div(class="hand") { (hand_cards) }
(hint)
}
} }
} }
} }
@@ -695,6 +733,8 @@ fn game_over_view(over: Option<(Scores, Option<String>)>, game: Option<GameView>
Some((scores, winner)) => { Some((scores, winner)) => {
let winner = winner.unwrap_or_else(|| "?".to_string()); let winner = winner.unwrap_or_else(|| "?".to_string());
let line = format!("Team {winner} wins {} – {}", scores.a, scores.b); let line = format!("Team {winner} wins {} – {}", scores.a, scores.b);
// Spectators have no match history to look at.
let spectator = game.as_ref().is_some_and(|g| g.spectator);
// Explain the final hand's scoring before the result. // Explain the final hand's scoring before the result.
let final_summary = game let final_summary = game
.and_then(|g| g.last_hand) .and_then(|g| g.last_hand)
@@ -702,6 +742,9 @@ fn game_over_view(over: Option<(Scores, Option<String>)>, game: Option<GameView>
let rows = summary_rows(s); let rows = summary_rows(s);
view! { (rows) } view! { (rows) }
}); });
let my_matches = (!spectator).then(|| view! {
a(class="btn btn-outline-light", href="/history") { "My matches" }
});
view! { view! {
div(class="overlay") { div(class="overlay") {
div(class="picker summary-panel") { div(class="picker summary-panel") {
@@ -710,7 +753,7 @@ fn game_over_view(over: Option<(Scores, Option<String>)>, game: Option<GameView>
p(class="final-score") { (line) } p(class="final-score") { (line) }
div(class="gameover-actions") { div(class="gameover-actions") {
a(class="btn btn-warning", href="/") { "Back to lobby" } a(class="btn btn-warning", href="/") { "Back to lobby" }
a(class="btn btn-outline-light", href="/history") { "My matches" } (my_matches)
} }
} }
} }
+18 -6
View File
@@ -68,6 +68,23 @@ fn game_name(game_types: &[GameTypeInfo], game_type: &str) -> String {
.unwrap_or_else(|| game_type.to_string()) .unwrap_or_else(|| game_type.to_string())
} }
/// The match's finish instant in the browser's local timezone, as
/// `YYYY-MM-DD HH:MM`. Falls back to the raw UTC prefix when the browser
/// cannot parse the timestamp.
fn local_finished(iso: &str) -> String {
let date = js_sys::Date::new(&wasm_bindgen::JsValue::from_str(iso));
if date.get_time().is_nan() {
return iso.replace('T', " ").chars().take(16).collect();
}
crate::model::fmt_datetime_parts(
date.get_full_year(),
date.get_month(),
date.get_date(),
date.get_hours(),
date.get_minutes(),
)
}
#[component] #[component]
pub fn HistoryPage() -> View { pub fn HistoryPage() -> View {
let page = create_signal(Option::<MatchesPage>::None); let page = create_signal(Option::<MatchesPage>::None);
@@ -131,12 +148,7 @@ pub fn HistoryPage() -> View {
.get_clone() .get_clone()
.into_iter() .into_iter()
.map(|m| { .map(|m| {
let finished = m let finished = local_finished(&m.finished_at);
.finished_at
.replace('T', " ")
.chars()
.take(16)
.collect::<String>();
let game = game_name(&names, &m.game_type); let game = game_name(&names, &m.game_type);
let mut cells: Vec<View> = team_columns(&m) let mut cells: Vec<View> = team_columns(&m)
.iter() .iter()
+53 -10
View File
@@ -14,7 +14,9 @@ use sycamore_router::navigate;
use crate::api; use crate::api;
use crate::components::card::card_back; use crate::components::card::card_back;
use crate::components::toast::toast; use crate::components::toast::toast;
use crate::model::{GameTypeInfo, JoinPreview, OptionField, OptionKind, PlayerRating, User}; use crate::model::{
AuthProvider, GameTypeInfo, JoinPreview, OptionField, OptionKind, PlayerRating, User,
};
/// Used when the game-types fetch fails: match creation must still work. /// Used when the game-types fetch fails: match creation must still work.
fn fallback_game_types() -> Vec<GameTypeInfo> { fn fallback_game_types() -> Vec<GameTypeInfo> {
@@ -28,6 +30,15 @@ fn fallback_game_types() -> Vec<GameTypeInfo> {
}] }]
} }
/// Human-readable label for a provider name ("google" -> "Google").
fn provider_label(name: &str) -> String {
let mut chars = name.chars();
match chars.next() {
Some(first) => first.to_uppercase().collect::<String>() + chars.as_str(),
None => String::new(),
}
}
fn json_to_string(value: &serde_json::Value) -> String { fn json_to_string(value: &serde_json::Value) -> String {
match value { match value {
serde_json::Value::String(s) => s.clone(), serde_json::Value::String(s) => s.clone(),
@@ -283,6 +294,8 @@ fn join_preview_panel(
pub fn LobbyPage() -> View { pub fn LobbyPage() -> View {
// Outer None = still loading; Some(None) = logged out. // Outer None = still loading; Some(None) = logged out.
let user = create_signal(Option::<Option<User>>::None); let user = create_signal(Option::<Option<User>>::None);
// Configured login providers, for the logged-out card's buttons.
let auth_providers = create_signal(Vec::<AuthProvider>::new());
// The player's Elo ratings, one row per game type played. // The player's Elo ratings, one row per game type played.
let ratings = create_signal(Vec::<PlayerRating>::new()); let ratings = create_signal(Vec::<PlayerRating>::new());
let error = create_signal(Option::<String>::None); let error = create_signal(Option::<String>::None);
@@ -400,6 +413,12 @@ pub fn LobbyPage() -> View {
} }
}); });
spawn_local(async move {
if let Ok(list) = api::auth_providers().await {
auth_providers.set(list);
}
});
let on_create = move |_| { let on_create = move |_| {
let game_type = selected_game.get_clone(); let game_type = selected_game.get_clone();
let opts: serde_json::Map<String, serde_json::Value> = let opts: serde_json::Map<String, serde_json::Value> =
@@ -462,15 +481,39 @@ pub fn LobbyPage() -> View {
} }
p(class="text-secondary") { "Log in with your account to play." } p(class="text-secondary") { "Log in with your account to play." }
// rel="external" opts out of the SPA router's click // rel="external" opts out of the SPA router's click
// interception: this must be a full page navigation // interception: these must be full page navigations
// to the backend's OIDC login endpoint. // to the backend's OIDC login endpoints.
a( (move || {
class="btn btn-warning btn-lg w-100", let providers = auth_providers.get_clone();
href="/auth/login", if providers.is_empty() {
rel="external", // Providers could not be fetched: the
) { // generic route redirects to the first
"Log in" // configured provider.
} view! {
a(
class="btn btn-warning btn-lg w-100",
href="/auth/login",
rel="external",
) {
"Log in"
}
}
} else {
view! {
div(class="d-grid gap-2") {
(providers.iter().cloned().map(|p| view! {
a(
class="btn btn-warning btn-lg w-100",
href=p.login_url,
rel="external",
) {
"Log in with " (provider_label(&p.name))
}
}).collect::<Vec<_>>())
}
}
}
})
} }
} }
} }
+5 -2
View File
@@ -55,8 +55,11 @@ impl GameSocket {
/// could not be created. /// could not be created.
/// ///
/// `on_close` fires exactly once when the connection ends; it receives the /// `on_close` fires exactly once when the connection ends; it receives the
/// server close code when one was sent (e.g. 4401 unauthenticated, 4403 not /// server close code when one was sent (e.g. 1000 for a normal closure) or
/// seated, 4404 unknown game) or `None` for an abnormal network loss. /// `None` for an abnormal loss. A *refused* handshake is not distinguishable
/// here: browsers report it as an abnormal closure (1006) and never expose
/// the HTTP status the server answered with, so callers must classify the
/// failure out of band.
pub fn connect( pub fn connect(
game_id: &str, game_id: &str,
on_message: impl Fn(ServerMessage) + 'static, on_message: impl Fn(ServerMessage) + 'static,
+9
View File
@@ -266,6 +266,15 @@ body {
margin-left: 0.25rem; margin-left: 0.25rem;
} }
.spectator-badge {
background: var(--panel-light);
color: var(--muted);
font-weight: 700;
border-radius: 999px;
padding: 0.15rem 0.6rem;
border: 1px solid var(--muted);
}
/* ---------- overlays ---------- */ /* ---------- overlays ---------- */
.overlay { .overlay {