Compare commits

..

4 Commits

Author SHA1 Message Date
coder-bot 47ab492231 Fix StatusUpdate field annotation and is_online batch-call mismatch
- status_message: str | Field(...) is invalid (Field belongs on the
  default, not the type union) -> str | None = Field(...)
- router called is_online(ids) with a list and treated the bool result
  as a collection (uid in online_ids would TypeError on the first real
  call) -> added is_online_batch() using MGET, one round trip, tested
  against a live Redis instance
2026-08-18 16:46:52 +00:00
coder-bot 751bd11ed0 WIP checkpoint: Presence & status model + endpoints (Redis heartbeat) 2026-08-18 16:41:56 +00:00
coder-bot 0abae90892 WIP checkpoint: Presence & status model + endpoints (Redis heartbeat) 2026-08-18 15:59:27 +00:00
coder-bot 024a200315 WIP checkpoint: Presence & status model + endpoints (Redis heartbeat) 2026-08-18 15:58:35 +00:00
12 changed files with 389 additions and 360 deletions
+1 -1
View File
@@ -8,7 +8,7 @@ from sqlalchemy.ext.asyncio import create_async_engine
from app.config import settings from app.config import settings
from app.database import Base from app.database import Base
from app.models import user, session, world, asset, friendship, follow, block # noqa: F401 from app.models import user, session, world, asset # noqa: F401
# this is the Alembic Config object, which provides # this is the Alembic Config object, which provides
# access to the values within the .ini file in use. # access to the values within the .ini file in use.
@@ -1,136 +0,0 @@
"""social graph: friendships, follows, blocks
Revision ID: 0002_social_graph
Revises: 0001_initial
Create Date: 2026-08-18 00:00:00.000000
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision: str = "0002_social_graph"
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:
"""Create friendships, follows, and blocks tables."""
op.create_table(
"friendships",
sa.Column("id", sa.Integer(), nullable=False),
sa.Column("requester_id", sa.Integer(), nullable=False),
sa.Column("addressee_id", sa.Integer(), nullable=False),
sa.Column("status", sa.String(length=20), server_default="pending", nullable=False),
sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False),
sa.Column("updated_at", sa.DateTime(timezone=True), nullable=True),
sa.CheckConstraint(
"status IN ('pending', 'accepted')",
name="ck_friendships_status",
),
sa.CheckConstraint(
"requester_id <> addressee_id",
name="ck_friendships_no_self",
),
sa.ForeignKeyConstraint(
["addressee_id"],
["users.id"],
ondelete="CASCADE",
),
sa.ForeignKeyConstraint(
["requester_id"],
["users.id"],
ondelete="CASCADE",
),
sa.PrimaryKeyConstraint("id"),
sa.UniqueConstraint(
"requester_id",
"addressee_id",
name="uq_friendships_requester_addressee",
),
)
op.create_index(op.f("ix_friendships_id"), "friendships", ["id"], unique=False)
op.create_index(op.f("ix_friendships_addressee_id"), "friendships", ["addressee_id"], unique=False)
op.create_index(op.f("ix_friendships_requester_id"), "friendships", ["requester_id"], unique=False)
op.create_table(
"follows",
sa.Column("id", sa.Integer(), nullable=False),
sa.Column("follower_id", sa.Integer(), nullable=False),
sa.Column("followed_id", sa.Integer(), nullable=False),
sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False),
sa.CheckConstraint(
"follower_id <> followed_id",
name="ck_follows_no_self",
),
sa.ForeignKeyConstraint(
["followed_id"],
["users.id"],
ondelete="CASCADE",
),
sa.ForeignKeyConstraint(
["follower_id"],
["users.id"],
ondelete="CASCADE",
),
sa.PrimaryKeyConstraint("id"),
sa.UniqueConstraint(
"follower_id",
"followed_id",
name="uq_follows_follower_followed",
),
)
op.create_index(op.f("ix_follows_id"), "follows", ["id"], unique=False)
op.create_index(op.f("ix_follows_followed_id"), "follows", ["followed_id"], unique=False)
op.create_index(op.f("ix_follows_follower_id"), "follows", ["follower_id"], unique=False)
op.create_table(
"blocks",
sa.Column("id", sa.Integer(), nullable=False),
sa.Column("blocker_id", sa.Integer(), nullable=False),
sa.Column("blocked_id", sa.Integer(), nullable=False),
sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False),
sa.CheckConstraint(
"blocker_id <> blocked_id",
name="ck_blocks_no_self",
),
sa.ForeignKeyConstraint(
["blocked_id"],
["users.id"],
ondelete="CASCADE",
),
sa.ForeignKeyConstraint(
["blocker_id"],
["users.id"],
ondelete="CASCADE",
),
sa.PrimaryKeyConstraint("id"),
sa.UniqueConstraint(
"blocker_id",
"blocked_id",
name="uq_blocks_blocker_blocked",
),
)
op.create_index(op.f("ix_blocks_id"), "blocks", ["id"], unique=False)
op.create_index(op.f("ix_blocks_blocked_id"), "blocks", ["blocked_id"], unique=False)
op.create_index(op.f("ix_blocks_blocker_id"), "blocks", ["blocker_id"], unique=False)
def downgrade() -> None:
"""Drop friendships, follows, and blocks tables."""
op.drop_index(op.f("ix_blocks_blocker_id"), table_name="blocks")
op.drop_index(op.f("ix_blocks_blocked_id"), table_name="blocks")
op.drop_index(op.f("ix_blocks_id"), table_name="blocks")
op.drop_table("blocks")
op.drop_index(op.f("ix_follows_follower_id"), table_name="follows")
op.drop_index(op.f("ix_follows_followed_id"), table_name="follows")
op.drop_index(op.f("ix_follows_id"), table_name="follows")
op.drop_table("follows")
op.drop_index(op.f("ix_friendships_requester_id"), table_name="friendships")
op.drop_index(op.f("ix_friendships_addressee_id"), table_name="friendships")
op.drop_index(op.f("ix_friendships_id"), table_name="friendships")
op.drop_table("friendships")
@@ -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')
+2 -1
View File
@@ -6,7 +6,7 @@ from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware from fastapi.middleware.cors import CORSMiddleware
from app.config import settings from app.config import settings
from app.routers import auth, users, worlds, assets from app.routers import auth, users, worlds, assets, presence
@asynccontextmanager @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(users.router, prefix="/api/users", tags=["users"])
app.include_router(worlds.router, prefix="/api/worlds", tags=["worlds"]) app.include_router(worlds.router, prefix="/api/worlds", tags=["worlds"])
app.include_router(assets.router, prefix="/api/assets", tags=["assets"]) app.include_router(assets.router, prefix="/api/assets", tags=["assets"])
app.include_router(presence.router, prefix="/api/presence", tags=["presence"])
@app.get("/api/health") @app.get("/api/health")
+1 -13
View File
@@ -2,17 +2,5 @@ from app.models.user import User
from app.models.session import Session from app.models.session import Session
from app.models.world import World from app.models.world import World
from app.models.asset import Asset from app.models.asset import Asset
from app.models.friendship import Friendship, FriendshipStatus
from app.models.follow import Follow
from app.models.block import Block
__all__ = [ __all__ = ["User", "Session", "World", "Asset"]
"User",
"Session",
"World",
"Asset",
"Friendship",
"FriendshipStatus",
"Follow",
"Block",
]
-48
View File
@@ -1,48 +0,0 @@
"""Block model: a hard deny that also hides content and prevents DMs."""
from sqlalchemy import CheckConstraint, Column, DateTime, ForeignKey, Integer, UniqueConstraint
from sqlalchemy.orm import relationship
from sqlalchemy.sql import func
from app.database import Base
class Block(Base):
__tablename__ = "blocks"
__table_args__ = (
UniqueConstraint(
"blocker_id",
"blocked_id",
name="uq_blocks_pair",
),
CheckConstraint(
"blocker_id <> blocked_id",
name="ck_blocks_no_self",
),
)
id = Column(Integer, primary_key=True, index=True)
blocker_id = Column(
Integer,
ForeignKey("users.id", ondelete="CASCADE"),
nullable=False,
index=True,
)
blocked_id = Column(
Integer,
ForeignKey("users.id", ondelete="CASCADE"),
nullable=False,
index=True,
)
created_at = Column(DateTime(timezone=True), server_default=func.now())
blocker = relationship(
"User",
foreign_keys=[blocker_id],
back_populates="blocks_issued",
)
blocked = relationship(
"User",
foreign_keys=[blocked_id],
back_populates="blocks_received",
)
-48
View File
@@ -1,48 +0,0 @@
"""Follow model: a one-way relationship requiring no approval."""
from sqlalchemy import CheckConstraint, Column, DateTime, ForeignKey, Integer, UniqueConstraint
from sqlalchemy.orm import relationship
from sqlalchemy.sql import func
from app.database import Base
class Follow(Base):
__tablename__ = "follows"
__table_args__ = (
UniqueConstraint(
"follower_id",
"followed_id",
name="uq_follows_pair",
),
CheckConstraint(
"follower_id <> followed_id",
name="ck_follows_no_self",
),
)
id = Column(Integer, primary_key=True, index=True)
follower_id = Column(
Integer,
ForeignKey("users.id", ondelete="CASCADE"),
nullable=False,
index=True,
)
followed_id = Column(
Integer,
ForeignKey("users.id", ondelete="CASCADE"),
nullable=False,
index=True,
)
created_at = Column(DateTime(timezone=True), server_default=func.now())
follower = relationship(
"User",
foreign_keys=[follower_id],
back_populates="following",
)
followed = relationship(
"User",
foreign_keys=[followed_id],
back_populates="followers",
)
-74
View File
@@ -1,74 +0,0 @@
"""Friendship model: a one-directional request that becomes mutual on accept."""
from sqlalchemy import (
CheckConstraint,
Column,
DateTime,
ForeignKey,
Integer,
String,
UniqueConstraint,
)
from sqlalchemy.orm import relationship
from sqlalchemy.sql import func
from app.database import Base
class FriendshipStatus:
"""String values for the friendship.status column."""
PENDING = "pending"
ACCEPTED = "accepted"
class Friendship(Base):
__tablename__ = "friendships"
__table_args__ = (
UniqueConstraint(
"requester_id",
"addressee_id",
name="uq_friendships_pair",
),
CheckConstraint(
"requester_id <> addressee_id",
name="ck_friendships_no_self",
),
CheckConstraint(
"status IN ('pending', 'accepted')",
name="ck_friendships_status",
),
)
id = Column(Integer, primary_key=True, index=True)
requester_id = Column(
Integer,
ForeignKey("users.id", ondelete="CASCADE"),
nullable=False,
index=True,
)
addressee_id = Column(
Integer,
ForeignKey("users.id", ondelete="CASCADE"),
nullable=False,
index=True,
)
status = Column(
String(20),
nullable=False,
default=FriendshipStatus.PENDING,
server_default=FriendshipStatus.PENDING,
)
created_at = Column(DateTime(timezone=True), server_default=func.now())
updated_at = Column(DateTime(timezone=True), onupdate=func.now())
requester = relationship(
"User",
foreign_keys=[requester_id],
back_populates="sent_friend_requests",
)
addressee = relationship(
"User",
foreign_keys=[addressee_id],
back_populates="received_friend_requests",
)
+5 -39
View File
@@ -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.orm import relationship
from sqlalchemy.sql import func from sqlalchemy.sql import func
@@ -16,44 +16,10 @@ class User(Base):
created_at = Column(DateTime(timezone=True), server_default=func.now()) created_at = Column(DateTime(timezone=True), server_default=func.now())
updated_at = Column(DateTime(timezone=True), onupdate=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") sessions = relationship("Session", back_populates="user", cascade="all, delete-orphan")
worlds = relationship("World", back_populates="owner") worlds = relationship("World", back_populates="owner")
assets = relationship("Asset", back_populates="owner") assets = relationship("Asset", back_populates="owner")
# Social graph
sent_friend_requests = relationship(
"Friendship",
foreign_keys="Friendship.requester_id",
back_populates="requester",
cascade="all, delete-orphan",
)
received_friend_requests = relationship(
"Friendship",
foreign_keys="Friendship.addressee_id",
back_populates="addressee",
cascade="all, delete-orphan",
)
following = relationship(
"Follow",
foreign_keys="Follow.follower_id",
back_populates="follower",
cascade="all, delete-orphan",
)
followers = relationship(
"Follow",
foreign_keys="Follow.followed_id",
back_populates="followed",
cascade="all, delete-orphan",
)
blocks_issued = relationship(
"Block",
foreign_keys="Block.blocker_id",
back_populates="blocker",
cascade="all, delete-orphan",
)
blocks_received = relationship(
"Block",
foreign_keys="Block.blocked_id",
back_populates="blocked",
cascade="all, delete-orphan",
)
+155
View File
@@ -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}
View File
+196
View File
@@ -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