Compare commits

..

4 Commits

Author SHA1 Message Date
daddy 9661b9a81c Add social graph models: friendships, follows, blocks
Models + Alembic migration (0007, head) only, per docs/planning/social.md
US-1 -- the friends/follow API router is a separate sibling task.

- friendships: requester_id/addressee_id (FK users, CASCADE), status
  (pending/accepted native enum), unique pair constraint, indexed FKs.
- follows: follower_id/followed_id (FK users, CASCADE), unique pair.
- blocks: blocker_id/blocked_id (FK users, CASCADE), unique pair.
- Relationships wired onto User for both directions of each table.
- tests/test_social_models.py exercises request/accept, the unique-pair
  constraint, follow, block, and DB-level CASCADE delete against real
  Postgres. Full backend suite (152 tests) still green.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Y2FESCaMXeJpC9gSctiEeD
2026-09-15 00:10:15 +00:00
daddy 4eb4476fc3 Presence & status model + endpoints (Redis heartbeat) -- recovered
checkpoint / checkpoint (push) Successful in 2m25s
Real feature built in August, checkpointed mid-task by the old fleet's
graceful-shutdown mechanism, then orphaned when checkpoint_focus_
feature_ids scoped the fleet away from M8 and it was never resumed --
the PM task had shown done/reviewed for a month with zero actual code on
main. Ported forward from the 225-commits-stale branch onto current
main's conventions (DbDep pattern, migration rebased onto 0005's head),
fixing two real bugs found in the process: the original PresenceStore's
explicit connect() was never called from anywhere (would 500 on every
real request), and a naive fix hits 'event loop is closed' under this
repo's TestClient(app) pattern -- fixed with lazy per-event-loop client
caching, same class of issue already documented for asyncpg in
conftest.py. conftest.py extended to also bring up ephemeral Redis
symmetrically with its existing Postgres bootstrap.

144/144 backend tests independently re-run via the fully automated
hermetic path (zero manual setup, matching what a fresh contributor
gets), plus a manual end-to-end heartbeat/TTL-expiry/batch-query check
against real Postgres+Redis.
2026-09-15 00:01:24 +00:00
daddy f34a4806ef Port presence feature forward from orphaned agent/dd2a126b57a1 branch
Recovers a real August feature that was never merged (3 WIP checkpoints +
fix commit 47ab492, 225 commits behind main): a Redis-backed heartbeat
presence store plus /api/presence/{heartbeat,presence,status} endpoints.
Ported onto current main's conventions rather than rebuilt:

- app/models/user.py: last_seen_at/status_message columns (unchanged from
  the orphaned branch; still compatible with the current model).
- alembic/versions/0006_add_presence_columns.py: same column migration,
  rebased as the new head after 0005_add_world_content_tables (was
  2a9b9873244a off 0001_initial).
- app/routers/presence.py: same three endpoints, updated to the DbDep
  pattern other routers (avatar.py) now use; retains the 47ab492 fix
  (is_online_batch via MGET, correct StatusUpdate Field annotation).
