"""REST API integration tests (SPEC §9.1) driving the ASGI app with seeded data."""

from __future__ import annotations

import uuid
from collections.abc import AsyncIterator
from dataclasses import dataclass, field
from datetime import UTC, datetime
from email.message import EmailMessage

import httpx
import pytest_asyncio
from httpx import ASGITransport
from sqlalchemy import delete, select
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from sqlalchemy.pool import NullPool

from app.config import get_settings
from app.identity.passwords import hash_password
from app.identity.sessions import create_session
from app.indexer.indexer import persist_message
from app.main import create_app
from app.models import Account, Mailbox, Message, User
from app.providers.base import RemoteMessage
from app.search.pg import PgSearchBackend


@dataclass
class ApiCtx:
    client: httpx.AsyncClient
    user_id: uuid.UUID
    account_id: uuid.UUID | None = None
    archive_id: uuid.UUID | None = None
    session: AsyncSession | None = None
    thread_ids: list[uuid.UUID] = field(default_factory=list)
    message_ids: list[uuid.UUID] = field(default_factory=list)


def _raw(
    subject: str, body: str, *, mid: str, refs: str | None = None, html: str | None = None
) -> bytes:
    m = EmailMessage()
    m["Subject"] = subject
    m["From"] = "Alice <alice@example.com>"
    m["To"] = "me@example.com"
    m["Message-ID"] = mid
    m["Date"] = "Mon, 09 Mar 2026 08:36:37 +0100"
    if refs:
        m["References"] = refs
        m["In-Reply-To"] = refs
    m.set_content(body)
    if html:
        m.add_alternative(html, subtype="html")
    return m.as_bytes()


@pytest_asyncio.fixture
async def api() -> AsyncIterator[ApiCtx]:
    engine = create_async_engine(get_settings().database_url, poolclass=NullPool)
    sm = async_sessionmaker(engine, expire_on_commit=False)
    session = sm()
    user = User(handle=f"api{uuid.uuid4().hex[:8]}@test.local", password_hash=hash_password("pw"))
    session.add(user)
    await session.flush()
    account = Account(
        user_id=user.id,
        email="me@example.com",
        imap_config={},
        smtp_config={},
        secret_ct=b"",
        discovery_src="manual",
    )
    session.add(account)
    await session.flush()
    mailbox = Mailbox(account_id=account.id, user_id=user.id, path="INBOX", special_use="inbox")
    archive = Mailbox(account_id=account.id, user_id=user.id, path="Archive", special_use="archive")
    session.add_all([mailbox, archive])
    await session.commit()

    backend = PgSearchBackend()
    ctx = ApiCtx(
        client=None,  # type: ignore[arg-type]
        user_id=user.id,
        account_id=account.id,
        archive_id=archive.id,
        session=session,
    )
    specs = [
        (_raw("Project kickoff", "kickoff meeting notes and revenue plan", mid="<a@x.com>"), 1),
        (
            _raw(
                "Re: Project kickoff",
                "sounds good, revenue looks strong",
                mid="<b@x.com>",
                refs="<a@x.com>",
            ),
            2,
        ),
        (
            _raw(
                "Lunch",
                "grab lunch tomorrow?",
                mid="<c@x.com>",
                html="<p>grab <b>lunch</b> <script>x()</script>tomorrow?</p>",
            ),
            3,
        ),
    ]
    for raw, u in specs:
        remote = RemoteMessage(uid=u, raw=raw, flags=frozenset(), internal_date=datetime.now(UTC))
        result = await persist_message(
            session,
            user_id=user.id,
            account_id=account.id,
            mailbox_id=mailbox.id,
            mailbox_special_use="inbox",
            remote=remote,
        )
        await session.commit()
        assert result is not None
        ctx.message_ids.append(result.message.id)
        if result.message.thread_id not in ctx.thread_ids:
            ctx.thread_ids.append(result.message.thread_id)
        await backend.index(result.search_doc)

    token = await create_session(user.id, "pytest")
    transport = ASGITransport(app=create_app())
    ctx.client = httpx.AsyncClient(
        transport=transport, base_url="http://testserver", cookies={"session": token}
    )
    try:
        yield ctx
    finally:
        await ctx.client.aclose()
        await session.execute(delete(User).where(User.id == user.id))
        await session.commit()
        await session.close()
        await engine.dispose()


