# retoor <retoor@molodetz.nl>

import asyncio
import logging
from datetime import datetime
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,
    _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 ConversationOut, 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.moderation.screening import ContentRefused
from devplacepy.services.messaging import (
    issue_ticket,
    message_frame,
    message_hub,
    message_relay,
    persist_message,
    redeem_ticket,
    stamp_content_revision,
)

logger = logging.getLogger(__name__)
router = APIRouter()

MAX_WS_ATTACHMENTS = 5

CONVERSATION_MESSAGE_LIMIT = 500

MESSAGE_GROUP_GAP_SECONDS = 300

def _grouped_with_previous(sender_uid, created_at, previous_sender_uid, previous_created_at) -> bool:
    if previous_sender_uid is None or sender_uid != previous_sender_uid:
        return False
    try:
        current_dt = datetime.fromisoformat(created_at)
        previous_dt = datetime.fromisoformat(previous_created_at)
    except (TypeError, ValueError):
        return False
    return (current_dt - previous_dt).total_seconds() <= MESSAGE_GROUP_GAP_SECONDS

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, before: 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)
    before = (before or "").strip()
    if before:
        msgs = list(
            db.query(
                "SELECT * FROM messages"
                " WHERE ((sender_uid = :me AND receiver_uid = :other)"
                "    OR (sender_uid = :other AND receiver_uid = :me))"
                "   AND created_at < :before"
                " ORDER BY created_at DESC, id DESC LIMIT :lim",
                me=user_uid,
                other=other_uid,
                before=before,
                lim=CONVERSATION_MESSAGE_LIMIT,
            )
        )
    else:
        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 {}
    previous_sender_uid = None
    previous_created_at = None
    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"], []),
                "grouped": _grouped_with_previous(
                    m["sender_uid"], m["created_at"], previous_sender_uid, previous_created_at
                ),
            }
        )
        previous_sender_uid = m["sender_uid"]
        previous_created_at = m["created_at"]
    return result, other_user

@router.get("", response_class=HTMLResponse)
async def messages_page(
    request: Request, with_uid: str = None, search: str = "", before: 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, before=before
        )
        current_conversation = with_uid
        other_online = presence.is_online(other_user)
        other_last_seen = other_user.get("last_seen") if other_user else None
        if not before:
            mark_conversation_read(user["uid"], with_uid)
            mark_notifications_read_by_target(
                user["uid"], f"/messages?with_uid={with_uid}"
            )
            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.get("/conversations")
async def list_conversations(request: Request):
    user = require_user(request)
    conversations = get_conversations(user["uid"])
    payload = [ConversationOut.model_validate(c).model_dump() for c in conversations]
    return JSONResponse({"conversations": payload})

@router.post("/ws-ticket")
async def create_ws_ticket(request: Request):
    user = require_user(request)
    token = issue_ticket(user["uid"])
    return JSONResponse({"ticket": token, "expires_in": 30})

@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")

    ai_processed = await _finalize_and_broadcast(
        user, message, request, client_id=data.client_id, wait_ai=True
    )
    frame = message_frame(
        message, user.get("username", ""), data.client_id,
        sender_role=user.get("role"), ai_processed=ai_processed,
    )
    return action_result(
        request, f"/messages?with_uid={receiver_uid}", data=frame
    )

async def broadcast_message(
    sender: dict, message: dict, client_id: Optional[str] = None,
    ai_processed: bool = False, ai_pending: bool = False,
) -> None:
    frame = message_frame(
        message, sender.get("username", ""), client_id,
        sender_role=sender.get("role"), ai_processed=ai_processed,
    )
    frame["ai_pending"] = ai_pending
    if not ai_processed:
        message_hub.mark_delivered(message["uid"])
    targets = [message["sender_uid"], message["receiver_uid"]]
    await message_hub.send_to_users(targets, frame)

