"""Thread listing and detail endpoints (SPEC §9.1 Mail)."""

from __future__ import annotations

import uuid
from collections.abc import Collection
from typing import Annotated, Any

from fastapi import APIRouter, Depends, Query
from sqlalchemy import Text, func, or_, select
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.sql.elements import ColumnElement

from app import subscriptions as subs
from app.api.errors import raise_api_error
from app.api.schemas import (
    Address,
    MessageEnvelope,
    ThreadDetail,
    ThreadListOut,
    ThreadSummary,
)
from app.api.settings import allowed_image_senders, blocks_trackers
from app.db import get_session
from app.identity.deps import CurrentUser
from app.models import BodyPart, Mailbox, Message, SubscriptionState, Thread, User

router = APIRouter(prefix="/threads", tags=["mail"])

DbSession = Annotated[AsyncSession, Depends(get_session)]


def _addresses(raw: list[dict[str, str]]) -> list[Address]:
    return [Address(name=a.get("name", ""), email=a.get("email", "")) for a in raw]


def _summary(
    thread: Thread, message_count: int | None = None, senders: list[Address] | None = None
) -> ThreadSummary:
    return ThreadSummary(
        id=thread.id,
        subject=thread.subject_norm or "(no subject)",
        snippet=thread.snippet,
        participants=_addresses(thread.participants),
        senders=senders or [],
        account_ids=list(thread.account_ids),
        # Count only the messages this view actually shows (deleted/spam excluded
        # elsewhere); falls back to the denormalized total.
        message_count=message_count if message_count is not None else thread.message_count,
        unread_count=thread.unread_count,
        has_attachments=thread.has_attachments,
        pinned=thread.pinned,
        last_message_at=thread.last_message_at,
        snoozed_until=thread.snoozed_until,
    )


def _envelope(msg: Message, allowed_senders: Collection[str] = ()) -> MessageEnvelope:
    sender = msg.from_addr[0].get("email", "").lower() if msg.from_addr else ""
    return MessageEnvelope(
        id=msg.id,
        account_id=msg.account_id,
        mailbox_id=msg.mailbox_id,
        thread_id=msg.thread_id,
        subject=msg.subject,
        from_addr=_addresses(msg.from_addr),
        to_addr=_addresses(msg.to_addr),
        cc_addr=_addresses(msg.cc_addr),
        bcc_addr=_addresses(msg.bcc_addr),
        date=msg.date_hdr or msg.internal_date,
        flags=list(msg.flags),
        is_unread="\\Seen" not in msg.flags,
        is_starred="\\Flagged" in msg.flags,
        has_attachments=msg.has_attachments,
        has_remote_images=msg.has_remote_images,
        images_allowed=bool(sender) and sender in allowed_senders,
        size_bytes=msg.size_bytes,
        snippet=msg.snippet,
        content_language=msg.content_language or "",
    )