async def test_requires_auth() -> None:
    transport = ASGITransport(app=create_app())
    async with httpx.AsyncClient(transport=transport, base_url="http://testserver") as client:
        resp = await client.get("/api/threads")
    assert resp.status_code == 401
    assert resp.json()["error"]["code"] == "not_authenticated"


async def test_list_threads(api: ApiCtx) -> None:
    resp = await api.client.get("/api/threads")
    assert resp.status_code == 200
    data = resp.json()
    # 2 threads: the kickoff reply-chain + lunch
    assert len(data["threads"]) == 2
    # newest first
    subjects = [t["subject"] for t in data["threads"]]
    assert "Lunch" in subjects
    kickoff = next(t for t in data["threads"] if "kickoff" in t["subject"].lower())
    assert kickoff["message_count"] == 2


async def test_folder_view(api: ApiCtx) -> None:
    # all seeded mail is in an inbox mailbox
    inbox = await api.client.get("/api/threads", params={"view": "folder:inbox"})
    assert inbox.status_code == 200
    assert len(inbox.json()["threads"]) == 2  # kickoff chain + lunch
    # no messages are in a sent folder
    sent = await api.client.get("/api/threads", params={"view": "folder:sent"})
    assert sent.status_code == 200
    assert sent.json()["threads"] == []


async def test_account_view_shows_only_that_accounts_inbox(api: ApiCtx) -> None:
    assert api.session is not None
    # The account view starts equal to the inbox (all seeded mail is in inbox).
    resp = await api.client.get("/api/threads", params={"view": f"account:{api.account_id}"})
    assert resp.status_code == 200
    assert len(resp.json()["threads"]) == 2

    # Move the whole "Lunch" thread out of the inbox into Archive.
    lunch_id = next(t["id"] for t in resp.json()["threads"] if t["subject"] == "Lunch")
    msgs = list(
        await api.session.scalars(select(Message).where(Message.thread_id == uuid.UUID(lunch_id)))
    )
    for m in msgs:
        m.mailbox_id = api.archive_id
    await api.session.commit()

    # It must drop out of both the unified inbox and the account view...
    inbox = await api.client.get("/api/threads", params={"view": "folder:inbox"})
    assert [t["subject"] for t in inbox.json()["threads"]] == ["Project kickoff"]
    account = await api.client.get("/api/threads", params={"view": f"account:{api.account_id}"})
    assert [t["subject"] for t in account.json()["threads"]] == ["Project kickoff"]
    # ...but still be reachable under the Archive folder.
    archive = await api.client.get("/api/threads", params={"view": "folder:archive"})
    assert [t["subject"] for t in archive.json()["threads"]] == ["Lunch"]


async def test_attachments_list_and_zip(api: ApiCtx) -> None:
    import io
    import zipfile

    from app.indexer import blobstore
    from app.models import Attachment

    assert api.session is not None
    mid = api.message_ids[-1]  # the "Lunch" message
    sha = blobstore.store(api.user_id, b"hello attachment")
    api.session.add(
        Attachment(
            message_id=mid,
            user_id=api.user_id,
            filename="notes.txt",
            mime_type="text/plain",
            size_bytes=16,
            content_id=None,
            blob_sha256=sha,
        )
    )
    await api.session.commit()

    listing = await api.client.get(f"/api/messages/{mid}/attachments")
    assert listing.status_code == 200
    body = listing.json()
    assert [a["filename"] for a in body] == ["notes.txt"]
    assert body[0]["size_bytes"] == 16

    zipped = await api.client.get(f"/api/messages/{mid}/attachments.zip")
    assert zipped.status_code == 200
    assert zipped.headers["content-type"] == "application/zip"
    assert "Attachments.zip" in zipped.headers["content-disposition"]
    archive = zipfile.ZipFile(io.BytesIO(zipped.content))
    assert archive.namelist() == ["notes.txt"]
    assert archive.read("notes.txt") == b"hello attachment"


