modbison/backend/app/services/presence.py

"""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)

    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()
        keys = [self._heartbeat_key(uid) for uid in user_ids]
        values = await self.client.mget(keys)
        return {uid for uid, v in zip(user_ids, values) if v is not None}

    # ------------------------------------------------------------------
    # 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