"""Redis-backed presence store. Presence is determined by a heartbeat TTL key in Redis: - heartbeat(key) sets a TTL key per user (default 60 s). - is_online(user_id) checks whether the key exists. - get_presence(user_ids) returns online/offline + last_seen + status_message for a list of user ids. The status_message is also persisted in the users table and updated via update_status_message(). """ from __future__ import annotations import logging from datetime import datetime, timezone from typing import Any 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 # Maximum length for status messages (characters). _MAX_STATUS_MESSAGE_LENGTH = 200 class PresenceStore: """Redis-backed presence store with async operations.""" def __init__(self, redis_url: str = settings.REDIS_URL, heartbeat_ttl: int = _DEFAULT_HEARTBEAT_TTL): self._redis_url = redis_url self._heartbeat_ttl = heartbeat_ttl self._client: aioredis.Redis | None = None # ------------------------------------------------------------------ # Lifecycle # ------------------------------------------------------------------ async def connect(self) -> None: """Connect to Redis.""" self._client = aioredis.from_url( self._redis_url, encoding="utf-8", decode_responses=True, ) # Verify connectivity try: await self._client.ping() except Exception as exc: logger.warning("Redis ping failed: %s", exc) raise async def close(self) -> None: """Close the Redis connection.""" if self._client is not None: await self._client.close() self._client = None @property def client(self) -> aioredis.Redis: """Return the Redis client, raising if not connected.""" if self._client is None: raise RuntimeError("PresenceStore is not connected. Call connect() first.") return self._client # ------------------------------------------------------------------ # 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. """ key = self._heartbeat_key(user_id) await self.client.set(key, "1", ex=self._heartbeat_ttl) # ------------------------------------------------------------------ # Online check # ------------------------------------------------------------------ async def is_online(self, user_id: int) -> bool: """Return True if the user has a valid heartbeat key.""" key = self._heartbeat_key(user_id) exists = await self.client.exists(key) return bool(exists) # ------------------------------------------------------------------ # Presence lookup # ------------------------------------------------------------------ async def get_presence( self, user_ids: list[int], db: Any, # AsyncSession — imported here to avoid circular imports ) -> dict[int, dict[str, Any]]: """Return presence info for a list of user ids. Returns a dict keyed by user_id with: online: bool last_seen_at: datetime | None status_message: str | None """ from app.models.user import User from sqlalchemy import select result = await db.execute(select(User).where(User.id.in_(user_ids))) users = result.scalars().all() user_map = {u.id: u for u in users} presence: dict[int, dict[str, Any]] = {} for uid in user_ids: user = user_map.get(uid) online = await self.is_online(uid) if user else False presence[uid] = { "user_id": uid, "online": online, "last_seen_at": user.last_seen_at if user else None, "status_message": user.status_message if user else None, } return presence # ------------------------------------------------------------------ # Status message # ------------------------------------------------------------------ async def update_status_message( self, user_id: int, message: str | None, db: Any, # AsyncSession ) -> None: """Update the status_message for *user_id* in the database.""" from app.models.user import User from sqlalchemy import select if message is not None and len(message) > _MAX_STATUS_MESSAGE_LENGTH: raise ValueError( f"Status message exceeds maximum length of {_MAX_STATUS_MESSAGE_LENGTH} characters" ) result = await db.execute(select(User).where(User.id == user_id)) user = result.scalar_one_or_none() if user is None: raise ValueError(f"User {user_id} not found") user.status_message = message user.last_seen_at = datetime.now(timezone.utc) await db.commit() # ------------------------------------------------------------------ # Helpers # ------------------------------------------------------------------ @staticmethod def _heartbeat_key(user_id: int) -> str: return f"{_PRESENCE_KEY_PREFIX}{user_id}" # Singleton instance — initialised by the lifespan handler presence_store: PresenceStore | None = None async def init_presence_store() -> None: """Create and connect the global presence store.""" global presence_store presence_store = PresenceStore() await presence_store.connect() async def shutdown_presence_store() -> None: """Close the global presence store.""" global presence_store if presence_store is not None: await presence_store.close() presence_store = None