async def test_attachment_download_non_ascii_filename(api: ApiCtx) -> None:
    # A Polish filename (ł, ń) is not Latin-1, which would crash the response with
    # a 500 if put raw in Content-Disposition. It must serve 200 with RFC 6266.
    from app.indexer import blobstore
    from app.models import Attachment

    assert api.session is not None
    mid = api.message_ids[-1]
    payload = b"PK\x03\x04 zip-ish bytes"
    sha = blobstore.store(api.user_id, payload)
    name = "oferta - składowe ceny.ods"
    api.session.add(
        Attachment(
            message_id=mid,
            user_id=api.user_id,
            filename=name,
            mime_type="application/vnd.oasis.opendocument.spreadsheet",
            size_bytes=len(payload),
            content_id=None,
            blob_sha256=sha,
        )
    )
    await api.session.commit()

    aid = next(
        a["id"] for a in (await api.client.get(f"/api/messages/{mid}/attachments")).json()
        if a["filename"] == name
    )
    resp = await api.client.get(f"/api/messages/{mid}/attachments/{aid}")
    assert resp.status_code == 200  # not 500
    assert resp.content == payload
    cd = resp.headers["content-disposition"]
    assert "filename*=UTF-8''" in cd  # UTF-8 encoded name for modern clients
    assert "sk%C5%82adowe" in cd  # ł percent-encoded


async def test_quota_reports_usage_and_default(api: ApiCtx) -> None:
    resp = await api.client.get("/api/settings/quota")
    assert resp.status_code == 200
    body = resp.json()
    assert body["quota_bytes"] == 5 * 1024**3  # default 5 GiB
    assert body["used_bytes"] > 0  # seeded messages have a size


async def test_thread_detail_messages_ascending(api: ApiCtx) -> None:
    # find the kickoff thread (2 messages)
    resp = await api.client.get("/api/threads")
    kickoff = next(t for t in resp.json()["threads"] if t["message_count"] == 2)
    detail = await api.client.get(f"/api/threads/{kickoff['id']}")
    assert detail.status_code == 200
    body = detail.json()
    assert body["order"] == "asc"
    dates = [m["date"] for m in body["messages"]]
    assert dates == sorted(dates)  # ascending


async def test_thread_detail_hides_trashed_message_unless_viewing_trash(api: ApiCtx) -> None:
    assert api.session is not None
    # Give the account a Trash folder and delete a single message of the 2-message
    # kickoff thread (move it to Trash) — as the per-message Delete action does.
    trash = Mailbox(
        account_id=api.account_id, user_id=api.user_id, path="Trash", special_use="trash"
    )
    api.session.add(trash)
    await api.session.flush()
    resp = await api.client.get("/api/threads")
    kickoff = next(t for t in resp.json()["threads"] if t["message_count"] == 2)
    detail = (await api.client.get(f"/api/threads/{kickoff['id']}")).json()
    victim = uuid.UUID(detail["messages"][-1]["id"])
    msg = await api.session.get(Message, victim)
    assert msg is not None
    msg.mailbox_id = trash.id
    await api.session.commit()

    # Viewing the thread from the Inbox: the trashed message is gone.
    inbox_view = await api.client.get(
        f"/api/threads/{kickoff['id']}", params={"view": "folder:inbox"}
    )
    ids = [m["id"] for m in inbox_view.json()["messages"]]
    assert str(victim) not in ids
    assert len(ids) == 1

    # Viewing from the Trash folder: only the trashed message is shown (Trash is
    # message-scoped), not the rest of the conversation.
    trash_view = await api.client.get(
        f"/api/threads/{kickoff['id']}", params={"view": "folder:trash"}
    )
    ids_trash = [m["id"] for m in trash_view.json()["messages"]]
    assert ids_trash == [str(victim)]

    # The Inbox LIST count drops the trashed message (was 2, now 1); the Trash list
    # counts only the one trashed message.
    inbox_list = await api.client.get("/api/threads", params={"view": "folder:inbox"})
    row = next(t for t in inbox_list.json()["threads"] if t["id"] == kickoff["id"])
    assert row["message_count"] == 1
    trash_list = await api.client.get("/api/threads", params={"view": "folder:trash"})
    trow = next(t for t in trash_list.json()["threads"] if t["id"] == kickoff["id"])
    assert trow["message_count"] == 1


