import json import logging from collections.abc import Mapping from typing import Any from fastapi import APIRouter, Depends, HTTPException, Query, Request, Response, status from sqlalchemy import text from sqlalchemy.exc import IntegrityError from sqlalchemy.orm import Session try: # For module mode: `uvicorn app.main:app` from app.db import get_db from app.db_compat import ( current_timestamp_sql, get_db_dialect_name, inserted_primary_key, select_for_update_clause, ) from app.security import get_current_user_id from app.schemas.message import ( MessageCreateRequest, MessageCreateResponse, MessageItem, MessageListResponse, ) except ModuleNotFoundError: # For script mode: `python app/main.py` or VS Code "Run Python File" from db import get_db from db_compat import ( current_timestamp_sql, get_db_dialect_name, inserted_primary_key, select_for_update_clause, ) from security import get_current_user_id from schemas.message import ( MessageCreateRequest, MessageCreateResponse, MessageItem, MessageListResponse, ) router = APIRouter(prefix="/messages", tags=["messages"]) logger = logging.getLogger("app.messages") PARENT_PARTICIPANT_TYPE = 1 PARENT_DIRECT_CONVERSATION_TYPE = 3 def _build_preview(content_type: int, content_text: str | None) -> str: if content_type == 1: preview = f"user: {(content_text or '').strip()}" elif content_type == 2: preview = "user: [audio]" elif content_type == 3: preview = "user: [image]" else: preview = "user: [json]" return preview[:255] def _normalize_content_json(value: Any) -> dict[str, Any] | None: if value is None: return None if isinstance(value, dict): return value if isinstance(value, str): try: parsed = json.loads(value) if isinstance(parsed, dict): return parsed except json.JSONDecodeError: return None return None def _row_to_message_item(row: Mapping[str, Any]) -> MessageItem: sender_user_id = row.get("sender_user_id") if sender_user_id is not None: sender_user_id = int(sender_user_id) return MessageItem( id=int(row["id"]), conversation_id=int(row["conversation_id"]), seq=int(row["seq"]), sender_user_id=sender_user_id, role=int(row.get("role", 1)), content_type=int(row["content_type"]), content_text=row["content_text"], content_json=_normalize_content_json(row["content_json"]), media_file_key=row["media_file_key"], media_duration_ms=row["media_duration_ms"], client_msg_id=row["client_msg_id"], created_at=row["created_at"], ) def _get_existing_message( db: Session, conversation_id: int, client_msg_id: str ) -> Mapping[str, Any] | None: return ( db.execute( text( """ SELECT id, conversation_id, seq, CASE WHEN sender_type = :parent_participant_type THEN sender_id ELSE NULL END AS sender_user_id, 1 AS role, content_type, content_text, content_json, media_file_key, media_duration_ms, client_msg_id, created_at FROM im_messages WHERE conversation_id = :conversation_id AND client_msg_id = :client_msg_id LIMIT 1 """ ), { "conversation_id": conversation_id, "client_msg_id": client_msg_id, "parent_participant_type": PARENT_PARTICIPANT_TYPE, }, ) .mappings() .first() ) def _build_parent_direct_pair(user_a_id: int, user_b_id: int) -> tuple[str, str, str]: low_id, high_id = sorted((user_a_id, user_b_id)) participant_a_id = str(low_id) participant_b_id = str(high_id) return participant_a_id, participant_b_id, f"{participant_a_id}:{participant_b_id}" def _next_primary_key(db: Session, table_name: str) -> int: if table_name not in {"im_conversations", "im_messages"}: raise ValueError(f"unsupported table name: {table_name}") return int( db.execute(text(f"SELECT COALESCE(MAX(id), 0) + 1 FROM {table_name}")).scalar_one() ) def _get_direct_conversation( db: Session, user_low_id: int, user_high_id: int, ) -> Mapping[str, Any] | None: _, _, pair_key = _build_parent_direct_pair(user_low_id, user_high_id) return ( db.execute( text( """ SELECT id, last_seq, status FROM im_conversations WHERE conversation_type = :conversation_type AND pair_key = :pair_key LIMIT 1 """ ), { "conversation_type": PARENT_DIRECT_CONVERSATION_TYPE, "pair_key": pair_key, }, ) .mappings() .first() ) def _get_or_create_direct_conversation( db: Session, user_low_id: int, user_high_id: int, ) -> int: existing = _get_direct_conversation( db=db, user_low_id=user_low_id, user_high_id=user_high_id, ) if existing: return int(existing["id"]) now_sql = current_timestamp_sql(db) participant_a_id, participant_b_id, pair_key = _build_parent_direct_pair(user_low_id, user_high_id) conversation_id = None if get_db_dialect_name(db) == "sqlite": conversation_id = _next_primary_key(db, "im_conversations") try: insert_sql = f""" INSERT INTO im_conversations ( {'id,' if conversation_id is not None else ''} conversation_type, participant_a_type, participant_a_id, participant_b_type, participant_b_id, pair_key, status, last_seq, message_count, created_at, updated_at ) VALUES ( {':id,' if conversation_id is not None else ''} :conversation_type, :participant_a_type, :participant_a_id, :participant_b_type, :participant_b_id, :pair_key, 1, 0, 0, {now_sql}, {now_sql} ) """ params = { "conversation_type": PARENT_DIRECT_CONVERSATION_TYPE, "participant_a_type": PARENT_PARTICIPANT_TYPE, "participant_a_id": participant_a_id, "participant_b_type": PARENT_PARTICIPANT_TYPE, "participant_b_id": participant_b_id, "pair_key": pair_key, } if conversation_id is not None: params["id"] = conversation_id result = db.execute( text( insert_sql ), params, ) if conversation_id is not None: return conversation_id return inserted_primary_key(result) except IntegrityError: existing = _get_direct_conversation( db=db, user_low_id=user_low_id, user_high_id=user_high_id, ) if existing: return int(existing["id"]) raise def _get_active_parent_profiles( db: Session, *, sender_user_id: int, peer_user_id: int, ) -> dict[int, Mapping[str, Any]]: rows = ( db.execute( text( """ SELECT user_id, nickname, avatar_url FROM parents WHERE user_id IN (:sender_user_id, :peer_user_id) AND status = 1 """ ), {"sender_user_id": sender_user_id, "peer_user_id": peer_user_id}, ) .mappings() .all() ) return {int(row["user_id"]): row for row in rows} def _assert_conversation_access( db: Session, conversation_id: int, current_user_id: int, ) -> None: conversation_row = ( db.execute( text( """ SELECT id, participant_a_type, participant_a_id, participant_b_type, participant_b_id FROM im_conversations WHERE id = :conversation_id LIMIT 1 """ ), {"conversation_id": conversation_id}, ) .mappings() .first() ) if not conversation_row: raise HTTPException(status_code=404, detail="conversation not found") current_user_id_str = str(current_user_id) is_participant = ( ( int(conversation_row["participant_a_type"]) == PARENT_PARTICIPANT_TYPE and conversation_row["participant_a_id"] == current_user_id_str ) or ( int(conversation_row["participant_b_type"]) == PARENT_PARTICIPANT_TYPE and conversation_row["participant_b_id"] == current_user_id_str ) ) if not is_participant: raise HTTPException(status_code=403, detail="no permission for this conversation") @router.post("", response_model=MessageCreateResponse, status_code=status.HTTP_201_CREATED) def create_message( payload: MessageCreateRequest, request: Request, response: Response, current_user_id: int = Depends(get_current_user_id), db: Session = Depends(get_db), ) -> MessageCreateResponse: sender_user_id = current_user_id peer_user_id = payload.peer_user_id if sender_user_id == peer_user_id: raise HTTPException(status_code=422, detail="peer_user_id cannot be same as current user") user_low_id = min(sender_user_id, peer_user_id) user_high_id = max(sender_user_id, peer_user_id) now_sql = current_timestamp_sql(db) with db.begin(): parents = _get_active_parent_profiles( db, sender_user_id=sender_user_id, peer_user_id=peer_user_id, ) if len(parents) != 2: raise HTTPException(status_code=404, detail="sender or peer user not found") conversation_id = _get_or_create_direct_conversation( db=db, user_low_id=user_low_id, user_high_id=user_high_id, ) conversation = ( db.execute( text( f""" SELECT id, last_seq, status FROM im_conversations WHERE id = :conversation_id{select_for_update_clause(db)} """ ), {"conversation_id": conversation_id}, ) .mappings() .first() ) if not conversation: raise HTTPException(status_code=500, detail="failed to load conversation") if int(conversation["status"]) != 1: raise HTTPException(status_code=409, detail="conversation is not active") existing = _get_existing_message(db, conversation_id, payload.client_msg_id) if existing: logger.info( "message idempotent hit", extra={ "event": "message_create", "request_id": getattr(request.state, "request_id", None), "user_id": sender_user_id, "conversation_id": conversation_id, "message_id": int(existing["id"]), "seq": int(existing["seq"]), "client_msg_id": payload.client_msg_id, "idempotent": True, }, ) response.status_code = status.HTTP_200_OK return MessageCreateResponse( idempotent=True, message=_row_to_message_item(existing), ) next_seq = int(conversation["last_seq"]) + 1 preview = _build_preview(payload.content_type, payload.content_text) message_id = None if get_db_dialect_name(db) == "sqlite": message_id = _next_primary_key(db, "im_messages") insert_sql = f""" INSERT INTO im_messages ( {'id,' if message_id is not None else ''} conversation_id, seq, sender_type, sender_id, receiver_type, receiver_id, content_type, content_text, content_json, media_file_key, media_duration_ms, client_msg_id, sender_name_snapshot, sender_avatar_snapshot, receiver_name_snapshot, receiver_avatar_snapshot, created_at ) VALUES ( {':id,' if message_id is not None else ''} :conversation_id, :seq, :sender_type, :sender_id, :receiver_type, :receiver_id, :content_type, :content_text, :content_json, :media_file_key, :media_duration_ms, :client_msg_id, :sender_name_snapshot, :sender_avatar_snapshot, :receiver_name_snapshot, :receiver_avatar_snapshot, {now_sql} ) """ insert_params = { "conversation_id": conversation_id, "seq": next_seq, "sender_type": PARENT_PARTICIPANT_TYPE, "sender_id": str(sender_user_id), "receiver_type": PARENT_PARTICIPANT_TYPE, "receiver_id": str(peer_user_id), "content_type": payload.content_type, "content_text": payload.content_text, "content_json": json.dumps(payload.content_json, ensure_ascii=False) if payload.content_json is not None else None, "media_file_key": payload.media_file_key, "media_duration_ms": payload.media_duration_ms, "client_msg_id": payload.client_msg_id, "sender_name_snapshot": parents[sender_user_id]["nickname"], "sender_avatar_snapshot": parents[sender_user_id]["avatar_url"], "receiver_name_snapshot": parents[peer_user_id]["nickname"], "receiver_avatar_snapshot": parents[peer_user_id]["avatar_url"], } if message_id is not None: insert_params["id"] = message_id insert_result = db.execute( text( insert_sql ), insert_params, ) db.execute( text( f""" UPDATE im_conversations SET last_seq = :last_seq, message_count = message_count + 1, last_message_preview = :last_message_preview, last_message_at = {now_sql}, updated_at = {now_sql} WHERE id = :conversation_id """ ), { "conversation_id": conversation_id, "last_seq": next_seq, "last_message_preview": preview, }, ) created = ( db.execute( text( """ SELECT id, conversation_id, seq, CASE WHEN sender_type = :parent_participant_type THEN sender_id ELSE NULL END AS sender_user_id, 1 AS role, content_type, content_text, content_json, media_file_key, media_duration_ms, client_msg_id, created_at FROM im_messages WHERE id = :message_id LIMIT 1 """ ), { "message_id": message_id if message_id is not None else inserted_primary_key(insert_result), "parent_participant_type": PARENT_PARTICIPANT_TYPE, }, ) .mappings() .first() ) if not created: raise HTTPException(status_code=500, detail="failed to load created message") logger.info( "message created", extra={ "event": "message_create", "request_id": getattr(request.state, "request_id", None), "user_id": sender_user_id, "conversation_id": conversation_id, "message_id": int(created["id"]), "seq": int(created["seq"]), "client_msg_id": payload.client_msg_id, "idempotent": False, }, ) return MessageCreateResponse( idempotent=False, message=_row_to_message_item(created), ) @router.get("", response_model=MessageListResponse) def list_messages( request: Request, conversation_id: int = Query(gt=0), cursor_seq: int | None = Query(default=None, ge=1), limit: int = Query(default=20, ge=1, le=100), current_user_id: int = Depends(get_current_user_id), db: Session = Depends(get_db), ) -> MessageListResponse: _assert_conversation_access( db=db, conversation_id=conversation_id, current_user_id=current_user_id, ) sql = """ SELECT id, conversation_id, seq, CASE WHEN sender_type = :parent_participant_type THEN sender_id ELSE NULL END AS sender_user_id, 1 AS role, content_type, content_text, content_json, media_file_key, media_duration_ms, client_msg_id, created_at FROM im_messages WHERE conversation_id = :conversation_id AND deleted_at IS NULL """ params: dict[str, Any] = { "conversation_id": conversation_id, "fetch_limit": limit + 1, "parent_participant_type": PARENT_PARTICIPANT_TYPE, } if cursor_seq is not None: sql += " AND seq < :cursor_seq" params["cursor_seq"] = cursor_seq sql += " ORDER BY seq DESC LIMIT :fetch_limit" rows = db.execute(text(sql), params).mappings().all() has_more = len(rows) > limit rows = rows[:limit] rows.reverse() items = [_row_to_message_item(row) for row in rows] next_cursor_seq = items[0].seq if has_more and items else None logger.info( "messages listed", extra={ "event": "message_list", "request_id": getattr(request.state, "request_id", None), "user_id": current_user_id, "conversation_id": conversation_id, "cursor_seq": cursor_seq, "limit": limit, "count": len(items), "has_more": has_more, }, ) return MessageListResponse( conversation_id=conversation_id, has_more=has_more, next_cursor_seq=next_cursor_seq, items=items, )