@router.get("", response_model=ThreadListOut)
async def list_threads(
    user: CurrentUser,
    db: DbSession,
    view: str = Query("unified"),
    limit: int = Query(50, ge=1, le=200),
    offset: int = Query(0, ge=0),
    filter: str = Query("all"),  # all | unread | pinned
    order: str = Query("desc"),  # desc (newest first) | asc (oldest first)
    ids: str = Query(""),  # comma-separated thread ids (e.g. a plugin's label view)
) -> ThreadListOut:
    # Collect WHERE conditions once so the page query and the total-count query
    # stay in lockstep (offset pagination needs an accurate grand total).
    conditions: list[ColumnElement[bool]] = [Thread.user_id == user.id]

    if ids:
        # Explicit id set (ignores view/folder) — used by plugin-driven views like
        # "all mail with label X", which supply the matching thread ids.
        try:
            wanted = [uuid.UUID(x) for x in ids.split(",") if x.strip()]
        except ValueError:
            raise_api_error(400, "bad_ids", "invalid thread id in ids")
        conditions.append(Thread.id.in_(wanted or [uuid.uuid4()]))
    elif view.startswith("account:"):
        # A single account's mail: threads with a message in ANY of THIS account's
        # mailboxes except Trash/Spam (those are only seen via their own folders).
        try:
            account_id = uuid.UUID(view.split(":", 1)[1])
        except ValueError:
            raise_api_error(400, "bad_view", "invalid account id in view")
        in_account = (
            select(Message.thread_id)
            .join(Mailbox, Mailbox.id == Message.mailbox_id)
            .where(
                Message.user_id == user.id,
                Message.account_id == account_id,
                or_(
                    Mailbox.special_use.is_(None),
                    Mailbox.special_use.notin_(_REMOVAL_FOLDERS),
                ),
            )
        )
        conditions.append(Thread.id.in_(in_account))
    elif view.startswith("folder:"):
        # Threads with at least one message in a mailbox of this special_use,
        # across all of the user's accounts (unified per-folder view).
        special = view.split(":", 1)[1]
        in_folder = (
            select(Message.thread_id)
            .join(Mailbox, Mailbox.id == Message.mailbox_id)
            .where(Message.user_id == user.id, Mailbox.special_use == special)
        )
        conditions.append(Thread.id.in_(in_folder))
        if special == "inbox":
            conditions.append(Thread.snoozed_until.is_(None))  # back when it wakes
    elif view == "snoozed":
        conditions.append(Thread.snoozed_until.isnot(None))
    elif view != "unified":
        raise_api_error(400, "bad_view", "unsupported view")

    if filter == "unread":
        conditions.append(Thread.unread_count > 0)
    elif filter == "pinned":
        conditions.append(Thread.pinned.is_(True))

    total = (
        await db.scalar(
            select(func.count()).select_from(select(Thread.id).where(*conditions).subquery())
        )
    ) or 0

    # A thread woken from snooze sorts as if it had just arrived.
    sort_at = func.greatest(
        Thread.last_message_at, func.coalesce(Thread.unsnoozed_at, Thread.last_message_at)
    )
    if view == "snoozed":
        # Soonest back first, whatever the list order.
        order_by: tuple[ColumnElement[Any], ...] = (
            Thread.snoozed_until.asc(),
            Thread.id.asc(),
        )
    elif order == "asc":
        order_by = (sort_at.asc(), Thread.id.asc())
    else:
        order_by = (sort_at.desc(), Thread.id.desc())
    stmt = select(Thread).where(*conditions).order_by(*order_by).offset(offset).limit(limit)
    threads = list(await db.scalars(stmt))

    # Per-thread visible message count matching what get_thread shows for this view:
    # Trash/Spam count only their own messages; every other view excludes them. This
    # keeps the list count in step with a single deleted message.
    current_special = view.split(":", 1)[1] if view.startswith("folder:") else None
    counts: dict[uuid.UUID, int] = {}
    if threads:
        # count DISTINCT deliveries so collapsed cross-account copies count once,
        # matching what get_thread renders.
        delivery = func.coalesce(Message.delivery_id, func.cast(Message.id, Text))
        count_stmt = (
            select(Message.thread_id, func.count(func.distinct(delivery)))
            .join(Mailbox, Mailbox.id == Message.mailbox_id)
            .where(Message.user_id == user.id, Message.thread_id.in_([t.id for t in threads]))
            .group_by(Message.thread_id)
        )
        count_stmt = count_stmt.where(_visible_in(current_special))
        counts = {r[0]: r[1] for r in (await db.execute(count_stmt)).all() if r[0] is not None}
    senders = await thread_senders(db, user.id, [t.id for t in threads], current_special)

    return ThreadListOut(
        threads=[_summary(t, counts.get(t.id), senders.get(t.id)) for t in threads],
        total=total,
    )


async def _mark_subscriptions(
    db: AsyncSession,
    user_id: uuid.UUID,
    messages: list[Message],
    envelopes: list[MessageEnvelope],
) -> None:
    """Tell the reading pane which messages can be unsubscribed from, and how."""
    keys: dict[uuid.UUID, str] = {}
    for m, env in zip(messages, envelopes, strict=True):
        if not m.list_unsubscribe:
            continue
        targets = subs.unsubscribe_targets(m.list_unsubscribe)
        if m.list_unsubscribe_post and targets.https:
            env.unsubscribe_via = "one_click"
        elif targets.mailto:
            env.unsubscribe_via = "mailto"
        elif targets.https:
            env.unsubscribe_via = "web"
        else:
            continue
        sender = m.from_addr[0].get("email", "") if m.from_addr else ""
        env.subscription_key = keys[m.id] = subs.group_key(m.list_id, sender)
    if keys:
        done = set(
            await db.scalars(
                select(SubscriptionState.key).where(
                    SubscriptionState.user_id == user_id,
                    SubscriptionState.key.in_(set(keys.values())),
                )
            )
        )
        for env in envelopes:
            env.unsubscribed = env.subscription_key in done


def _visible_in(current_special: str | None) -> ColumnElement[bool]:
    """Which of a thread's messages a view shows (see ``_REMOVAL_FOLDERS``): Trash/Spam
    show only their own messages; every other view hides them."""
    if current_special in _REMOVAL_FOLDERS:
        return Mailbox.special_use == current_special
    return or_(Mailbox.special_use.is_(None), Mailbox.special_use.notin_(_REMOVAL_FOLDERS))


