# retoor <retoor@molodetz.nl>
import asyncio
import logging
from typing import Annotated, Optional
from fastapi import Depends, APIRouter, Request, WebSocket, WebSocketDisconnect
from devplacepy.models import MessageForm
from fastapi.responses import HTMLResponse, JSONResponse
from devplacepy.database import (
get_table,
db,
get_users_by_uids,
search_users_by_username,
get_blocked_uids,
mark_notifications_read_by_target,
)
from devplacepy.attachments import get_attachments_batch
from devplacepy.templating import clear_messages_cache
from devplacepy.utils import (
require_user,
time_ago,
is_admin,
_user_from_session,
_user_from_api_key,
)
from devplacepy.seo import base_seo_context
from devplacepy.responses import respond, action_result
from devplacepy.schemas import MessagesOut
from devplacepy.services import presence
from devplacepy.services.audit import record as audit
from devplacepy.services.correction import PENDING_SCOPE_KEY
from devplacepy.dependencies import json_or_form
from devplacepy.services.messaging import (
message_frame,
message_hub,
message_relay,
persist_message,
)
logger = logging.getLogger(__name__)
router = APIRouter()
MAX_WS_ATTACHMENTS = 5
CONVERSATION_MESSAGE_LIMIT = 500
def mark_conversation_read(user_uid: str, other_uid: str) -> None:
if "messages" not in db.tables:
return
with db:
db.query(
"UPDATE messages SET read = 1 WHERE receiver_uid = :me AND sender_uid = :other AND read = 0",
me=user_uid,
other=other_uid,
)
clear_messages_cache(user_uid)
def get_conversations(user_uid: str):
if "messages" not in db.tables:
return []
latest = list(
db.query(
"SELECT * FROM ("
" SELECT *,"
" CASE WHEN sender_uid = :me THEN receiver_uid ELSE sender_uid END AS other_uid,"
" ROW_NUMBER() OVER ("
" PARTITION BY CASE WHEN sender_uid = :me THEN receiver_uid ELSE sender_uid END"
" ORDER BY created_at DESC, id DESC"
" ) AS rn"
" FROM messages"
" WHERE sender_uid = :me OR receiver_uid = :me"
") WHERE rn = 1 ORDER BY created_at DESC",
me=user_uid,
)
)
blocked = get_blocked_uids(user_uid)
conversations = []
other_uids = []
for msg in latest:
other_uid = msg["other_uid"]
if other_uid in blocked:
continue
other_uids.append(other_uid)
conversations.append(
{
"other_uid": other_uid,
"other_user": None,
"last_message": msg["content"],
"last_message_at": msg["created_at"],
"unread": msg["receiver_uid"] == user_uid and not msg["read"],
}
)
if other_uids:
users_map = get_users_by_uids(other_uids)
for conv in conversations:
conv["other_user"] = users_map.get(conv["other_uid"])
for conv in conversations:
conv.pop("other_uid", None)
return conversations
def get_conversation_messages(user_uid: str, other_uid: str):
if other_uid in get_blocked_uids(user_uid):
return [], None
if "messages" not in db.tables:
return [], get_users_by_uids([other_uid]).get(other_uid)
msgs = list(
db.query(
"SELECT * FROM messages"
" WHERE (sender_uid = :me AND receiver_uid = :other)"
" OR (sender_uid = :other AND receiver_uid = :me)"
" ORDER BY created_at DESC, id DESC LIMIT :lim",
me=user_uid,
other=other_uid,
lim=CONVERSATION_MESSAGE_LIMIT,
)
)
msgs.reverse()
user_ids = list({m["sender_uid"] for m in msgs} | {other_uid})
users_map = get_users_by_uids(user_ids)
other_user = users_map.get(other_uid)
result = []
msg_uids = [m["uid"] for m in msgs]
attachments_map = get_attachments_batch("message", msg_uids) if msg_uids else {}
for m in msgs:
result.append(
{
"message": m,
"sender": users_map.get(m["sender_uid"]),
"is_mine": m["sender_uid"] == user_uid,
"time_ago": time_ago(m["created_at"]),
"attachments": attachments_map.get(m["uid"], []),
}
)
return result, other_user
@router.get("", response_class=HTMLResponse)
async def messages_page(request: Request, with_uid: str = None, search: str = ""):
user = require_user(request)
conversations = get_conversations(user["uid"])
if search:
users_table = get_table("users")
found = users_table.find_one(username=search)
if found:
with_uid = found["uid"]
current_conversation = None
messages = []
other_user = None
other_online = False
other_last_seen = None
if with_uid:
messages, other_user = get_conversation_messages(user["uid"], with_uid)
mark_conversation_read(user["uid"], with_uid)
mark_notifications_read_by_target(
user["uid"], f"/messages?with_uid={with_uid}"
)
current_conversation = with_uid
other_online = presence.is_online(other_user)
other_last_seen = other_user.get("last_seen") if other_user else None
audit.record(
request,
"message.read_on_view",
user=user,
target_type="user",
target_uid=with_uid,
target_label=other_user.get("username") if other_user else None,
metadata={"message_count": len(messages)},
summary=f"{user['username']} read messages from {other_user.get('username') if other_user else with_uid}",
links=[audit.target("user", with_uid, other_user.get("username") if other_user else None)],
)
seo_ctx = base_seo_context(
request,
title="Messages",
robots="noindex,nofollow",
breadcrumbs=[
{"name": "Home", "url": "/feed"},
{"name": "Messages", "url": "/messages"},
],
)
return respond(
request,
"messages.html",
{
**seo_ctx,
"request": request,
"user": user,
"conversations": conversations,
"messages": messages,
"other_user": other_user,
"current_conversation": current_conversation,
"search": search,
"other_online": other_online,
"other_last_seen": other_last_seen,
},
model=MessagesOut,
)
@router.get("/search")
async def search_users(request: Request, q: str = ""):
user = require_user(request)
if not q or len(q) < 1:
return JSONResponse({"results": []})
results = search_users_by_username(q, exclude_uid=user["uid"])
return JSONResponse({"results": results})
@router.post("/send")
async def send_message(request: Request, data: Annotated[MessageForm, Depends(json_or_form(MessageForm))]):
user = require_user(request)
receiver_uid = data.receiver_uid
message = persist_message(
user,
receiver_uid,
data.content,
data.attachment_uids,
request=request,
origin="web",
)
if message is None:
return action_result(request, "/messages")
await _finalize_and_broadcast(user, message, request)
return action_result(
request, f"/messages?with_uid={receiver_uid}", data={"uid": message["uid"]}
)
async def broadcast_message(
sender: dict, message: dict, client_id: Optional[str] = None
) -> None:
frame = message_frame(message, sender.get("username", ""), client_id, sender_role=sender.get("role"))
message_hub.mark_delivered(message["uid"])
targets = [message["sender_uid"], message["receiver_uid"]]
await message_hub.send_to_users(targets, frame)
async def _finalize_and_broadcast(
sender: dict, message: dict, request: object, client_id: Optional[str] = None
) -> None:
message_hub.mark_delivered(message["uid"])
scope = getattr(request, "scope", None)
pending = scope.get(PENDING_SCOPE_KEY) if scope is not None else None
if pending:
await asyncio.gather(*pending, return_exceptions=True)
pending.clear()
row = get_table("messages").find_one(uid=message["uid"])
if row:
message["content"] = row["content"]
await broadcast_message(sender, message, client_id)
def _resolve_ws_user(websocket: WebSocket):
user = _user_from_session(websocket)
if user:
return user
key = websocket.headers.get("x-api-key", "").strip()
if not key:
scheme, _, credentials = websocket.headers.get("authorization", "").partition(
" "
)
if scheme.lower() == "bearer":
key = credentials.strip()
if key:
return _user_from_api_key(key)
return None
@router.websocket("/ws")
async def messages_ws(websocket: WebSocket):
await websocket.accept()
user = _resolve_ws_user(websocket)
if not user:
await websocket.close(code=1008)
return
user_uid = user["uid"]
message_hub.register(user_uid, websocket)
message_relay.start()
try:
await websocket.send_json({"type": "ready", "user_uid": user_uid})
while True:
data = await websocket.receive_json()
kind = data.get("type")
if kind == "send":
receiver_uid = str(data.get("receiver_uid", "")).strip()
content = str(data.get("content", ""))
client_id = data.get("client_id")
raw_attachments = data.get("attachment_uids") or []
attachment_uids = [str(a) for a in raw_attachments][:MAX_WS_ATTACHMENTS]
if not receiver_uid:
continue
message = persist_message(
user,
receiver_uid,
content,
attachment_uids,
request=websocket,
origin="websocket",
)
if message is None:
await websocket.send_json(
{"type": "error", "client_id": client_id, "text": "Message not sent."}
)
continue
await _finalize_and_broadcast(user, message, websocket, client_id)
elif kind == "typing":
receiver_uid = str(data.get("receiver_uid", "")).strip()
if receiver_uid:
await message_hub.send_to_user(
receiver_uid, {"type": "typing", "from_uid": user_uid}
)
elif kind == "read":
with_uid = str(data.get("with_uid", "")).strip()
if with_uid:
mark_conversation_read(user_uid, with_uid)
other_user = get_users_by_uids([with_uid]).get(with_uid)
other_username = other_user.get("username") if other_user else None
audit.record(
websocket,
"message.read_on_view",
user=user,
target_type="user",
target_uid=with_uid,
target_label=other_username,
summary=f"{user['username']} read messages from {other_username or with_uid}",
links=[audit.target("user", with_uid, other_username)],
)
await message_hub.send_to_user(
with_uid, {"type": "read", "by_uid": user_uid}
)
except WebSocketDisconnect:
pass
except Exception: # noqa: BLE001
logger.exception("messages websocket loop failed for %s", user_uid)
finally:
message_hub.unregister(user_uid, websocket)
Reference in New Issue View Git Blame Copy Permalink