diff --git a/backend/alembic/versions/0006_add_presence_columns.py b/backend/alembic/versions/0006_add_presence_columns.py new file mode 100644 index 0000000..65f391b --- /dev/null +++ b/backend/alembic/versions/0006_add_presence_columns.py @@ -0,0 +1,33 @@ +"""add presence columns to users + +Revision ID: 0006_add_presence_columns +Revises: 0005_add_world_content_tables +Create Date: 2026-09-14 00:00:00.000000 + +Ported forward from the orphaned agent/dd2a126b57a1 branch (originally +2a9b9873244a_add_presence_columns.py, down_revision 0001_initial) and +rebased onto the current migration head. See app/services/presence.py / +app/routers/presence.py. +""" +from typing import Sequence, Union + +from alembic import op +import sqlalchemy as sa + +# revision identifiers, used by Alembic. +revision: str = "0006_add_presence_columns" +down_revision: Union[str, None] = "0005_add_world_content_tables" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + """Add last_seen_at and status_message columns to users table.""" + op.add_column("users", sa.Column("last_seen_at", sa.DateTime(timezone=True), nullable=True)) + op.add_column("users", sa.Column("status_message", sa.Text(), nullable=True)) + + +def downgrade() -> None: + """Remove presence columns from users table.""" + op.drop_column("users", "status_message") + op.drop_column("users", "last_seen_at") diff --git a/backend/app/main.py b/backend/app/main.py index 5c139a2..e6dcd48 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -6,7 +6,7 @@ from fastapi import FastAPI from fastapi.middleware.cors import CORSMiddleware from app.config import settings -from app.routers import auth, users, worlds, assets, avatar, world_content, internal +from app.routers import auth, users, worlds, assets, avatar, world_content, internal, presence from app.services.game_servers import supervisor @@ -44,6 +44,7 @@ app.include_router(world_content.router, prefix="/api/worlds", tags=["world-cont app.include_router(assets.router, prefix="/api/assets", tags=["assets"]) app.include_router(avatar.router, prefix="/api/avatar", tags=["avatar"]) app.include_router(internal.router, prefix="/api/internal", tags=["internal"]) +app.include_router(presence.router, prefix="/api/presence", tags=["presence"]) @app.get("/api/health") diff --git a/backend/app/models/user.py b/backend/app/models/user.py index 59ad556..02257b5 100644 --- a/backend/app/models/user.py +++ b/backend/app/models/user.py @@ -1,7 +1,7 @@ from datetime import date from typing import Optional -from sqlalchemy import Column, Integer, String, Boolean, DateTime, Date +from sqlalchemy import Column, Integer, String, Boolean, DateTime, Date, Text from sqlalchemy.orm import relationship from sqlalchemy.sql import func @@ -20,6 +20,10 @@ class User(Base): created_at = Column(DateTime(timezone=True), server_default=func.now()) updated_at = Column(DateTime(timezone=True), onupdate=func.now()) + # Presence / status (backend/app/services/presence.py, app/routers/presence.py) + last_seen_at = Column(DateTime(timezone=True), nullable=True) + status_message = Column(Text, nullable=True) + sessions = relationship("Session", back_populates="user", cascade="all, delete-orphan") refresh_tokens = relationship("RefreshToken", back_populates="user", cascade="all, delete-orphan") worlds = relationship("World", back_populates="owner") diff --git a/backend/app/routers/presence.py b/backend/app/routers/presence.py new file mode 100644 index 0000000..e74d9a3 --- /dev/null +++ b/backend/app/routers/presence.py @@ -0,0 +1,146 @@ +"""Presence & status router. + +Endpoints: +- POST /api/presence/heartbeat -- record a heartbeat (sets Redis TTL key) +- GET /api/presence/presence -- query presence for a list of user ids +- PATCH /api/presence/status -- update the authenticated user's status message +""" + +from datetime import datetime, timezone +from typing import Annotated + +from fastapi import APIRouter, HTTPException, status, Depends +from pydantic import BaseModel, Field, field_validator +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession + +from app.database import get_db +from app.dependencies import CurrentUser +from app.models.user import User +from app.services.presence import presence_store + +router = APIRouter() + +DbDep = Annotated[AsyncSession, Depends(get_db)] + +# ──────────────────────────── Pydantic schemas ──────────────────────────────── + +_STATUS_MAX_LEN = 140 + + +class HeartbeatRequest(BaseModel): + """Empty body -- the caller's identity comes from the JWT.""" + + +class PresenceUser(BaseModel): + user_id: int + is_online: bool + last_seen_at: datetime | None + status_message: str | None + + +class PresenceResponse(BaseModel): + users: list[PresenceUser] + + +class StatusUpdate(BaseModel): + status_message: str | None = Field(default=None, max_length=_STATUS_MAX_LEN) + + @field_validator("status_message") + @classmethod + def strip_message(cls, v: str | None) -> str | None: + if v is not None: + v = v.strip() + return v + + +# ──────────────────────────────── Endpoints ─────────────────────────────────── + + +@router.post("/heartbeat", response_model=dict) +async def heartbeat( + _req: HeartbeatRequest, + current_user: CurrentUser, + db: DbDep, +): + """Record a heartbeat for the authenticated user. + + Sets a Redis key with a TTL so the server can derive online/offline + status without trusting client claims. Also persists last_seen_at to + the database. + """ + # Update Redis TTL (presence heartbeat) + await presence_store.heartbeat(current_user.id) + + # Persist last_seen_at to the database + now = datetime.now(timezone.utc) + result = await db.execute(select(User).where(User.id == current_user.id)) + user = result.scalar_one_or_none() + if user: + user.last_seen_at = now + await db.commit() + + return {"status": "ok", "user_id": current_user.id} + + +@router.get("/presence", response_model=PresenceResponse) +async def get_presence( + user_ids: str, + current_user: CurrentUser, + db: DbDep, +): + """Return presence info for a comma-separated list of user ids. + + Online/offline is derived from the Redis heartbeat TTL. + last_seen_at and status_message come from the database. + """ + ids = [int(uid.strip()) for uid in user_ids.split(",") if uid.strip()] + if not ids: + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail="user_ids must contain at least one integer", + ) + + # Check Redis for online status (single round trip via MGET). + online_ids = await presence_store.is_online_batch(ids) + + # Fetch DB records for last_seen_at and status_message + result = await db.execute(select(User).where(User.id.in_(ids))) + users = {u.id: u for u in result.scalars().all()} + + presence_users = [] + for uid in ids: + user = users.get(uid) + if user is None: + continue + presence_users.append( + PresenceUser( + user_id=uid, + is_online=uid in online_ids, + last_seen_at=user.last_seen_at, + status_message=user.status_message, + ) + ) + + return PresenceResponse(users=presence_users) + + +@router.patch("/status", response_model=dict) +async def update_status( + body: StatusUpdate, + current_user: CurrentUser, + db: DbDep, +): + """Update the authenticated user's status message (length-limited).""" + result = await db.execute(select(User).where(User.id == current_user.id)) + user = result.scalar_one_or_none() + if not user: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail="User not found", + ) + + user.status_message = body.status_message + await db.commit() + + return {"status": "ok", "status_message": body.status_message} diff --git a/backend/app/services/presence.py b/backend/app/services/presence.py new file mode 100644 index 0000000..c3c1105 --- /dev/null +++ b/backend/app/services/presence.py @@ -0,0 +1,145 @@ +"""Redis-backed presence store. + +Presence is determined by a heartbeat TTL key in Redis: +- heartbeat(user_id) sets a TTL key per user (default 60 s). +- is_online(user_id) / is_online_batch(user_ids) check whether the key(s) + exist. + +status_message and last_seen_at are persisted in the ``users`` table by the +router (app/routers/presence.py); this store only owns the Redis side. + +The Redis client is created lazily, on first actual use, rather than +requiring an explicit ``connect()`` call during app startup. This matters +because FastAPI's lifespan handler does not run under the plain +``TestClient(app)`` instantiation this codebase's tests use (no +``with`` block), so anything that only connected during lifespan startup +would never connect in tests. Lazy connection means the store works +identically whether the app was started for real or is being driven by +pytest. + +The client is also cached per-*event loop* rather than globally: the same +plain ``TestClient(app)`` pattern can dispatch different requests onto +different event loops (see the NullPool comment on ``_override_db`` in +tests/conftest.py for the analogous asyncpg problem), and an +``aioredis.Redis`` connection, like an asyncpg connection, breaks with +"Event loop is closed" if reused across loops. In the real running server +there is exactly one event loop for the process lifetime, so this +degrades to a single cached, pooled connection there. +""" + +from __future__ import annotations + +import asyncio +import logging + +import redis.asyncio as aioredis + +from app.config import settings + +logger = logging.getLogger(__name__) + +# Key prefix for heartbeat entries. +_PRESENCE_KEY_PREFIX = "presence:heartbeat:" + +# Default heartbeat TTL in seconds (1 minute). +# Clients should heartbeat at least once per interval. +_DEFAULT_HEARTBEAT_TTL = 60 + + +class PresenceStore: + """Redis-backed presence store with async, lazily-connected operations.""" + + def __init__(self, redis_url: str | None = None, heartbeat_ttl: int = _DEFAULT_HEARTBEAT_TTL): + # redis_url is resolved from settings at first use (not at + # construction time) so tests that override settings.REDIS_URL + # before the first call still take effect -- this module is + # imported (and the module-level singleton below constructed) + # well before any such override runs. + self._redis_url = redis_url + self._heartbeat_ttl = heartbeat_ttl + # Keyed by id(running loop) -- see the module docstring. + self._clients: dict[int, aioredis.Redis] = {} + self._connect_lock = asyncio.Lock() + + # ------------------------------------------------------------------ + # Lifecycle + # ------------------------------------------------------------------ + + async def _ensure_client(self) -> aioredis.Redis: + """Return a connected client for the current event loop, connecting + on first use per loop.""" + loop_key = id(asyncio.get_running_loop()) + client = self._clients.get(loop_key) + if client is None: + async with self._connect_lock: + client = self._clients.get(loop_key) + if client is None: + url = self._redis_url or settings.REDIS_URL + client = aioredis.from_url( + url, + encoding="utf-8", + decode_responses=True, + ) + try: + await client.ping() + except Exception: + logger.warning("Redis ping failed for %s", url) + raise + self._clients[loop_key] = client + return client + + async def close(self) -> None: + """Close the Redis connection for the current event loop, if one was + opened.""" + loop_key = id(asyncio.get_running_loop()) + client = self._clients.pop(loop_key, None) + if client is not None: + await client.aclose() + + # ------------------------------------------------------------------ + # Heartbeat + # ------------------------------------------------------------------ + + async def heartbeat(self, user_id: int) -> None: + """Record a heartbeat for *user_id*. + + Sets a TTL key so that the user is considered online as long as + heartbeats arrive at least once per heartbeat_ttl seconds. + """ + client = await self._ensure_client() + await client.set(self._heartbeat_key(user_id), "1", ex=self._heartbeat_ttl) + + # ------------------------------------------------------------------ + # Online checks + # ------------------------------------------------------------------ + + async def is_online(self, user_id: int) -> bool: + """Return True if the user has a valid heartbeat key.""" + client = await self._ensure_client() + exists = await client.exists(self._heartbeat_key(user_id)) + return bool(exists) + + async def is_online_batch(self, user_ids: list[int]) -> set[int]: + """Return the subset of user_ids that currently have a valid + heartbeat key. Uses MGET (one round trip) rather than one EXISTS + per user. + """ + if not user_ids: + return set() + client = await self._ensure_client() + keys = [self._heartbeat_key(uid) for uid in user_ids] + values = await client.mget(keys) + return {uid for uid, v in zip(user_ids, values) if v is not None} + + # ------------------------------------------------------------------ + # Helpers + # ------------------------------------------------------------------ + + @staticmethod + def _heartbeat_key(user_id: int) -> str: + return f"{_PRESENCE_KEY_PREFIX}{user_id}" + + +# Singleton instance shared by the router, so all requests share the same +# connection pool. +presence_store = PresenceStore() diff --git a/backend/tests/conftest.py b/backend/tests/conftest.py index 26a1fae..fff67a8 100644 --- a/backend/tests/conftest.py +++ b/backend/tests/conftest.py @@ -1,15 +1,18 @@ -"""Shared pytest fixtures + hermetic-DB bootstrap. +"""Shared pytest fixtures + hermetic-DB/Redis bootstrap. The integration tests hit a real Postgres (Postgres-specific types: JSONB, -native ENUMs). To make ``cd backend && pytest`` work from a clean checkout -with no manual setup, ``pytest_configure`` below brings up an ephemeral -Postgres via docker compose and runs the migrations IF ``DATABASE_URL`` is -not already pointing at a reachable database. When it is (dev loop, or the -checkpoint runner which stands the DB up itself), that path is skipped and -the existing database is reused. +native ENUMs) and, for presence tests, a real Redis (TTL semantics aren't +worth faking). To make ``cd backend && pytest`` work from a clean checkout +with no manual setup, ``pytest_configure`` below brings up ephemeral +Postgres/Redis containers via docker compose and runs the migrations IF +``DATABASE_URL``/``REDIS_URL`` aren't already pointing at something +reachable. When they are (dev loop, or the checkpoint runner which stands +these up itself), that path is skipped and the existing services are +reused. -If Docker isn't available and no DB is reachable, the whole DB-backed suite -is skipped with a clear message rather than erroring out 60+ times. +If Docker isn't available and nothing is reachable, the whole +DB/Redis-backed suite is skipped with a clear message rather than erroring +out dozens of times. """ import asyncio @@ -44,8 +47,9 @@ TEST_TABLES = ( BACKEND_DIR = Path(__file__).resolve().parents[1] _COMPOSE_PROJECT = "agp-pytest" _EPHEMERAL_PG_PORT = 25432 +_EPHEMERAL_REDIS_PORT = 26379 -# Set by _ensure_database() when it starts a container it owns; torn down in +# Set by pytest_configure() when it starts container(s) it owns; torn down in # pytest_unconfigure. _started_container = False _db_unavailable_reason: str | None = None @@ -64,6 +68,11 @@ def _db_reachable(url: str) -> bool: return _port_open(parts.hostname or "localhost", parts.port or 5432) +def _redis_reachable(url: str) -> bool: + parts = urlsplit(url) + return _port_open(parts.hostname or "localhost", parts.port or 6379) + + def _run(cmd: list[str], **kw) -> subprocess.CompletedProcess: return subprocess.run(cmd, cwd=BACKEND_DIR, capture_output=True, text=True, **kw) @@ -71,14 +80,17 @@ def _run(cmd: list[str], **kw) -> subprocess.CompletedProcess: def pytest_configure(config): global _started_container, _db_unavailable_reason - if _db_reachable(settings.DATABASE_URL): - return # dev loop / checkpoint runner already provided one + need_db = not _db_reachable(settings.DATABASE_URL) + need_redis = not _redis_reachable(settings.REDIS_URL) + + if not need_db and not need_redis: + return # dev loop / checkpoint runner already provided both if not _run(["docker", "--version"]).returncode == 0: _db_unavailable_reason = ( - "no reachable DATABASE_URL and docker is unavailable -- run the " - "checkpoint suite (scripts/run_checkpoint_tests.py) or start Postgres " - "manually" + "no reachable DATABASE_URL/REDIS_URL and docker is unavailable -- run " + "the checkpoint suite (scripts/run_checkpoint_tests.py) or start " + "Postgres/Redis manually" ) return @@ -86,33 +98,55 @@ def pytest_configure(config): env["COMPOSE_PROJECT_NAME"] = _COMPOSE_PROJECT env["POSTGRES_HOST_PORT"] = str(_EPHEMERAL_PG_PORT) env["POSTGRES_PORT"] = str(_EPHEMERAL_PG_PORT) - up = _run(["docker", "compose", "up", "-d", "postgres"], env=env) + env["REDIS_HOST_PORT"] = str(_EPHEMERAL_REDIS_PORT) + + services = [s for s, need in (("postgres", need_db), ("redis", need_redis)) if need] + up = _run(["docker", "compose", "up", "-d", *services], env=env) if up.returncode != 0: - _db_unavailable_reason = f"docker compose up postgres failed:\n{up.stderr[-800:]}" + _db_unavailable_reason = f"docker compose up {' '.join(services)} failed:\n{up.stderr[-800:]}" return _started_container = True - for _ in range(30): - r = _run( - ["docker", "compose", "exec", "-T", "postgres", - "pg_isready", "-U", "agp_user", "-d", "agp_db"], - env=env, - ) - if r.returncode == 0: - break - time.sleep(1) - else: - _db_unavailable_reason = "ephemeral Postgres never became ready" - return + if need_db: + for _ in range(30): + r = _run( + ["docker", "compose", "exec", "-T", "postgres", + "pg_isready", "-U", "agp_user", "-d", "agp_db"], + env=env, + ) + if r.returncode == 0: + break + time.sleep(1) + else: + _db_unavailable_reason = "ephemeral Postgres never became ready" + return - url = f"postgresql+asyncpg://agp_user:agp_password@localhost:{_EPHEMERAL_PG_PORT}/agp_db" - os.environ["DATABASE_URL"] = url - settings.DATABASE_URL = url + url = f"postgresql+asyncpg://agp_user:agp_password@localhost:{_EPHEMERAL_PG_PORT}/agp_db" + os.environ["DATABASE_URL"] = url + settings.DATABASE_URL = url - mig = _run([sys.executable, "-m", "alembic", "upgrade", "head"], - env={**os.environ, "DATABASE_URL": url}) - if mig.returncode != 0: - _db_unavailable_reason = f"alembic upgrade head failed:\n{mig.stderr[-800:]}" + mig = _run([sys.executable, "-m", "alembic", "upgrade", "head"], + env={**os.environ, "DATABASE_URL": url}) + if mig.returncode != 0: + _db_unavailable_reason = f"alembic upgrade head failed:\n{mig.stderr[-800:]}" + return + + if need_redis: + for _ in range(30): + r = _run( + ["docker", "compose", "exec", "-T", "redis", "redis-cli", "ping"], + env=env, + ) + if r.returncode == 0 and "PONG" in r.stdout: + break + time.sleep(1) + else: + _db_unavailable_reason = "ephemeral Redis never became ready" + return + + redis_url = f"redis://localhost:{_EPHEMERAL_REDIS_PORT}/0" + os.environ["REDIS_URL"] = redis_url + settings.REDIS_URL = redis_url def pytest_unconfigure(config): @@ -122,6 +156,7 @@ def pytest_unconfigure(config): env["COMPOSE_PROJECT_NAME"] = _COMPOSE_PROJECT env["POSTGRES_HOST_PORT"] = str(_EPHEMERAL_PG_PORT) env["POSTGRES_PORT"] = str(_EPHEMERAL_PG_PORT) + env["REDIS_HOST_PORT"] = str(_EPHEMERAL_REDIS_PORT) _run(["docker", "compose", "down", "-v"], env=env) diff --git a/backend/tests/test_presence.py b/backend/tests/test_presence.py new file mode 100644 index 0000000..6f50bd7 --- /dev/null +++ b/backend/tests/test_presence.py @@ -0,0 +1,221 @@ +"""Integration tests for the presence feature: + +- POST /api/presence/heartbeat +- GET /api/presence/presence +- PATCH /api/presence/status + +Hits the real app against the configured Postgres *and* a real Redis +(conftest.py brings up an ephemeral instance of each when neither is +already reachable) -- presence is fundamentally "does a TTL key exist in +Redis", which isn't worth faking with a mock. + +Also covers the regression fixed in 47ab492 ("Fix StatusUpdate field +annotation and is_online batch-call mismatch"): the router used to call +``PresenceStore.is_online(ids)`` with a *list* of ids and then evaluate +``uid in online_ids`` against the returned bool, which raised a TypeError +on the first real multi-id batch call, and ``StatusUpdate.status_message`` +used the invalid annotation ``str | Field(...)``. Both are exercised below. +""" + +import asyncio + +import pytest +from fastapi.testclient import TestClient + +from app.main import app +from app.services.presence import PresenceStore + + +def _register(client: TestClient, username: str) -> dict: + resp = client.post( + "/api/auth/register", + json={ + "username": username, + "email": f"{username}@example.com", + "password": "secret123", + "date_of_birth": "1990-01-01", + }, + ) + assert resp.status_code == 201, resp.text + token = resp.json()["access_token"] + headers = {"Authorization": f"Bearer {token}"} + me = client.get("/api/users/me", headers=headers) + assert me.status_code == 200, me.text + return {"headers": headers, "id": me.json()["id"]} + + +class TestHeartbeat: + def test_heartbeat_sets_online(self): + client = TestClient(app) + user = _register(client, "presence_hb1") + + resp = client.post("/api/presence/heartbeat", json={}, headers=user["headers"]) + assert resp.status_code == 200, resp.text + assert resp.json() == {"status": "ok", "user_id": user["id"]} + + presence = client.get( + "/api/presence/presence", + params={"user_ids": str(user["id"])}, + headers=user["headers"], + ) + assert presence.status_code == 200, presence.text + body = presence.json()["users"] + assert len(body) == 1 + assert body[0]["user_id"] == user["id"] + assert body[0]["is_online"] is True + assert body[0]["last_seen_at"] is not None + + def test_requires_auth(self): + client = TestClient(app) + resp = client.post("/api/presence/heartbeat", json={}) + assert resp.status_code == 401 + + def test_never_seen_user_is_offline_with_null_last_seen(self): + client = TestClient(app) + user = _register(client, "presence_never_hb") + + presence = client.get( + "/api/presence/presence", + params={"user_ids": str(user["id"])}, + headers=user["headers"], + ) + assert presence.status_code == 200 + body = presence.json()["users"][0] + assert body["is_online"] is False + assert body["last_seen_at"] is None + + +class TestBatchPresence: + """Covers the batch-call path fixed in 47ab492 (is_online_batch via MGET).""" + + def test_batch_query_mixes_online_offline_and_unknown_ids(self): + client = TestClient(app) + online_user = _register(client, "presence_batch_online") + offline_user = _register(client, "presence_batch_offline") + + client.post("/api/presence/heartbeat", json={}, headers=online_user["headers"]) + + unknown_id = max(online_user["id"], offline_user["id"]) + 999_999 + ids = f"{online_user['id']},{offline_user['id']},{unknown_id}" + + resp = client.get( + "/api/presence/presence", + params={"user_ids": ids}, + headers=online_user["headers"], + ) + assert resp.status_code == 200, resp.text + by_id = {u["user_id"]: u for u in resp.json()["users"]} + + # Unknown id is silently dropped (not in the DB), not a 500 or 404. + assert unknown_id not in by_id + assert by_id[online_user["id"]]["is_online"] is True + assert by_id[offline_user["id"]]["is_online"] is False + + def test_empty_user_ids_is_a_400(self): + client = TestClient(app) + user = _register(client, "presence_empty_ids") + resp = client.get( + "/api/presence/presence", + params={"user_ids": ""}, + headers=user["headers"], + ) + assert resp.status_code == 400 + + +class TestStatusMessage: + def test_update_and_read_back_status_message(self): + client = TestClient(app) + user = _register(client, "presence_status1") + + resp = client.patch( + "/api/presence/status", + json={"status_message": " brb, making coffee "}, + headers=user["headers"], + ) + assert resp.status_code == 200, resp.text + # The router strips whitespace (StatusUpdate.strip_message). + assert resp.json()["status_message"] == "brb, making coffee" + + presence = client.get( + "/api/presence/presence", + params={"user_ids": str(user["id"])}, + headers=user["headers"], + ) + assert presence.json()["users"][0]["status_message"] == "brb, making coffee" + + def test_status_message_over_max_length_is_rejected(self): + client = TestClient(app) + user = _register(client, "presence_status_toolong") + resp = client.patch( + "/api/presence/status", + json={"status_message": "x" * 141}, + headers=user["headers"], + ) + # StatusUpdate.status_message: str | None = Field(..., max_length=140) + # This is exactly the annotation 47ab492 fixed (it used to be the + # invalid `str | Field(...)`, which breaks model construction + # entirely rather than validating length). + assert resp.status_code == 422 + + def test_status_message_at_max_length_is_accepted(self): + client = TestClient(app) + user = _register(client, "presence_status_maxlen") + resp = client.patch( + "/api/presence/status", + json={"status_message": "x" * 140}, + headers=user["headers"], + ) + assert resp.status_code == 200, resp.text + + def test_null_status_message_clears_it(self): + client = TestClient(app) + user = _register(client, "presence_status_clear") + client.patch("/api/presence/status", json={"status_message": "hi"}, headers=user["headers"]) + resp = client.patch("/api/presence/status", json={"status_message": None}, headers=user["headers"]) + assert resp.status_code == 200 + assert resp.json()["status_message"] is None + + +class TestPresenceStoreTTL: + """Exercises the real Redis TTL semantics directly against the service, + with a short heartbeat_ttl so the test doesn't need to sleep for the + real 60s default used in production. This is the cheap, real proxy for + "TTL expiry means offline" the task asked for -- same code path + (real Redis SET ... EX / EXISTS / MGET), just a much shorter window. + """ + + async def test_ttl_expiry_flips_online_to_offline(self): + store = PresenceStore(heartbeat_ttl=1) + try: + uid = 987654321 # doesn't need to exist in Postgres; Redis-only check + assert await store.is_online(uid) is False + + await store.heartbeat(uid) + assert await store.is_online(uid) is True + + await asyncio.sleep(1.5) + assert await store.is_online(uid) is False + finally: + await store.close() + + async def test_is_online_batch_matches_is_online_semantics(self): + """Regression coverage for 47ab492: is_online_batch must return a + set of ids usable with `in`, not a bool.""" + store = PresenceStore(heartbeat_ttl=30) + try: + online_uid, offline_uid = 111222333, 444555666 + await store.heartbeat(online_uid) + + result = await store.is_online_batch([online_uid, offline_uid]) + assert isinstance(result, set) + assert online_uid in result + assert offline_uid not in result + finally: + await store.close() + + async def test_is_online_batch_empty_list(self): + store = PresenceStore() + try: + assert await store.is_online_batch([]) == set() + finally: + await store.close()