229 lines
7.6 KiB
Python
229 lines
7.6 KiB
Python
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
|