diff --git a/backend/alembic/versions/2a9b9873244a_add_presence_columns.py b/backend/alembic/versions/2a9b9873244a_add_presence_columns.py new file mode 100644 index 0000000..5926ed4 --- /dev/null +++ b/backend/alembic/versions/2a9b9873244a_add_presence_columns.py @@ -0,0 +1,29 @@ +"""add presence columns to users + +Revision ID: 2a9b9873244a +Revises: 0001_initial +Create Date: 2026-08-18 15:59:09.298919 +""" +from typing import Sequence, Union + +from alembic import op +import sqlalchemy as sa + + +# revision identifiers, used by Alembic. +revision: str = '2a9b9873244a' +down_revision: Union[str, None] = '0001_initial' +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 1260c12..db15dc9 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 +from app.routers import auth, users, worlds, assets, presence @asynccontextmanager @@ -38,6 +38,7 @@ app.include_router(auth.router, prefix="/api/auth", tags=["auth"]) app.include_router(users.router, prefix="/api/users", tags=["users"]) app.include_router(worlds.router, prefix="/api/worlds", tags=["worlds"]) app.include_router(assets.router, prefix="/api/assets", tags=["assets"]) +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 ff48caa..1b752f7 100644 --- a/backend/app/models/user.py +++ b/backend/app/models/user.py @@ -1,4 +1,4 @@ -from sqlalchemy import Column, Integer, String, Boolean, DateTime +from sqlalchemy import Column, Integer, String, Boolean, DateTime, Text from sqlalchemy.orm import relationship from sqlalchemy.sql import func @@ -16,6 +16,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 + 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") worlds = relationship("World", back_populates="owner") assets = relationship("Asset", back_populates="owner") diff --git a/backend/app/routers/presence.py b/backend/app/routers/presence.py new file mode 100644 index 0000000..769c669 --- /dev/null +++ b/backend/app/routers/presence.py @@ -0,0 +1,155 @@ +"""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, Depends, HTTPException, status +from pydantic import BaseModel, Field, field_validator +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession + +from app.config import settings +from app.database import get_db +from app.dependencies import CurrentUser +from app.models.user import User +from app.services.presence import PresenceStore + +router = APIRouter() + +# ──────────────────────────── 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 ─────────────────────────────────── + +# The presence store is created once (singleton) so it shares the same Redis +# connection pool across all requests. +_presence_store = PresenceStore() + + +@router.post("/heartbeat", response_model=dict) +async def heartbeat( + _req: HeartbeatRequest, + current_user: CurrentUser, + db: Annotated[AsyncSession, Depends(get_db)], +): + """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: Annotated[AsyncSession, Depends(get_db)], +): + """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 + 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: Annotated[AsyncSession, Depends(get_db)], +): + """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/__init__.py b/backend/app/services/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/backend/app/services/presence.py b/backend/app/services/presence.py new file mode 100644 index 0000000..2faa2dc --- /dev/null +++ b/backend/app/services/presence.py @@ -0,0 +1,196 @@ +"""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