async def test_folder_counts(api: ApiCtx) -> None:
    assert api.session is not None
    # All three seeded inbox messages are unread; no drafts yet.
    resp = await api.client.get("/api/mailboxes/counts")
    assert resp.status_code == 200
    assert resp.json() == {"inbox_unread": 3, "drafts": 0}

    # Marking one message \Seen lowers the unread count.
    msg = await api.session.get(Message, api.message_ids[0])
    assert msg is not None
    msg.flags = ["\\Seen"]
    await api.session.commit()
    body = (await api.client.get("/api/mailboxes/counts")).json()
    assert body["inbox_unread"] == 2


async def test_thread_collapses_cross_account_duplicate(api: ApiCtx) -> None:
    assert api.session is not None
    # The kickoff thread's first message; give it a real delivery_id.
    kickoff = next(t for t in (await api.client.get("/api/threads")).json()["threads"]
                   if t["message_count"] == 2)
    detail = (await api.client.get(f"/api/threads/{kickoff['id']}")).json()
    first_id = uuid.UUID(detail["messages"][0]["id"])
    first = await api.session.get(Message, first_id)
    assert first is not None
    first.delivery_id = "<shared@x.com>"

    # A second account receives the SAME email (same delivery_id) into the same thread.
    acct2 = Account(
        user_id=api.user_id, email="me2@example.com", imap_config={}, smtp_config={},
        secret_ct=b"", discovery_src="manual",
    )
    api.session.add(acct2)
    await api.session.flush()
    inbox2 = Mailbox(account_id=acct2.id, user_id=api.user_id, path="INBOX", special_use="inbox")
    api.session.add(inbox2)
    await api.session.flush()
    api.session.add(Message(
        user_id=api.user_id, account_id=acct2.id, mailbox_id=inbox2.id, uid=1,
        thread_id=first.thread_id, message_id_hdr="<synthetic@synthetic.local>",
        delivery_id="<shared@x.com>", subject=first.subject,
        internal_date=datetime.now(UTC), blob_sha256="z" * 64,  # later -> first stays the rep
    ))
    await api.session.commit()

    body = (await api.client.get(f"/api/threads/{kickoff['id']}")).json()
    ids = [m["id"] for m in body["messages"]]
    # Still two rows (kickoff + reply) — the duplicate collapsed into the first, not
    # a third row — and the first now notes the other account.
    assert len(ids) == 2
    rep = next(m for m in body["messages"] if m["id"] == str(first_id))
    assert rep["also_in_account_ids"] == [str(acct2.id)]


async def test_message_html_is_sanitized(api: ApiCtx) -> None:
    # the lunch message has HTML with a script
    resp = await api.client.get("/api/threads")
    lunch = next(t for t in resp.json()["threads"] if t["subject"] == "Lunch")
    detail = (await api.client.get(f"/api/threads/{lunch['id']}")).json()
    mid = detail["messages"][0]["id"]
    html = await api.client.get(f"/api/messages/{mid}/html")
    assert html.status_code == 200
    assert "<script" not in html.text.lower()
    assert "lunch" in html.text.lower()


async def test_search_grouped_by_thread(api: ApiCtx) -> None:
    resp = await api.client.get("/api/search", params={"q": "revenue"})
    assert resp.status_code == 200
    data = resp.json()
    assert len(data["threads"]) == 1  # both kickoff messages -> one thread
    assert data["threads"][0]["thread"]["message_count"] == 2


async def test_accounts_list(api: ApiCtx) -> None:
    resp = await api.client.get("/api/accounts")
    assert resp.status_code == 200
    assert resp.json()[0]["email"] == "me@example.com"


async def test_thread_pagination_offset(api: ApiCtx) -> None:
    first = await api.client.get("/api/threads", params={"limit": 1, "offset": 0})
    body = first.json()
    assert len(body["threads"]) == 1
    assert body["total"] >= 2  # more than one page of threads exists
    second = await api.client.get("/api/threads", params={"limit": 1, "offset": 1})
    body2 = second.json()
    assert len(body2["threads"]) == 1
    assert body2["total"] == body["total"]  # grand total is stable across pages
    assert body2["threads"][0]["id"] != body["threads"][0]["id"]