From 39a971f00e08026111dd8cd17e5526813cb7f232 Mon Sep 17 00:00:00 2001 From: daddy Date: Tue, 15 Sep 2026 17:55:47 +0000 Subject: [PATCH] Implement in-experience (world/room) chat endpoints and pub/sub hook (US-4) --- backend/app/main.py | 4 +- backend/app/routers/chat_world.py | 315 ++++++++++++++++++++++++++++ backend/app/services/chat_pubsub.py | 48 +++++ backend/tests/test_chat_world.py | 246 ++++++++++++++++++++++ 4 files changed, 612 insertions(+), 1 deletion(-) create mode 100644 backend/app/routers/chat_world.py create mode 100644 backend/app/services/chat_pubsub.py create mode 100644 backend/tests/test_chat_world.py diff --git a/backend/app/main.py b/backend/app/main.py index 5489f54..900b73a 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, avatar, world_content, internal, presence, groups, social_friends, profiles, chat_dm, chat_reports +from app.routers import auth, users, worlds, assets, avatar, world_content, internal, presence, groups, social_friends, profiles, chat_dm, chat_reports, chat_world from app.services.game_servers import supervisor @@ -51,6 +51,8 @@ app.include_router(profiles.router, prefix="/api/profile", tags=["profile"]) app.include_router(chat_dm.router, prefix="/api/chat/dm", tags=["chat-dm"]) app.include_router(chat_reports.chat_router, prefix="/api/chat", tags=["chat-reports"]) app.include_router(chat_reports.moderation_router, prefix="/api/moderation", tags=["moderation"]) +app.include_router(chat_world.router, prefix="/api/chat", tags=["chat-world"]) +app.include_router(chat_world.router, prefix="/v1/chat", tags=["chat-world"]) @app.get("/api/health") diff --git a/backend/app/routers/chat_world.py b/backend/app/routers/chat_world.py new file mode 100644 index 0000000..9eda708 --- /dev/null +++ b/backend/app/routers/chat_world.py @@ -0,0 +1,315 @@ +"""In-experience world and room chat router. + +docs/planning/social.md §4/§5 (US-4). Builds the world- and room-scoped +chat REST endpoints on top of ``app/models/chat.py`` (Message), +``app/services/text_filter.py``, ``app/rate_limit.py``, and +``app/services/chat_pubsub.py``. + +Endpoints: +- POST /api/chat/world/{world_id}/messages -- send a message to a world channel +- GET /api/chat/world/{world_id}/messages -- paginated world chat history +- POST /api/chat/room/{room_id}/messages -- send a message to a room channel +- GET /api/chat/room/{room_id}/messages -- paginated room chat history + +Features & invariants: +- **Visibility enforcement**: + - PUBLIC worlds: accessible to any authenticated user. + - PRIVATE worlds: accessible only to the world owner. + - FRIENDS_ONLY worlds: accessible only to the owner and accepted friends of the owner. +- **Room scoping**: + - If a room_id is specified, the room must exist and belong to that world. +- **Text filtering**: + - Reuses ``default_text_filter()``: + - ALLOW: stored as-is, ``filtered=False``. + - REDACT: starred out, ``filtered=True``. + - REJECT: rejected with 422 Unprocessable Entity. +- **Rate limiting**: + - ``rate_limit("chat_world_send", max_hits=30, window_seconds=60)``. +- **Author display**: + - Returns author info (id, username, avatar_asset_id) for rendering. +- **Real-time hook**: + - Publishes an event to Redis pub/sub channels (``world:{world_id}:chat``, + and ``room:{room_id}:chat`` if room-scoped) so the WebSocket gateway can + broadcast in real time. +""" + +from __future__ import annotations + +from datetime import datetime +from typing import Annotated + +from fastapi import APIRouter, Depends, HTTPException, Query, status +from pydantic import BaseModel, Field +from sqlalchemy import and_, or_, select +from sqlalchemy.ext.asyncio import AsyncSession +from sqlalchemy.orm import joinedload + +from app.database import get_db +from app.dependencies import CurrentUser +from app.models.chat import Message +from app.models.social import Friendship, FriendshipStatus +from app.models.user import User +from app.models.world import World, WorldVisibility +from app.models.world_content import WorldRoom +from app.rate_limit import rate_limit +from app.services.chat_pubsub import publish_world_chat_event +from app.services.text_filter import FilterAction, default_text_filter + +router = APIRouter() + +DbDep = Annotated[AsyncSession, Depends(get_db)] + +_text_filter = default_text_filter() +chat_world_send_rate_limit = rate_limit("chat_world_send", max_hits=30, window_seconds=60) + +_BODY_MAX_LEN = 4000 + +# ──────────────────────────── Pydantic schemas ──────────────────────────── + + +class SendWorldMessageRequest(BaseModel): + body: str = Field(..., min_length=1, max_length=_BODY_MAX_LEN) + room_id: int | None = None + + +class AuthorInfo(BaseModel): + id: int | None + username: str + avatar_asset_id: int | None = None + + model_config = {"from_attributes": True} + + +class WorldMessageResponse(BaseModel): + id: int + world_id: int + room_id: int | None + sender_id: int | None + author: AuthorInfo | None + body: str + filtered: bool + created_at: datetime + + model_config = {"from_attributes": True} + + +class WorldMessageListResponse(BaseModel): + messages: list[WorldMessageResponse] + next_cursor: int | None = None + + +# ──────────────────────────────── Helpers ────────────────────────────────── + + +async def _can_access_world(db: AsyncSession, world: World, user_id: int) -> bool: + if world.visibility == WorldVisibility.PUBLIC: + return True + if world.owner_id == user_id: + return True + if world.visibility == WorldVisibility.FRIENDS_ONLY: + res = await db.execute( + select(Friendship.id).where( + or_( + and_(Friendship.requester_id == world.owner_id, Friendship.addressee_id == user_id), + and_(Friendship.requester_id == user_id, Friendship.addressee_id == world.owner_id), + ), + Friendship.status == FriendshipStatus.ACCEPTED, + ).limit(1) + ) + return res.scalar_one_or_none() is not None + return False + + +async def _get_world_and_check_access(db: AsyncSession, world_id: int, current_user_id: int) -> World: + result = await db.execute(select(World).where(World.id == world_id)) + world = result.scalar_one_or_none() + if world is None: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="World not found") + if not await _can_access_world(db, world, current_user_id): + raise HTTPException( + status_code=status.HTTP_403_FORBIDDEN, detail="You do not have access to this world" + ) + return world + + +async def _get_room_and_check_access( + db: AsyncSession, room_id: int, current_user_id: int +) -> tuple[WorldRoom, World]: + result = await db.execute(select(WorldRoom).where(WorldRoom.id == room_id)) + room = result.scalar_one_or_none() + if room is None: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Room not found") + world = await _get_world_and_check_access(db, room.world_id, current_user_id) + return room, world + + +def _to_world_message_response(msg: Message) -> WorldMessageResponse: + author: AuthorInfo | None = None + if msg.sender is not None: + avatar_asset_id = None + if msg.sender.profile is not None: + avatar_asset_id = msg.sender.profile.avatar_asset_id + author = AuthorInfo( + id=msg.sender.id, + username=msg.sender.username, + avatar_asset_id=avatar_asset_id, + ) + elif msg.sender_id is None: + author = AuthorInfo( + id=None, + username="[deleted]", + avatar_asset_id=None, + ) + + return WorldMessageResponse( + id=msg.id, + world_id=msg.world_id, + room_id=msg.room_id, + sender_id=msg.sender_id, + author=author, + body=msg.body, + filtered=msg.filtered, + created_at=msg.created_at, + ) + + +# ──────────────────────────────── Endpoints ──────────────────────────────── + + +@router.post( + "/world/{world_id}/messages", + response_model=WorldMessageResponse, + status_code=status.HTTP_201_CREATED, + dependencies=[Depends(chat_world_send_rate_limit)], +) +async def send_world_message( + world_id: int, + body: SendWorldMessageRequest, + current_user: CurrentUser, + db: DbDep, +): + """Send a message to a world (and optional room) channel.""" + world = await _get_world_and_check_access(db, world_id, current_user.id) + + if body.room_id is not None: + result = await db.execute(select(WorldRoom).where(WorldRoom.id == body.room_id)) + room = result.scalar_one_or_none() + if room is None or room.world_id != world.id: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, detail="Room not found in this world" + ) + + filter_res = _text_filter.check(body.body) + if filter_res.action is FilterAction.REJECT: + raise HTTPException( + status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, + detail="Message rejected by the content filter.", + ) + + msg = Message( + sender_id=current_user.id, + world_id=world.id, + room_id=body.room_id, + body=filter_res.filtered_text, + filtered=(filter_res.action is FilterAction.REDACT), + ) + db.add(msg) + await db.commit() + + stmt = ( + select(Message) + .options(joinedload(Message.sender).joinedload(User.profile)) + .where(Message.id == msg.id) + ) + msg_loaded = (await db.execute(stmt)).scalar_one() + resp = _to_world_message_response(msg_loaded) + + # Trigger async Redis pub/sub hook + await publish_world_chat_event(world.id, resp.model_dump(mode="json"), room_id=body.room_id) + + return resp + + +@router.get( + "/world/{world_id}/messages", + response_model=WorldMessageListResponse, +) +async def get_world_chat_history( + world_id: int, + current_user: CurrentUser, + db: DbDep, + room_id: int | None = Query(None, description="Optional room id filter"), + limit: int = Query(50, ge=1, le=100), + before: int | None = Query( + None, description="Return messages older than this message id (pagination cursor)." + ), +): + """Paginated world chat history, newest first.""" + world = await _get_world_and_check_access(db, world_id, current_user.id) + + if room_id is not None: + result = await db.execute(select(WorldRoom).where(WorldRoom.id == room_id)) + room = result.scalar_one_or_none() + if room is None or room.world_id != world.id: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, detail="Room not found in this world" + ) + + stmt = ( + select(Message) + .options(joinedload(Message.sender).joinedload(User.profile)) + .where(Message.world_id == world.id) + ) + if room_id is not None: + stmt = stmt.where(Message.room_id == room_id) + if before is not None: + stmt = stmt.where(Message.id < before) + stmt = stmt.order_by(Message.id.desc()).limit(limit + 1) + + rows = (await db.execute(stmt)).scalars().all() + has_more = len(rows) > limit + page = rows[:limit] + next_cursor = page[-1].id if has_more and page else None + + return WorldMessageListResponse( + messages=[_to_world_message_response(m) for m in page], + next_cursor=next_cursor, + ) + + +@router.post( + "/room/{room_id}/messages", + response_model=WorldMessageResponse, + status_code=status.HTTP_201_CREATED, + dependencies=[Depends(chat_world_send_rate_limit)], +) +async def send_room_message( + room_id: int, + body: SendWorldMessageRequest, + current_user: CurrentUser, + db: DbDep, +): + """Send a message directly addressed to a room channel.""" + room, world = await _get_room_and_check_access(db, room_id, current_user.id) + forward_req = SendWorldMessageRequest(body=body.body, room_id=room.id) + return await send_world_message(world.id, forward_req, current_user, db) + + +@router.get( + "/room/{room_id}/messages", + response_model=WorldMessageListResponse, +) +async def get_room_chat_history( + room_id: int, + current_user: CurrentUser, + db: DbDep, + limit: int = Query(50, ge=1, le=100), + before: int | None = Query( + None, description="Return messages older than this message id (pagination cursor)." + ), +): + """Paginated room chat history, newest first.""" + room, world = await _get_room_and_check_access(db, room_id, current_user.id) + return await get_world_chat_history( + world.id, current_user, db, room_id=room.id, limit=limit, before=before + ) diff --git a/backend/app/services/chat_pubsub.py b/backend/app/services/chat_pubsub.py new file mode 100644 index 0000000..bb5b657 --- /dev/null +++ b/backend/app/services/chat_pubsub.py @@ -0,0 +1,48 @@ +"""Pub/sub messaging bus for chat events over Redis. + +Provides asynchronous publish hooks for world/room chat and DMs, allowing the +WebSocket real-time gateway (and multi-instance replicas) to receive and fan out +chat messages in real time. +""" + +from __future__ import annotations + +import json +import logging +from typing import Any + +import redis.asyncio as aioredis + +from app.config import settings + +logger = logging.getLogger(__name__) + + +async def publish_world_chat_event( + world_id: int, payload: dict[str, Any], room_id: int | None = None +) -> None: + """Publish a world chat message event to Redis pub/sub for real-time delivery.""" + try: + client = aioredis.from_url(settings.REDIS_URL, encoding="utf-8", decode_responses=True) + async with client: + channel = f"world:{world_id}:chat" + data = json.dumps(payload, default=str) + await client.publish(channel, data) + if room_id is not None: + await client.publish(f"room:{room_id}:chat", data) + except Exception as exc: + logger.debug("Failed to publish world chat event to Redis: %s", exc) + + +async def publish_dm_chat_event( + conversation_id: int, payload: dict[str, Any] +) -> None: + """Publish a DM chat message event to Redis pub/sub for real-time delivery.""" + try: + client = aioredis.from_url(settings.REDIS_URL, encoding="utf-8", decode_responses=True) + async with client: + channel = f"dm:{conversation_id}:chat" + data = json.dumps(payload, default=str) + await client.publish(channel, data) + except Exception as exc: + logger.debug("Failed to publish DM chat event to Redis: %s", exc) diff --git a/backend/tests/test_chat_world.py b/backend/tests/test_chat_world.py new file mode 100644 index 0000000..80e4cbe --- /dev/null +++ b/backend/tests/test_chat_world.py @@ -0,0 +1,246 @@ +"""Integration tests for in-experience world/room chat: +- POST /api/chat/world/{world_id}/messages +- GET /api/chat/world/{world_id}/messages +- POST /api/chat/room/{room_id}/messages +- GET /api/chat/room/{room_id}/messages +""" + +import pytest +from fastapi.testclient import TestClient + +from app.main import app + + +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"], "username": username} + + +def _make_world(client: TestClient, headers: dict, visibility: str = "public") -> int: + r = client.post("/api/worlds/", headers=headers, json={"name": "TestWorld", "visibility": visibility}) + assert r.status_code == 201, r.text + return r.json()["id"] + + +def _make_friends(client: TestClient, user_a: dict, user_b: dict) -> None: + req = client.post("/api/friends/requests", json={"user_id": user_b["id"]}, headers=user_a["headers"]) + assert req.status_code == 201, req.text + req_id = req.json()["id"] + acc = client.post(f"/api/friends/requests/{req_id}/accept", headers=user_b["headers"]) + assert acc.status_code == 200, acc.text + + +class TestWorldChat: + def test_send_and_get_world_message(self): + client = TestClient(app) + alice = _register(client, "wch_alice1") + bob = _register(client, "wch_bob1") + + world_id = _make_world(client, alice["headers"], visibility="public") + + # Bob posts to Alice's public world + post_resp = client.post( + f"/api/chat/world/{world_id}/messages", + headers=bob["headers"], + json={"body": "Hello world from Bob!"}, + ) + assert post_resp.status_code == 201, post_resp.text + msg_data = post_resp.json() + assert msg_data["world_id"] == world_id + assert msg_data["room_id"] is None + assert msg_data["sender_id"] == bob["id"] + assert msg_data["body"] == "Hello world from Bob!" + assert msg_data["filtered"] is False + assert msg_data["author"]["username"] == "wch_bob1" + assert msg_data["author"]["id"] == bob["id"] + + # Alice gets world chat history + get_resp = client.get( + f"/api/chat/world/{world_id}/messages", + headers=alice["headers"], + ) + assert get_resp.status_code == 200, get_resp.text + history = get_resp.json() + assert len(history["messages"]) == 1 + assert history["messages"][0]["id"] == msg_data["id"] + assert history["messages"][0]["author"]["username"] == "wch_bob1" + + def test_room_scoped_messages(self): + client = TestClient(app) + alice = _register(client, "wch_alice2") + world_id = _make_world(client, alice["headers"]) + + # Fetch auto-created main room + rooms_resp = client.get(f"/api/worlds/{world_id}/rooms", headers=alice["headers"]) + assert rooms_resp.status_code == 200, rooms_resp.text + rooms = rooms_resp.json() + assert len(rooms) >= 1 + room_id = rooms[0]["id"] + + # Post directly via room endpoint + post_room = client.post( + f"/api/chat/room/{room_id}/messages", + headers=alice["headers"], + json={"body": "Message in room"}, + ) + assert post_room.status_code == 201, post_room.text + msg_data = post_room.json() + assert msg_data["world_id"] == world_id + assert msg_data["room_id"] == room_id + assert msg_data["body"] == "Message in room" + + # Get room history + get_room = client.get( + f"/api/chat/room/{room_id}/messages", + headers=alice["headers"], + ) + assert get_room.status_code == 200, get_room.text + messages = get_room.json()["messages"] + assert len(messages) == 1 + assert messages[0]["id"] == msg_data["id"] + + # Also visible when querying world history with room_id filter + get_world_room = client.get( + f"/api/chat/world/{world_id}/messages?room_id={room_id}", + headers=alice["headers"], + ) + assert get_world_room.status_code == 200, get_world_room.text + assert len(get_world_room.json()["messages"]) == 1 + + def test_visibility_access_control(self): + client = TestClient(app) + alice = _register(client, "wch_alice3") + bob = _register(client, "wch_bob3") + charlie = _register(client, "wch_charlie3") + + # Private world: only Alice + priv_world_id = _make_world(client, alice["headers"], visibility="private") + + # Bob cannot send or read + bob_send = client.post( + f"/api/chat/world/{priv_world_id}/messages", + headers=bob["headers"], + json={"body": "Trying to trespass"}, + ) + assert bob_send.status_code == 403, bob_send.text + + bob_read = client.get( + f"/api/chat/world/{priv_world_id}/messages", + headers=bob["headers"], + ) + assert bob_read.status_code == 403, bob_read.text + + # Alice (owner) can send and read + alice_send = client.post( + f"/api/chat/world/{priv_world_id}/messages", + headers=alice["headers"], + json={"body": "My private sanctum"}, + ) + assert alice_send.status_code == 201, alice_send.text + + # Friends-only world + fr_world_id = _make_world(client, alice["headers"], visibility="friends_only") + _make_friends(client, alice, bob) + + # Bob (friend) can post + bob_fr_send = client.post( + f"/api/chat/world/{fr_world_id}/messages", + headers=bob["headers"], + json={"body": "Hello friend!"}, + ) + assert bob_fr_send.status_code == 201, bob_fr_send.text + + # Charlie (not friend) cannot post or read + charlie_send = client.post( + f"/api/chat/world/{fr_world_id}/messages", + headers=charlie["headers"], + json={"body": "Can I join?"}, + ) + assert charlie_send.status_code == 403, charlie_send.text + + charlie_read = client.get( + f"/api/chat/world/{fr_world_id}/messages", + headers=charlie["headers"], + ) + assert charlie_read.status_code == 403, charlie_read.text + + def test_not_found_validation(self): + client = TestClient(app) + alice = _register(client, "wch_alice4") + + # Non-existent world + res = client.post( + "/api/chat/world/999999/messages", + headers=alice["headers"], + json={"body": "Ghost world"}, + ) + assert res.status_code == 404, res.text + + # Non-existent room + world_id = _make_world(client, alice["headers"]) + res2 = client.post( + f"/api/chat/world/{world_id}/messages", + headers=alice["headers"], + json={"body": "Ghost room", "room_id": 999999}, + ) + assert res2.status_code == 404, res2.text + + def test_text_filtering_in_world_chat(self): + client = TestClient(app) + alice = _register(client, "wch_alice5") + world_id = _make_world(client, alice["headers"]) + + # Default filter redacts forbidden words + res = client.post( + f"/api/chat/world/{world_id}/messages", + headers=alice["headers"], + json={"body": "you are a bastard"}, + ) + assert res.status_code == 201, res.text + data = res.json() + assert data["filtered"] is True + assert "bastard" not in data["body"] + assert "*" in data["body"] + + def test_pagination_and_next_cursor(self): + client = TestClient(app) + alice = _register(client, "wch_alice6") + world_id = _make_world(client, alice["headers"]) + + for i in range(5): + r = client.post( + f"/api/chat/world/{world_id}/messages", + headers=alice["headers"], + json={"body": f"Msg {i}"}, + ) + assert r.status_code == 201 + + page1 = client.get( + f"/api/chat/world/{world_id}/messages?limit=2", + headers=alice["headers"], + ).json() + assert len(page1["messages"]) == 2 + assert page1["next_cursor"] is not None + assert page1["messages"][0]["body"] == "Msg 4" + assert page1["messages"][1]["body"] == "Msg 3" + + page2 = client.get( + f"/api/chat/world/{world_id}/messages?limit=2&before={page1['next_cursor']}", + headers=alice["headers"], + ).json() + assert len(page2["messages"]) == 2 + assert page2["messages"][0]["body"] == "Msg 2" + assert page2["messages"][1]["body"] == "Msg 1"