async def _await_ai_and_push(
    sender: dict, message: dict, pending: list, client_id: Optional[str]
) -> bool:
    await asyncio.gather(*pending, return_exceptions=True)
    pending.clear()
    row = get_table("messages").find_one(uid=message["uid"])
    if not row:
        return False
    changed = row.get("content") != message.get("content")
    if changed:
        message["content"] = row["content"]
        stamp_content_revision(message["uid"])
    await broadcast_message(sender, message, client_id, ai_processed=True)
    return changed

async def _finalize_and_broadcast(
    sender: dict,
    message: dict,
    request: object,
    client_id: Optional[str] = None,
    wait_ai: bool = True,
) -> bool:
    scope = getattr(request, "scope", None)
    pending = scope.get(PENDING_SCOPE_KEY) if scope is not None else None
    has_pending = bool(pending)
    await broadcast_message(
        sender, message, client_id, ai_processed=False, ai_pending=has_pending
    )
    if not pending:
        return False
    if wait_ai:
        return await _await_ai_and_push(sender, message, pending, client_id)
    asyncio.create_task(_await_ai_and_push(sender, message, pending, client_id))
    return False

SYNC_LIMIT = 200


async def _sync_missed(user_uid: str, data: dict, websocket: WebSocket) -> None:
    since = str(data.get("since") or "").strip()
    with_uid = str(data.get("with_uid") or "").strip()
    if not since or "messages" not in db.tables:
        return
    if with_uid:
        rows = list(
            db.query(
                "SELECT * FROM messages"
                " WHERE ((sender_uid = :me AND receiver_uid = :other)"
                "    OR (sender_uid = :other AND receiver_uid = :me))"
                "   AND (created_at > :since"
                "        OR (updated_at IS NOT NULL AND updated_at > :since))"
                " ORDER BY created_at ASC, id ASC LIMIT :lim",
                me=user_uid,
                other=with_uid,
                since=since,
                lim=SYNC_LIMIT,
            )
        )
    else:
        rows = list(
            db.query(
                "SELECT * FROM messages"
                " WHERE (sender_uid = :me OR receiver_uid = :me)"
                "   AND (created_at > :since"
                "        OR (updated_at IS NOT NULL AND updated_at > :since))"
                " ORDER BY created_at ASC, id ASC LIMIT :lim",
                me=user_uid,
                since=since,
                lim=SYNC_LIMIT,
            )
        )
    if not rows:
        return
    sender_uids = {row["sender_uid"] for row in rows}
    senders = get_users_by_uids(list(sender_uids)) if sender_uids else {}
    for row in rows:
        sender = senders.get(row["sender_uid"]) or {}
        frame = message_frame(
            dict(row),
            sender.get("username", ""),
            sender_role=sender.get("role"),
            ai_processed=bool(row.get("updated_at")),
        )
        try:
            await websocket.send_json(frame)
        except Exception:  # noqa: BLE001
            logger.debug("sync frame dropped for %s", user_uid)
            return


def _resolve_ws_user(websocket: WebSocket):
    user = _user_from_session(websocket)
    if user:
        return user
    ticket = websocket.query_params.get("ticket", "").strip()
    if ticket:
        user_uid = redeem_ticket(ticket)
        if user_uid:
            return get_table("users").find_one(uid=user_uid)
    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


def _ws_may_write(user: dict) -> bool:
    from devplacepy.database import suspension_active
    from devplacepy.routers.auth.terms import needs_acceptance

    return not suspension_active(user) and not needs_acceptance(user)

@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
    if not _ws_may_write(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
                try:
                    message = persist_message(
                        user,
                        receiver_uid,
                        content,
                        attachment_uids,
                        request=websocket,
                        origin="websocket",
                    )
                except ContentRefused as exc:
                    await websocket.send_json(
                        {
                            "type": "error",
                            "client_id": client_id,
                            "text": exc.message,
                        }
                    )
                    continue
                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, wait_ai=False
                )
            elif kind == "sync":
                await _sync_missed(user_uid, data, websocket)
            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)