async def thread_senders(
    db: AsyncSession,
    user_id: uuid.UUID,
    thread_ids: list[uuid.UUID],
    current_special: str | None = None,
) -> dict[uuid.UUID, list[Address]]:
    """Distinct From addresses per thread, oldest message first, over the messages
    the view shows. Recipients are left out: the list's sender column names who
    wrote, not everyone copied."""
    if not thread_ids:
        return {}
    rows = await db.execute(
        select(Message.thread_id, Message.from_addr)
        .join(Mailbox, Mailbox.id == Message.mailbox_id)
        .where(
            Message.user_id == user_id,
            Message.thread_id.in_(thread_ids),
            _visible_in(current_special),
        )
        .order_by(Message.internal_date.asc(), Message.id.asc())
    )
    senders: dict[uuid.UUID, list[Address]] = {}
    seen: dict[uuid.UUID, set[str]] = {}
    for thread_id, from_addr in rows.all():
        if thread_id is None:
            continue
        for addr in from_addr:
            email = (addr.get("email") or "").lower()
            if email and email not in seen.setdefault(thread_id, set()):
                seen[thread_id].add(email)
                senders.setdefault(thread_id, []).append(
                    Address(name=addr.get("name", ""), email=addr.get("email", ""))
                )
    return senders


async def _resolve_thread(db: AsyncSession, user: User, thread_id: uuid.UUID) -> Thread:
    thread = await db.scalar(
        select(Thread).where(Thread.id == thread_id, Thread.user_id == user.id)
    )
    if thread is None:
        raise_api_error(404, "not_found", "thread not found")
    return thread


# Trash/Spam are message-scoped folders: an individually deleted message lives
# there on its own. Viewing a thread from Trash/Spam shows ONLY the messages filed
# there; every other view shows the conversation with trashed/spam messages hidden.
_REMOVAL_FOLDERS = frozenset({"trash", "junk"})


@router.get("/{thread_id}", response_model=ThreadDetail)
async def get_thread(
    thread_id: uuid.UUID,
    user: CurrentUser,
    db: DbSession,
    sort: str = Query("asc"),
    view: str = Query("unified"),
) -> ThreadDetail:
    thread = await _resolve_thread(db, user, thread_id)
    current_special = view.split(":", 1)[1] if view.startswith("folder:") else None
    # Messages are always returned ASC (oldest first); the client reverses for
    # display per the requested order (invariant I-5 / §9.4).
    stmt = (
        select(Message)
        .join(Mailbox, Mailbox.id == Message.mailbox_id)
        .where(Message.thread_id == thread_id, Message.user_id == user.id)
        .order_by(Message.internal_date.asc(), Message.id.asc())
    )
    # Trash/Spam: only the messages actually filed there. Conversation view: hide
    # messages that were individually trashed or spammed.
    stmt = stmt.where(_visible_in(current_special))
    messages = list(await db.scalars(stmt))

    # Collapse duplicate deliveries: the same email (same delivery_id) that landed
    # in more than one of the user's accounts shows once, noting the other accounts.
    # Messages without a real Message-ID (delivery_id is NULL) never collapse.
    reps: dict[str, Message] = {}
    also: dict[str, list[uuid.UUID]] = {}
    ordered: list[Message] = []
    for m in messages:
        key = m.delivery_id or f"id:{m.id}"
        rep = reps.get(key)
        if rep is None:
            reps[key] = m
            also[key] = []
            ordered.append(m)
        elif m.account_id != rep.account_id and m.account_id not in also[key]:
            also[key].append(m.account_id)

    allowed_senders = allowed_image_senders(user)
    envelopes = []
    for m in ordered:
        env = _envelope(m, allowed_senders)
        env.also_in_account_ids = also[m.delivery_id or f"id:{m.id}"]
        envelopes.append(env)
    await _mark_subscriptions(db, user.id, ordered, envelopes)
    if not blocks_trackers(user) and ordered:
        # Trackers left in are remote images: offer "Load images" for them too.
        with_trackers = set(
            await db.scalars(
                select(BodyPart.message_id).where(
                    BodyPart.message_id.in_([m.id for m in ordered]),
                    BodyPart.kind == "html_sanitized",
                    BodyPart.trackers_blocked > 0,
                )
            )
        )
        for env in envelopes:
            env.has_remote_images = env.has_remote_images or env.id in with_trackers

    order = "desc" if sort == "desc" else "asc"
    return ThreadDetail(
        id=thread.id,
        subject=thread.subject_norm or "(no subject)",
        order=order,
        pinned=thread.pinned,
        messages=envelopes,
        snoozed_until=thread.snoozed_until,
    )