diff --git a/mini-program/app/routers/messages.py b/mini-program/app/routers/messages.py index 4b23553..865bfc8 100644 --- a/mini-program/app/routers/messages.py +++ b/mini-program/app/routers/messages.py @@ -5,11 +5,18 @@ 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, @@ -20,6 +27,12 @@ try: 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, @@ -32,6 +45,9 @@ except ModuleNotFoundError: 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: @@ -61,12 +77,16 @@ def _normalize_content_json(value: Any) -> dict[str, Any] | 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=row["sender_user_id"], - role=int(row["role"]), + 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"]), @@ -88,8 +108,11 @@ def _get_existing_message( id, conversation_id, seq, - sender_user_id, - role, + 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, @@ -97,19 +120,170 @@ def _get_existing_message( media_duration_ms, client_msg_id, created_at - FROM chat_message + 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}, + { + "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, @@ -119,8 +293,13 @@ def _assert_conversation_access( db.execute( text( """ - SELECT id, user_low_id, user_high_id - FROM chat_conversation + SELECT + id, + participant_a_type, + participant_a_id, + participant_b_type, + participant_b_id + FROM im_conversations WHERE id = :conversation_id LIMIT 1 """ @@ -133,10 +312,18 @@ def _assert_conversation_access( if not conversation_row: raise HTTPException(status_code=404, detail="conversation not found") - if current_user_id not in ( - int(conversation_row["user_low_id"]), - int(conversation_row["user_high_id"]), - ): + 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") @@ -155,62 +342,30 @@ def create_message( 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(): - user_count = db.execute( - text( - """ - SELECT COUNT(*) AS cnt - FROM chat_user - WHERE id IN (:sender_user_id, :peer_user_id) - AND status = 1 - """ - ), - {"sender_user_id": sender_user_id, "peer_user_id": peer_user_id}, - ).scalar_one() - - if int(user_count) != 2: + 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") - db.execute( - text( - """ - INSERT INTO chat_conversation ( - user_low_id, - user_high_id, - status, - last_seq, - message_count, - created_at, - updated_at - ) - VALUES ( - :user_low_id, - :user_high_id, - 1, - 0, - 0, - CURRENT_TIMESTAMP(3), - CURRENT_TIMESTAMP(3) - ) - ON DUPLICATE KEY UPDATE - id = LAST_INSERT_ID(id), - updated_at = CURRENT_TIMESTAMP(3) - """ - ), - {"user_low_id": user_low_id, "user_high_id": user_high_id}, + conversation_id = _get_or_create_direct_conversation( + db=db, + user_low_id=user_low_id, + user_high_id=user_high_id, ) - conversation_id = int(db.execute(text("SELECT LAST_INSERT_ID()")).scalar_one()) - conversation = ( db.execute( text( - """ + f""" SELECT id, last_seq, status - FROM chat_conversation - WHERE id = :conversation_id - FOR UPDATE + FROM im_conversations + WHERE id = :conversation_id{select_for_update_clause(db)} """ ), {"conversation_id": conversation_id}, @@ -247,63 +402,92 @@ def create_message( 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_result = db.execute( - text( - """ - INSERT INTO chat_message ( + insert_sql = f""" + INSERT INTO im_messages ( + {'id,' if message_id is not None else ''} conversation_id, seq, - sender_user_id, - role, + 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_user_id, - 1, + :sender_type, + :sender_id, + :receiver_type, + :receiver_id, :content_type, :content_text, :content_json, :media_file_key, :media_duration_ms, :client_msg_id, - CURRENT_TIMESTAMP(3) + :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 ), - { - "conversation_id": conversation_id, - "seq": next_seq, - "sender_user_id": sender_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, - }, + insert_params, ) db.execute( text( - """ - UPDATE chat_conversation + f""" + UPDATE im_conversations SET last_seq = :last_seq, message_count = message_count + 1, last_message_preview = :last_message_preview, - last_message_at = CURRENT_TIMESTAMP(3), - updated_at = CURRENT_TIMESTAMP(3) + last_message_at = {now_sql}, + updated_at = {now_sql} WHERE id = :conversation_id """ ), @@ -322,8 +506,11 @@ def create_message( id, conversation_id, seq, - sender_user_id, - role, + 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, @@ -331,12 +518,15 @@ def create_message( media_duration_ms, client_msg_id, created_at - FROM chat_message + FROM im_messages WHERE id = :message_id LIMIT 1 """ ), - {"message_id": insert_result.lastrowid}, + { + "message_id": message_id if message_id is not None else inserted_primary_key(insert_result), + "parent_participant_type": PARENT_PARTICIPANT_TYPE, + }, ) .mappings() .first() @@ -383,8 +573,11 @@ def list_messages( id, conversation_id, seq, - sender_user_id, - role, + 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, @@ -392,13 +585,14 @@ def list_messages( media_duration_ms, client_msg_id, created_at - FROM chat_message + 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"