493 lines
18 KiB
Python
493 lines
18 KiB
Python
"""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, computed_field, model_validator
|
|
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 | None = Field(default=None, max_length=_BODY_MAX_LEN)
|
|
content: str | None = Field(default=None, max_length=_BODY_MAX_LEN)
|
|
|
|
@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) -> SendDMRequest:
|
|
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 StartConversationRequest(BaseModel):
|
|
target_user_id: int
|
|
|
|
|
|
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}
|
|
|
|
@computed_field
|
|
@property
|
|
def content(self) -> str:
|
|
return self.body
|
|
|
|
|
|
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)
|
|
@router.post("/conversations")
|
|
async def start_or_get_conversation_endpoint(
|
|
body: StartConversationRequest,
|
|
current_user: CurrentUser,
|
|
db: DbDep,
|
|
):
|
|
"""Start or get existing conversation with target_user_id."""
|
|
if body.target_user_id == current_user.id:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST, detail="Cannot start a conversation with yourself"
|
|
)
|
|
await _get_target_user(db, body.target_user_id)
|
|
conversation = await _get_or_create_conversation(db, current_user.id, body.target_user_id)
|
|
return {"id": conversation.id, "target_user_id": body.target_user_id}
|
|
|
|
|
|
@router.get("/conversations/{conversation_id}/messages", response_model=DMMessageListResponse)
|
|
async def get_dm_conversation_history(
|
|
conversation_id: int,
|
|
current_user: CurrentUser,
|
|
db: DbDep,
|
|
limit: int = Query(50, ge=1, le=100),
|
|
before: int | None = Query(None),
|
|
):
|
|
"""Get messages for a conversation by conversation_id."""
|
|
conv_result = await db.execute(
|
|
select(Conversation).where(
|
|
Conversation.id == conversation_id,
|
|
or_(Conversation.user_a_id == current_user.id, Conversation.user_b_id == current_user.id),
|
|
)
|
|
)
|
|
conversation = conv_result.scalar_one_or_none()
|
|
if conversation is None:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Conversation not found")
|
|
peer_id = _peer_id(conversation, current_user.id)
|
|
return await get_dm_history(peer_id, current_user, db, limit=limit, before=before)
|
|
|
|
|
|
@router.post(
|
|
"/conversations/{conversation_id}/messages",
|
|
response_model=DMMessageResponse,
|
|
status_code=status.HTTP_201_CREATED,
|
|
dependencies=[Depends(chat_dm_send_rate_limit)],
|
|
)
|
|
async def send_dm_to_conversation(
|
|
conversation_id: int,
|
|
body: SendDMRequest,
|
|
current_user: CurrentUser,
|
|
db: DbDep,
|
|
):
|
|
"""Send a message to a conversation by conversation_id."""
|
|
conv_result = await db.execute(
|
|
select(Conversation).where(
|
|
Conversation.id == conversation_id,
|
|
or_(Conversation.user_a_id == current_user.id, Conversation.user_b_id == current_user.id),
|
|
)
|
|
)
|
|
conversation = conv_result.scalar_one_or_none()
|
|
if conversation is None:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Conversation not found")
|
|
peer_id = _peer_id(conversation, current_user.id)
|
|
return await send_dm(peer_id, body, current_user, db)
|