343 lines
12 KiB
Python
343 lines
12 KiB
Python
"""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, computed_field, model_validator
|
|
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 | None = Field(default=None, max_length=_BODY_MAX_LEN)
|
|
content: str | None = Field(default=None, max_length=_BODY_MAX_LEN)
|
|
room_id: int | None = None
|
|
|
|
@model_validator(mode="before")
|
|
@classmethod
|
|
def resolve_body_or_content(cls, data: object) -> object:
|
|
if isinstance(data, dict):
|
|
if not data.get("body") and data.get("content"):
|
|
data["body"] = data["content"]
|
|
elif not data.get("content") and data.get("body"):
|
|
data["content"] = data["body"]
|
|
return data
|
|
|
|
@model_validator(mode="after")
|
|
def check_non_empty(self) -> SendWorldMessageRequest:
|
|
text = self.body or self.content
|
|
if not text or len(text.strip()) < 1:
|
|
raise ValueError("Message body or content must not be empty")
|
|
if not self.body:
|
|
self.body = text
|
|
if not self.content:
|
|
self.content = text
|
|
return self
|
|
|
|
|
|
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}
|
|
|
|
@computed_field
|
|
@property
|
|
def content(self) -> str:
|
|
return self.body
|
|
|
|
|
|
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
|
|
)
|