"""Direct-message (DM) chat router. docs/planning/social.md §4/§5 (US-4). Builds the send/list REST surface on top of ``app/models/chat.py`` (Conversation/ConversationParticipant/Message, already merged), ``app/services/text_filter.py`` (already merged), and ``app/rate_limit.py`` (already merged). Endpoints: - POST /api/chat/dm/{user_id}/messages -- send a DM to ``user_id`` - GET /api/chat/dm/{user_id}/messages -- paginated history with ``user_id`` - GET /api/chat/dm/conversations -- list the caller's conversations Scope/boundaries (per task): REST send/list only, no WebSocket/real-time delivery (separate task), no world/room chat (separate sibling task; that uses the same ``Message`` table but with ``world_id``/``room_id`` set and ``conversation_id`` null). Design decisions this task owns: - **Blocking.** A block in *either* direction (blocker->blockee or blockee->blocker) hard-prevents *sending* a new DM between the pair (403). It does not hide history the pair already has -- ``GET .../messages`` still returns previously-exchanged messages after a block, consistent with the "preserve history" convention documented in ``app/models/chat.py``'s module docstring. Only sending is blocked. - **Text filtering.** Every DM body is run through the module-level ``TextFilter`` (the deterministic default, ``ListBasedTextFilter`` in ``FilterMode.REDACT``) before storing: - ALLOW -> store the body as-is, ``filtered=False``. - REDACT -> store ``filtered_text`` (matched spans starred out), ``filtered=True``. - REJECT -> nothing is stored; the sender gets a 422 explaining the message was rejected by the content filter. (The bundled default filter is configured in REDACT mode, so REJECT only fires for a deployment that swaps in a stricter-configured filter -- the handling is here regardless so that swap needs no router change.) - **Rate limiting.** Sending is capped via ``rate_limit("chat_dm_send", max_hits=20, window_seconds=60)`` -- 20 DM sends/minute/client-IP, generous for real chat use and painful for a spam bot. A no-op when ``settings.RATE_LIMIT_ENABLED`` is false (see ``app/rate_limit.py``), matching every other rate-limited endpoint. - **Unread tracking.** A last-read timestamp per participant (``ConversationParticipant.last_read_at``, migration 0010). Fetching a conversation's *latest* page of history (``GET .../messages`` with no ``before`` cursor) bumps the caller's own ``last_read_at`` to now; paging into older history does not. ``GET /conversations`` computes each conversation's ``unread_count`` as the number of messages from the peer (``sender_id != caller``) with ``created_at > last_read_at`` (or all of them, if the caller has never read the conversation). - **Pagination.** - History (``GET {user_id}/messages``): cursor-based, newest-first. ``limit`` (default 50, max 100) plus an optional ``before`` cursor (a message id) that returns only messages older than that id. Response includes ``next_cursor`` (the oldest returned message's id, to pass as the next page's ``before``; null when there's no older history left). - Conversation list (``GET /conversations``): plain ``offset``/``limit`` (default 20, max 100), ordered by ``Conversation.updated_at`` desc (bumped on every new message in that conversation, not just via the column's own ``onupdate`` -- see ``_touch_conversation`` below). - Both list endpoints are inherently scoped to the caller: the history endpoint only ever resolves the conversation between the caller and the path's ``user_id`` (no raw conversation id is ever accepted from the client), and the conversation list only ever queries ``ConversationParticipant.user_id == caller``. There is no way for a client to address another pair's conversation. """ from __future__ import annotations from datetime import datetime, timezone from typing import Annotated from fastapi import APIRouter, Depends, HTTPException, Query, status from pydantic import BaseModel, Field from sqlalchemy import and_, func, or_, select from sqlalchemy.ext.asyncio import AsyncSession from app.database import get_db from app.dependencies import CurrentUser from app.models.chat import Conversation, ConversationParticipant, Message from app.models.social import Block from app.models.user import User from app.rate_limit import rate_limit from app.services.chat_pubsub import publish_dm_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_dm_send_rate_limit = rate_limit("chat_dm_send", max_hits=20, window_seconds=60) _BODY_MAX_LEN = 4000 # ──────────────────────────── Pydantic schemas ──────────────────────────── class SendDMRequest(BaseModel): body: str = Field(..., min_length=1, max_length=_BODY_MAX_LEN) class DMMessageResponse(BaseModel): id: int conversation_id: int sender_id: int recipient_id: int body: str filtered: bool created_at: datetime model_config = {"from_attributes": True} class DMMessageListResponse(BaseModel): messages: list[DMMessageResponse] next_cursor: int | None = None class DMConversationSummary(BaseModel): conversation_id: int peer_id: int peer_username: str last_message: DMMessageResponse | None unread_count: int updated_at: datetime class DMConversationListResponse(BaseModel): conversations: list[DMConversationSummary] total: int # ──────────────────────────────── Helpers ────────────────────────────────── async def _get_target_user(db: AsyncSession, user_id: int) -> User: result = await db.execute(select(User).where(User.id == user_id)) user = result.scalar_one_or_none() if user is None: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found") return user async def _is_blocked_either_direction(db: AsyncSession, user_a: int, user_b: int) -> bool: result = await db.execute( select(Block.id).where( or_( and_(Block.blocker_id == user_a, Block.blocked_id == user_b), and_(Block.blocker_id == user_b, Block.blocked_id == user_a), ) ).limit(1) ) return result.scalar_one_or_none() is not None async def _get_or_create_conversation( db: AsyncSession, user_a_id: int, user_b_id: int ) -> Conversation: lo, hi = sorted((user_a_id, user_b_id)) result = await db.execute( select(Conversation).where( Conversation.user_a_id == lo, Conversation.user_b_id == hi ) ) conversation = result.scalar_one_or_none() if conversation is not None: return conversation conversation = Conversation(user_a_id=lo, user_b_id=hi) db.add(conversation) await db.flush() # assign conversation.id db.add(ConversationParticipant(conversation_id=conversation.id, user_id=lo)) db.add(ConversationParticipant(conversation_id=conversation.id, user_id=hi)) await db.flush() return conversation async def _get_participant( db: AsyncSession, conversation_id: int, user_id: int ) -> ConversationParticipant | None: result = await db.execute( select(ConversationParticipant).where( ConversationParticipant.conversation_id == conversation_id, ConversationParticipant.user_id == user_id, ) ) return result.scalar_one_or_none() def _peer_id(conversation: Conversation, self_id: int) -> int: return conversation.user_b_id if conversation.user_a_id == self_id else conversation.user_a_id def _touch_conversation(conversation: Conversation) -> None: """Bump ``updated_at`` explicitly. ``Conversation.updated_at`` has ``onupdate=func.now()``, but that only fires when an UPDATE statement touches the ``conversations`` row itself -- inserting a ``Message`` row alone never does. Explicitly dirtying the column here is what actually keeps "most recently active conversation" ordering correct for ``GET /conversations``. """ conversation.updated_at = datetime.now(timezone.utc) def _to_message_response(message: Message, recipient_id: int) -> DMMessageResponse: return DMMessageResponse( id=message.id, conversation_id=message.conversation_id, sender_id=message.sender_id, recipient_id=recipient_id, body=message.body, filtered=message.filtered, created_at=message.created_at, ) # ──────────────────────────────── Endpoints ──────────────────────────────── @router.post( "/{user_id}/messages", response_model=DMMessageResponse, status_code=status.HTTP_201_CREATED, dependencies=[Depends(chat_dm_send_rate_limit)], ) async def send_dm( user_id: int, body: SendDMRequest, current_user: CurrentUser, db: DbDep, ): """Send a DM to ``user_id``.""" if user_id == current_user.id: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Cannot send a DM to yourself" ) await _get_target_user(db, user_id) if await _is_blocked_either_direction(db, current_user.id, user_id): raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="Cannot message this user", ) result = _text_filter.check(body.body) if result.action is FilterAction.REJECT: raise HTTPException( status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="Message rejected by the content filter.", ) conversation = await _get_or_create_conversation(db, current_user.id, user_id) message = Message( sender_id=current_user.id, conversation_id=conversation.id, body=result.filtered_text, filtered=(result.action is FilterAction.REDACT), ) db.add(message) _touch_conversation(conversation) await db.commit() await db.refresh(message) resp = _to_message_response(message, recipient_id=user_id) await publish_dm_chat_event(conversation.id, resp.model_dump(mode="json")) return resp @router.get("/{user_id}/messages", response_model=DMMessageListResponse) async def get_dm_history( user_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 DM history between the caller and ``user_id``, newest first. Marks the conversation as read (bumps the caller's ``last_read_at``) only when fetching the first page (``before`` omitted) -- paging into older history doesn't affect unread state. """ await _get_target_user(db, user_id) lo, hi = sorted((current_user.id, user_id)) conv_result = await db.execute( select(Conversation).where(Conversation.user_a_id == lo, Conversation.user_b_id == hi) ) conversation = conv_result.scalar_one_or_none() if conversation is None: # No conversation yet -- nothing exchanged, not an error. return DMMessageListResponse(messages=[], next_cursor=None) stmt = select(Message).where(Message.conversation_id == conversation.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 if before is None: participant = await _get_participant(db, conversation.id, current_user.id) if participant is not None: participant.last_read_at = datetime.now(timezone.utc) await db.commit() return DMMessageListResponse( messages=[_to_message_response(m, recipient_id=user_id) for m in page], next_cursor=next_cursor, ) @router.get("/conversations", response_model=DMConversationListResponse) async def list_conversations( current_user: CurrentUser, db: DbDep, limit: int = Query(20, ge=1, le=100), offset: int = Query(0, ge=0), ): """List the caller's DM conversations, most recently active first.""" total_result = await db.execute( select(func.count()) .select_from(ConversationParticipant) .where(ConversationParticipant.user_id == current_user.id) ) total = total_result.scalar_one() stmt = ( select(Conversation, ConversationParticipant) .join( ConversationParticipant, ConversationParticipant.conversation_id == Conversation.id, ) .where(ConversationParticipant.user_id == current_user.id) .order_by(Conversation.updated_at.desc()) .offset(offset) .limit(limit) ) rows = (await db.execute(stmt)).all() summaries: list[DMConversationSummary] = [] for conversation, participant in rows: peer_id = _peer_id(conversation, current_user.id) peer = ( await db.execute(select(User).where(User.id == peer_id)) ).scalar_one_or_none() peer_username = peer.username if peer is not None else "(deleted user)" last_message_row = ( await db.execute( select(Message) .where(Message.conversation_id == conversation.id) .order_by(Message.id.desc()) .limit(1) ) ).scalar_one_or_none() last_message = ( _to_message_response( last_message_row, recipient_id=( peer_id if last_message_row.sender_id == current_user.id else current_user.id ), ) if last_message_row is not None else None ) unread_filters = [ Message.conversation_id == conversation.id, Message.sender_id != current_user.id, ] if participant.last_read_at is not None: unread_filters.append(Message.created_at > participant.last_read_at) unread_count = ( await db.execute(select(func.count()).select_from(Message).where(*unread_filters)) ).scalar_one() summaries.append( DMConversationSummary( conversation_id=conversation.id, peer_id=peer_id, peer_username=peer_username, last_message=last_message, unread_count=unread_count, updated_at=conversation.updated_at, ) ) return DMConversationListResponse(conversations=summaries, total=total)