's'

backend/dm.py

@@ -0,0 +1,228 @@
+from __future__ import annotations
+
+from datetime import datetime, timezone
+from typing import Any
+
+from .pb_client import pb, pb_quote
+
+MAX_MESSAGE_LENGTH = 2000
+
+
+def utc_now() -> str:
+    return datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S.000Z")
+
+
+def pair_key_for_users(user_a_id: str, user_b_id: str) -> str:
+    return "|".join(sorted([user_a_id, user_b_id]))
+
+
+def ordered_user_ids(user_a_id: str, user_b_id: str) -> tuple[str, str]:
+    return tuple(sorted([user_a_id, user_b_id]))  # type: ignore[return-value]
+
+
+def public_record(record: dict[str, Any]) -> dict[str, Any]:
+    return {k: v for k, v in record.items() if not k.startswith("@")}
+
+
+def conversation_participants(conversation: dict[str, Any]) -> set[str]:
+    return {str(conversation.get("userAId") or ""), str(conversation.get("userBId") or "")}
+
+
+def other_participant_id(conversation: dict[str, Any], user_id: str) -> str | None:
+    participants = conversation_participants(conversation)
+    participants.discard("")
+    participants.discard(user_id)
+    return next(iter(participants), None)
+
+
+def read_state_for(conversation: dict[str, Any]) -> dict[str, str]:
+    raw = conversation.get("readState")
+    if isinstance(raw, dict):
+        return {str(key): str(value) for key, value in raw.items() if value}
+    return {}
+
+
+async def ensure_conversation_for_match(
+    *,
+    user_a_id: str,
+    user_b_id: str,
+    match_connection_id: str = "",
+    intent: str = "friends",
+) -> dict[str, Any]:
+    pair_key = pair_key_for_users(user_a_id, user_b_id)
+    existing = await pb.first_record("conversations", filter=f'pairKey={pb_quote(pair_key)}')
+    if existing:
+        updates: dict[str, Any] = {}
+        if match_connection_id and not existing.get("matchConnectionId"):
+            updates["matchConnectionId"] = match_connection_id
+        if intent and not existing.get("intent"):
+            updates["intent"] = intent
+        if updates:
+            return await pb.update_record("conversations", existing["id"], updates)
+        return existing
+
+    user_a, user_b = ordered_user_ids(user_a_id, user_b_id)
+    return await pb.create_record(
+        "conversations",
+        {
+            "pairKey": pair_key,
+            "type": "match",
+            "userAId": user_a,
+            "userBId": user_b,
+            "matchConnectionId": match_connection_id,
+            "intent": intent,
+            "lastMessageAt": "",
+            "lastMessagePreview": "",
+            "readState": {},
+        },
+    )
+
+
+async def sync_connection_conversation(connection: dict[str, Any]) -> dict[str, Any]:
+    conversation = await ensure_conversation_for_match(
+        user_a_id=str(connection.get("userAId") or ""),
+        user_b_id=str(connection.get("userBId") or ""),
+        match_connection_id=str(connection.get("id") or ""),
+        intent=str(connection.get("intent") or "friends"),
+    )
+    connection_id = str(connection.get("id") or "")
+    conversation_id = str(conversation.get("id") or "")
+    if connection_id and conversation_id and str(connection.get("conversationId") or "") != conversation_id:
+        connection = await pb.update_record(
+            "match_connections",
+            connection_id,
+            {"conversationId": conversation_id},
+        )
+    if connection_id and str(conversation.get("matchConnectionId") or "") != connection_id:
+        conversation = await pb.update_record(
+            "conversations",
+            conversation_id,
+            {"matchConnectionId": connection_id},
+        )
+    return conversation
+
+
+async def backfill_conversations_for_user(user_id: str) -> None:
+    data = await pb.list_records(
+        "match_connections",
+        filter=f'userAId={pb_quote(user_id)} || userBId={pb_quote(user_id)}',
+        per_page=100,
+        sort="-created",
+    )
+    for connection in data.get("items", []):
+        await sync_connection_conversation(connection)
+
+
+async def conversation_for_user(conversation_id: str, user_id: str) -> dict[str, Any] | None:
+    conversation = await pb.first_record("conversations", filter=f'id={pb_quote(conversation_id)}')
+    if not conversation:
+        return None
+    if user_id not in conversation_participants(conversation):
+        return None
+    return conversation
+
+
+async def conversation_by_connection(connection_id: str, user_id: str) -> dict[str, Any] | None:
+    connection = await pb.first_record("match_connections", filter=f'id={pb_quote(connection_id)}')
+    if not connection:
+        return None
+    if user_id not in {str(connection.get("userAId") or ""), str(connection.get("userBId") or "")}:
+        return None
+    existing_id = str(connection.get("conversationId") or "")
+    if existing_id:
+        conversation = await conversation_for_user(existing_id, user_id)
+        if conversation:
+            return conversation
+    return await sync_connection_conversation(connection)
+
+
+def unread_count_for(conversation: dict[str, Any], user_id: str) -> int:
+    last_read = read_state_for(conversation).get(user_id, "")
+    preview = str(conversation.get("lastMessagePreview") or "").strip()
+    if not preview:
+        return 0
+    last_message_at = str(conversation.get("lastMessageAt") or "")
+    if not last_message_at:
+        return 0
+    if not last_read:
+        return 1
+    return 1 if last_message_at > last_read else 0
+
+
+async def list_conversations_for_user(user_id: str) -> list[dict[str, Any]]:
+    await backfill_conversations_for_user(user_id)
+    data = await pb.list_records(
+        "conversations",
+        filter=f'userAId={pb_quote(user_id)} || userBId={pb_quote(user_id)}',
+        per_page=100,
+        sort="-lastMessageAt,-updated,-created",
+    )
+    return list(data.get("items", []))
+
+
+async def list_messages(conversation_id: str, *, limit: int = 50, before: str = "") -> list[dict[str, Any]]:
+    safe_limit = max(1, min(limit, 100))
+    filter_parts = [f'conversationId={pb_quote(conversation_id)}']
+    if before:
+        filter_parts.append(f'created<{pb_quote(before)}')
+    data = await pb.list_records(
+        "messages",
+        filter=" && ".join(filter_parts),
+        per_page=safe_limit,
+        sort="-created",
+    )
+    items = list(data.get("items", []))
+    items.reverse()
+    return items
+
+
+async def mark_conversation_read(conversation: dict[str, Any], user_id: str) -> dict[str, Any]:
+    read_state = read_state_for(conversation)
+    read_state[user_id] = utc_now()
+    return await pb.update_record(
+        "conversations",
+        conversation["id"],
+        {"readState": read_state},
+    )
+
+
+async def send_message(
+    *,
+    conversation: dict[str, Any],
+    sender_id: str,
+    body: str,
+) -> tuple[dict[str, Any], str | None]:
+    text = body.strip()
+    if not text:
+        raise ValueError("消息不能为空")
+    if len(text) > MAX_MESSAGE_LENGTH:
+        raise ValueError(f"消息不能超过 {MAX_MESSAGE_LENGTH} 字")
+    if sender_id not in conversation_participants(conversation):
+        raise ValueError("无权发送消息")
+
+    preview = text.replace("\n", " ")
+    if len(preview) > 180:
+        preview = f"{preview[:177]}..."
+
+    message = await pb.create_record(
+        "messages",
+        {
+            "conversationId": conversation["id"],
+            "senderId": sender_id,
+            "body": text,
+        },
+    )
+    message_at = str(message.get("created") or utc_now())
+    read_state = read_state_for(conversation)
+    read_state[sender_id] = message_at
+    await pb.update_record(
+        "conversations",
+        conversation["id"],
+        {
+            "lastMessageAt": message_at,
+            "lastMessagePreview": preview,
+            "readState": read_state,
+        },
+    )
+    recipient_id = other_participant_id(conversation, sender_id)
+    return message, recipient_id