- app/services/presence.py: real adaptation beyond a mechanical port --
  the original PresenceStore required an explicit connect() that nothing
  ever called (main.py's lifespan never wired it up), so every real
  request would have raised "not connected". Made the Redis client connect
  lazily on first use instead, and cache it per running event loop rather
  than globally: this repo's TestClient(app) pattern (no `with` block) can
  dispatch different requests onto different event loops, which broke a
  single cached connection the same way NullPool's comment in conftest.py
  describes for asyncpg. Dropped the unused get_presence()/
  update_status_message() methods and the never-wired init/shutdown
  singleton helpers -- dead code even in the original.
- tests/conftest.py: extended the existing ephemeral-Postgres bootstrap to
  also bring up ephemeral Redis (docker-compose.yml already defined the
  service) when REDIS_URL isn't already reachable, so `pytest` stays
  hermetic.
- tests/test_presence.py: heartbeat sets online, batch presence over
  mixed online/offline/unknown ids (the exact 47ab492 regression), status
  message update/strip/length-limit, and TTL expiry via a short-TTL
  PresenceStore instance (real Redis SET/EXISTS/MGET, no 60s sleep needed).

Verified: full backend pytest suite (144 tests) passes against a fresh
Postgres+Redis. Also drove the real endpoints end-to-end against a running
uvicorn + docker-compose Postgres/Redis: heartbeat -> online, batch query,
forced TTL expiry -> offline with last_seen_at retained, status message
trim + 422 over 140 chars.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Y2FESCaMXeJpC9gSctiEeD
2026-09-14 23:59:14 +00:00
daddy 9f04cef447 Editor: save world edits to the server; re-enter and see the changes persist
checkpoint / checkpoint (push) Successful in 2m18s
ChunkSaveService (single PUT + base_revision, 409 -> refetch -> retry,
atomic batch PUT), wired into InGameEditor (Save/Ctrl+S, live_session
surfaced as a warning, not ignored). Owner-gated 'Edit World' entry from
the world browser that sets Session.editing without ever calling
join_world() -- editing is non-joined per world_persistence_impl.md §6.
game.gd fetches the real chunk over HTTP when entering edit mode.

19/19 new save-path smoke assertions (incl. a real two-writer 409
conflict resolved transparently, and reload confirming persistence) and
the full existing suite (18 Godot smoke tests + 132 backend pytest)
independently re-verified with zero regressions.
2026-09-14 23:38:46 +00:00
11 changed files with 1132 additions and 39 deletions
@@ -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")
@@ -0,0 +1,118 @@
"""add social graph tables (friendships, follows, blocks)
Revision ID: 0007_add_social_graph_tables
Revises: 0006_add_presence_columns
Create Date: 2026-09-15 00:00:00.000000
docs/planning/social.md US-1 / app/models/social.py. Models + migration
only -- the request/accept/block business logic lives in the sibling
"Friends & follow API router" task.
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
from sqlalchemy.dialects import postgresql
# revision identifiers, used by Alembic.
revision: str = "0007_add_social_graph_tables"
down_revision: Union[str, None] = "0006_add_presence_columns"
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
friendship_status = postgresql.ENUM(
"pending", "accepted", name="friendship_status"
)
friendship_status.create(op.get_bind(), checkfirst=True)
op.create_table(
"friendships",
sa.Column("id", sa.Integer(), primary_key=True),
sa.Column(
"requester_id",
sa.Integer(),
sa.ForeignKey("users.id", ondelete="CASCADE"),
nullable=False,
),
sa.Column(
"addressee_id",
sa.Integer(),
sa.ForeignKey("users.id", ondelete="CASCADE"),
nullable=False,
),
sa.Column(
"status",
postgresql.ENUM(
"pending", "accepted", name="friendship_status", create_type=False
),
nullable=False,
server_default="pending",
),
sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False),
sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False),
sa.UniqueConstraint("requester_id", "addressee_id", name="uq_friendships_pair"),
)
op.create_index("ix_friendships_requester_id", "friendships", ["requester_id"])
op.create_index("ix_friendships_addressee_id", "friendships", ["addressee_id"])
op.create_table(
"follows",
sa.Column("id", sa.Integer(), primary_key=True),
sa.Column(
"follower_id",
sa.Integer(),
sa.ForeignKey("users.id", ondelete="CASCADE"),
nullable=False,
),
sa.Column(
"followed_id",
sa.Integer(),
sa.ForeignKey("users.id", ondelete="CASCADE"),
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), server_default=sa.func.now(), nullable=False),
sa.UniqueConstraint("follower_id", "followed_id", name="uq_follows_pair"),
)
op.create_index("ix_follows_follower_id", "follows", ["follower_id"])
op.create_index("ix_follows_followed_id", "follows", ["followed_id"])
op.create_table(
"blocks",
sa.Column("id", sa.Integer(), primary_key=True),
sa.Column(
"blocker_id",
sa.Integer(),
sa.ForeignKey("users.id", ondelete="CASCADE"),
nullable=False,
),
sa.Column(
"blocked_id",
sa.Integer(),
sa.ForeignKey("users.id", ondelete="CASCADE"),
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), server_default=sa.func.now(), nullable=False),
sa.UniqueConstraint("blocker_id", "blocked_id", name="uq_blocks_pair"),
)
op.create_index("ix_blocks_blocker_id", "blocks", ["blocker_id"])
op.create_index("ix_blocks_blocked_id", "blocks", ["blocked_id"])
def downgrade() -> None:
op.drop_index("ix_blocks_blocked_id", table_name="blocks")
op.drop_index("ix_blocks_blocker_id", table_name="blocks")
op.drop_table("blocks")
op.drop_index("ix_follows_followed_id", table_name="follows")
op.drop_index("ix_follows_follower_id", table_name="follows")
op.drop_table("follows")
op.drop_index("ix_friendships_addressee_id", table_name="friendships")
op.drop_index("ix_friendships_requester_id", table_name="friendships")
op.drop_table("friendships")
postgresql.ENUM(name="friendship_status").drop(op.get_bind(), checkfirst=True)
+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, avatar, world_content, internal from app.routers import auth, users, worlds, assets, avatar, world_content, internal, presence
from app.services.game_servers import supervisor 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(assets.router, prefix="/api/assets", tags=["assets"])
app.include_router(avatar.router, prefix="/api/avatar", tags=["avatar"]) app.include_router(avatar.router, prefix="/api/avatar", tags=["avatar"])
app.include_router(internal.router, prefix="/api/internal", tags=["internal"]) app.include_router(internal.router, prefix="/api/internal", tags=["internal"])
app.include_router(presence.router, prefix="/api/presence", tags=["presence"])
@app.get("/api/health") @app.get("/api/health")
+2
View File
@@ -4,8 +4,10 @@ from app.models.world import World
from app.models.asset import Asset from app.models.asset import Asset
from app.models.avatar_loadout import AvatarLoadout from app.models.avatar_loadout import AvatarLoadout
from app.models.world_content import WorldRoom, WorldChunk, WorldChunkRevision from app.models.world_content import WorldRoom, WorldChunk, WorldChunkRevision
from app.models.social import Friendship, Follow, Block, FriendshipStatus
__all__ = [ __all__ = [
"User", "Session", "World", "Asset", "AvatarLoadout", "User", "Session", "World", "Asset", "AvatarLoadout",
"WorldRoom", "WorldChunk", "WorldChunkRevision", "WorldRoom", "WorldChunk", "WorldChunkRevision",
"Friendship", "Follow", "Block", "FriendshipStatus",
] ]
+106
View File
@@ -0,0 +1,106 @@
"""Social graph: friendships, follows, blocks.
docs/planning/social.md §4/§5 (US-1). Friends = mutual (request -> accept),
Follow = one-way (no approval), Block = hard deny. This module is models +
schema only -- the request/accept/block business logic (idempotent
duplicate/self handling, removing an existing friendship/follow on block,
etc.) belongs to the API router task that depends on this one.
"""
import enum
from sqlalchemy import Column, Integer, ForeignKey, DateTime, UniqueConstraint
from sqlalchemy.dialects.postgresql import ENUM
from sqlalchemy.orm import relationship
from sqlalchemy.sql import func
from app.database import Base
class FriendshipStatus(str, enum.Enum):
PENDING = "pending"
ACCEPTED = "accepted"
class Friendship(Base):
"""A friend request, directional until accepted.
``requester_id`` sent the request to ``addressee_id``. Once accepted the
friendship is mutual, but the row stays directional (who asked whom) for
audit purposes; the API layer treats it as symmetric once accepted.
"""
__tablename__ = "friendships"
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(
ENUM(
FriendshipStatus,
name="friendship_status",
create_type=False,
values_callable=lambda enum_cls: [e.value for e in enum_cls],
),
nullable=False,
default=FriendshipStatus.PENDING,
server_default=FriendshipStatus.PENDING.value,
)
created_at = Column(DateTime(timezone=True), server_default=func.now(), nullable=False)
updated_at = Column(DateTime(timezone=True), server_default=func.now(), onupdate=func.now(), nullable=False)
requester = relationship("User", foreign_keys=[requester_id], back_populates="sent_friend_requests")
addressee = relationship("User", foreign_keys=[addressee_id], back_populates="received_friend_requests")
__table_args__ = (
UniqueConstraint("requester_id", "addressee_id", name="uq_friendships_pair"),
)
class Follow(Base):
"""A one-way follow relationship: ``follower_id`` follows ``followed_id``."""
__tablename__ = "follows"
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(), nullable=False)
updated_at = Column(DateTime(timezone=True), server_default=func.now(), onupdate=func.now(), nullable=False)
follower = relationship("User", foreign_keys=[follower_id], back_populates="following")
followed = relationship("User", foreign_keys=[followed_id], back_populates="followers")
__table_args__ = (
UniqueConstraint("follower_id", "followed_id", name="uq_follows_pair"),
)
class Block(Base):
"""A hard deny: ``blocker_id`` has blocked ``blocked_id``."""
__tablename__ = "blocks"
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(), nullable=False)
updated_at = Column(DateTime(timezone=True), server_default=func.now(), onupdate=func.now(), nullable=False)
blocker = relationship("User", foreign_keys=[blocker_id], back_populates="blocks_initiated")
blocked = relationship("User", foreign_keys=[blocked_id], back_populates="blocked_by")
__table_args__ = (
UniqueConstraint("blocker_id", "blocked_id", name="uq_blocks_pair"),
)
+43 -1
View File
@@ -1,7 +1,7 @@
from datetime import date from datetime import date
from typing import Optional 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.orm import relationship
from sqlalchemy.sql import func from sqlalchemy.sql import func
@@ -20,6 +20,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 (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") sessions = relationship("Session", back_populates="user", cascade="all, delete-orphan")
refresh_tokens = relationship("RefreshToken", back_populates="user", cascade="all, delete-orphan") refresh_tokens = relationship("RefreshToken", back_populates="user", cascade="all, delete-orphan")
worlds = relationship("World", back_populates="owner") worlds = relationship("World", back_populates="owner")
@@ -31,6 +35,44 @@ class User(Base):
cascade="all, delete-orphan", cascade="all, delete-orphan",
) )
# Social graph (docs/planning/social.md US-1; app/models/social.py)
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_initiated = relationship(
"Block",
foreign_keys="Block.blocker_id",
back_populates="blocker",
cascade="all, delete-orphan",
)
blocked_by = relationship(
"Block",
foreign_keys="Block.blocked_id",
back_populates="blocked",
cascade="all, delete-orphan",
)
@property @property
def is_adult(self) -> bool: def is_adult(self) -> bool:
"""Return True if the user is 18 or older based on date_of_birth.""" """Return True if the user is 18 or older based on date_of_birth."""
+146
View File
@@ -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}
+145
View File
@@ -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()
+76 -37
View File
@@ -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, The integration tests hit a real Postgres (Postgres-specific types: JSONB,
native ENUMs). To make ``cd backend && pytest`` work from a clean checkout native ENUMs) and, for presence tests, a real Redis (TTL semantics aren't
with no manual setup, ``pytest_configure`` below brings up an ephemeral worth faking). To make ``cd backend && pytest`` work from a clean checkout
Postgres via docker compose and runs the migrations IF ``DATABASE_URL`` is with no manual setup, ``pytest_configure`` below brings up ephemeral
not already pointing at a reachable database. When it is (dev loop, or the Postgres/Redis containers via docker compose and runs the migrations IF
checkpoint runner which stands the DB up itself), that path is skipped and ``DATABASE_URL``/``REDIS_URL`` aren't already pointing at something
the existing database is reused. 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 If Docker isn't available and nothing is reachable, the whole
is skipped with a clear message rather than erroring out 60+ times. DB/Redis-backed suite is skipped with a clear message rather than erroring
out dozens of times.
""" """
import asyncio import asyncio
@@ -38,14 +41,19 @@ TEST_TABLES = (
# `worlds` the way world_rooms/world_chunks would be. List all three # `worlds` the way world_rooms/world_chunks would be. List all three
# explicitly rather than rely on cascade for some of them. # explicitly rather than rely on cascade for some of them.
"world_chunk_revisions", "world_chunks", "world_rooms", "world_chunk_revisions", "world_chunks", "world_rooms",
"worlds", "sessions", "refresh_tokens", "users", "worlds",
# Social graph (app/models/social.py) -- all three FK users with
# ON DELETE CASCADE, but list explicitly for clarity/consistency.
"friendships", "follows", "blocks",
"sessions", "refresh_tokens", "users",
) )
BACKEND_DIR = Path(__file__).resolve().parents[1] BACKEND_DIR = Path(__file__).resolve().parents[1]
_COMPOSE_PROJECT = "agp-pytest" _COMPOSE_PROJECT = "agp-pytest"
_EPHEMERAL_PG_PORT = 25432 _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. # pytest_unconfigure.
_started_container = False _started_container = False
_db_unavailable_reason: str | None = None _db_unavailable_reason: str | None = None
@@ -64,6 +72,11 @@ def _db_reachable(url: str) -> bool:
return _port_open(parts.hostname or "localhost", parts.port or 5432) 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: def _run(cmd: list[str], **kw) -> subprocess.CompletedProcess:
return subprocess.run(cmd, cwd=BACKEND_DIR, capture_output=True, text=True, **kw) return subprocess.run(cmd, cwd=BACKEND_DIR, capture_output=True, text=True, **kw)
@@ -71,14 +84,17 @@ def _run(cmd: list[str], **kw) -> subprocess.CompletedProcess:
def pytest_configure(config): def pytest_configure(config):
global _started_container, _db_unavailable_reason global _started_container, _db_unavailable_reason
if _db_reachable(settings.DATABASE_URL): need_db = not _db_reachable(settings.DATABASE_URL)
return # dev loop / checkpoint runner already provided one 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: if not _run(["docker", "--version"]).returncode == 0:
_db_unavailable_reason = ( _db_unavailable_reason = (
"no reachable DATABASE_URL and docker is unavailable -- run the " "no reachable DATABASE_URL/REDIS_URL and docker is unavailable -- run "
"checkpoint suite (scripts/run_checkpoint_tests.py) or start Postgres " "the checkpoint suite (scripts/run_checkpoint_tests.py) or start "
"manually" "Postgres/Redis manually"
) )
return return
@@ -86,33 +102,55 @@ def pytest_configure(config):
env["COMPOSE_PROJECT_NAME"] = _COMPOSE_PROJECT env["COMPOSE_PROJECT_NAME"] = _COMPOSE_PROJECT
env["POSTGRES_HOST_PORT"] = str(_EPHEMERAL_PG_PORT) env["POSTGRES_HOST_PORT"] = str(_EPHEMERAL_PG_PORT)
env["POSTGRES_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: 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 return
_started_container = True _started_container = True
for _ in range(30): if need_db:
r = _run( for _ in range(30):
["docker", "compose", "exec", "-T", "postgres", r = _run(
"pg_isready", "-U", "agp_user", "-d", "agp_db"], ["docker", "compose", "exec", "-T", "postgres",
env=env, "pg_isready", "-U", "agp_user", "-d", "agp_db"],
) env=env,
if r.returncode == 0: )
break if r.returncode == 0:
time.sleep(1) break
else: time.sleep(1)
_db_unavailable_reason = "ephemeral Postgres never became ready" else:
return _db_unavailable_reason = "ephemeral Postgres never became ready"
return
url = f"postgresql+asyncpg://agp_user:agp_password@localhost:{_EPHEMERAL_PG_PORT}/agp_db" url = f"postgresql+asyncpg://agp_user:agp_password@localhost:{_EPHEMERAL_PG_PORT}/agp_db"
os.environ["DATABASE_URL"] = url os.environ["DATABASE_URL"] = url
settings.DATABASE_URL = url settings.DATABASE_URL = url
mig = _run([sys.executable, "-m", "alembic", "upgrade", "head"], mig = _run([sys.executable, "-m", "alembic", "upgrade", "head"],
env={**os.environ, "DATABASE_URL": url}) env={**os.environ, "DATABASE_URL": url})
if mig.returncode != 0: if mig.returncode != 0:
_db_unavailable_reason = f"alembic upgrade head failed:\n{mig.stderr[-800:]}" _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): def pytest_unconfigure(config):
@@ -122,6 +160,7 @@ def pytest_unconfigure(config):
env["COMPOSE_PROJECT_NAME"] = _COMPOSE_PROJECT env["COMPOSE_PROJECT_NAME"] = _COMPOSE_PROJECT
env["POSTGRES_HOST_PORT"] = str(_EPHEMERAL_PG_PORT) env["POSTGRES_HOST_PORT"] = str(_EPHEMERAL_PG_PORT)
env["POSTGRES_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) _run(["docker", "compose", "down", "-v"], env=env)
+221
View File
@@ -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()
+240
View File
@@ -0,0 +1,240 @@
"""Model-level tests for the social graph: friendships, follows, blocks.
app/models/social.py / alembic/versions/0007_add_social_graph_tables.py
(docs/planning/social.md US-1). This is the models+migration task only --
there are no API endpoints yet (a sibling task owns the router), so these
tests exercise the ORM models and the real Postgres constraints directly
rather than going through HTTP.
"""
import pytest
from fastapi.testclient import TestClient
from sqlalchemy import delete, select
from sqlalchemy.exc import IntegrityError
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
from sqlalchemy.pool import NullPool
from app.config import settings
from app.main import app
from app.models.social import Friendship, Follow, Block, FriendshipStatus
from app.models.user import User
def _register(client: TestClient, username: str) -> int:
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
headers = {"Authorization": f"Bearer {resp.json()['access_token']}"}
me = client.get("/api/users/me", headers=headers)
assert me.status_code == 200, me.text
return me.json()["id"]
def _session_factory():
engine = create_async_engine(settings.DATABASE_URL, poolclass=NullPool)
return engine, async_sessionmaker(engine, expire_on_commit=False)
class TestFriendshipRequestAccept:
async def test_request_defaults_to_pending(self):
client = TestClient(app)
a = _register(client, "sg_fr_a1")
b = _register(client, "sg_fr_b1")
engine, factory = _session_factory()
try:
async with factory() as session:
fr = Friendship(requester_id=a, addressee_id=b)
session.add(fr)
await session.commit()
await session.refresh(fr)
assert fr.status == FriendshipStatus.PENDING
assert fr.requester_id == a
assert fr.addressee_id == b
finally:
await engine.dispose()
async def test_accept_updates_status(self):
client = TestClient(app)
a = _register(client, "sg_fr_a2")
b = _register(client, "sg_fr_b2")
engine, factory = _session_factory()
try:
async with factory() as session:
fr = Friendship(requester_id=a, addressee_id=b)
session.add(fr)
await session.commit()
fr_id = fr.id
async with factory() as session:
fr = await session.get(Friendship, fr_id)
fr.status = FriendshipStatus.ACCEPTED
await session.commit()
async with factory() as session:
fr = await session.get(Friendship, fr_id)
assert fr.status == FriendshipStatus.ACCEPTED
finally:
await engine.dispose()
class TestUniquePairConstraints:
async def test_duplicate_friendship_pair_rejected(self):
client = TestClient(app)
a = _register(client, "sg_uq_fr_a")
b = _register(client, "sg_uq_fr_b")
engine, factory = _session_factory()
try:
async with factory() as session:
session.add(Friendship(requester_id=a, addressee_id=b))
await session.commit()
async with factory() as session:
session.add(Friendship(requester_id=a, addressee_id=b))
with pytest.raises(IntegrityError):
await session.commit()
finally:
await engine.dispose()
async def test_duplicate_follow_pair_rejected(self):
client = TestClient(app)
a = _register(client, "sg_uq_fw_a")
b = _register(client, "sg_uq_fw_b")
engine, factory = _session_factory()
try:
async with factory() as session:
session.add(Follow(follower_id=a, followed_id=b))
await session.commit()
async with factory() as session:
session.add(Follow(follower_id=a, followed_id=b))
with pytest.raises(IntegrityError):
await session.commit()
finally:
await engine.dispose()
async def test_duplicate_block_pair_rejected(self):
client = TestClient(app)
a = _register(client, "sg_uq_bl_a")
b = _register(client, "sg_uq_bl_b")
engine, factory = _session_factory()
try:
async with factory() as session:
session.add(Block(blocker_id=a, blocked_id=b))
await session.commit()
async with factory() as session:
session.add(Block(blocker_id=a, blocked_id=b))
with pytest.raises(IntegrityError):
await session.commit()
finally:
await engine.dispose()
class TestFollow:
async def test_create_follow(self):
client = TestClient(app)
a = _register(client, "sg_follow_a")
b = _register(client, "sg_follow_b")
engine, factory = _session_factory()
try:
async with factory() as session:
session.add(Follow(follower_id=a, followed_id=b))
await session.commit()
async with factory() as session:
rows = (await session.execute(
select(Follow).where(Follow.follower_id == a, Follow.followed_id == b)
)).scalars().all()
assert len(rows) == 1
finally:
await engine.dispose()
class TestBlock:
async def test_create_block(self):
client = TestClient(app)
a = _register(client, "sg_block_a")
b = _register(client, "sg_block_b")
engine, factory = _session_factory()
try:
async with factory() as session:
session.add(Block(blocker_id=a, blocked_id=b))
await session.commit()
async with factory() as session:
rows = (await session.execute(
select(Block).where(Block.blocker_id == a, Block.blocked_id == b)
)).scalars().all()
assert len(rows) == 1
finally:
await engine.dispose()
class TestCascadeDelete:
async def test_deleting_user_removes_rows_in_all_three_tables(self):
"""Deleting a user removes friendships/follows/blocks referencing them
in *either* FK column, via the real DB ON DELETE CASCADE (deleted with
a Core statement, bypassing any ORM-level cascade, to prove the
database constraint itself does the job)."""
client = TestClient(app)
a = _register(client, "sg_cascade_a")
b = _register(client, "sg_cascade_b")
c = _register(client, "sg_cascade_c")
engine, factory = _session_factory()
try:
async with factory() as session:
# a as requester/follower/blocker, and also as addressee (c -> a)
session.add(Friendship(requester_id=a, addressee_id=b))
session.add(Friendship(requester_id=c, addressee_id=a))
session.add(Follow(follower_id=a, followed_id=b))
session.add(Follow(follower_id=c, followed_id=a))
session.add(Block(blocker_id=a, blocked_id=b))
session.add(Block(blocker_id=c, blocked_id=a))
await session.commit()
async with factory() as session:
await session.execute(delete(User).where(User.id == a))
await session.commit()
async with factory() as session:
friendships = (await session.execute(
select(Friendship).where(
(Friendship.requester_id == a) | (Friendship.addressee_id == a)
)
)).scalars().all()
follows = (await session.execute(
select(Follow).where(
(Follow.follower_id == a) | (Follow.followed_id == a)
)
)).scalars().all()
blocks = (await session.execute(
select(Block).where(
(Block.blocker_id == a) | (Block.blocked_id == a)
)
)).scalars().all()
assert friendships == []
assert follows == []
assert blocks == []
# Unrelated rows (b, c only) survive.
remaining_users = (await session.execute(
select(User.id).where(User.id.in_([b, c]))
)).scalars().all()
assert set(remaining_users) == {b, c}
finally:
await engine.dispose()