"""The files the researcher sent, and who is allowed to fetch them.

Measured against the real mailbox on 2026-08-26: the write-up import
walked past 319 attachment files (143.1 MiB) across 54 of 149 reports —
266 images, 25 logs, 19 PDFs and 9 PoC screen recordings. For a good
share of those reports the screenshot IS the evidence.

Two halves are tested here. The MIME walk, because it is the thing that
silently returned nothing for months; and the download authorisation,
because a presigned URL bypasses Helix auth once minted, so every check
has to happen before the signature.
"""
from __future__ import annotations

import base64
import hashlib
import uuid

import pytest
from sqlalchemy import select

from app.domains.bounty import mail_intake
from app.domains.bounty import service as svc
from app.domains.bounty.models import BountyReportAttachment
from app.identity.person_models import IdentityPerson

PNG = b"\x89PNG\r\n\x1a\n" + b"screenshot-bytes" * 8
VIDEO = b"\x00\x00\x00 ftypisom" + b"poc-recording" * 16


def _part(filename, mime, data, *, attachment_id="att-1", inline=False):
    """One Gmail attachment part as the API actually returns it."""
    headers = [{"name": "Content-Type", "value": mime}]
    if inline:
        headers.append({"name": "Content-Disposition", "value": "inline"})
    return {
        "mimeType": mime,
        "filename": filename,
        "headers": headers,
        "body": {"attachmentId": attachment_id, "size": len(data)},
    }


def _message(msg_id, sender, parts, *, text="Here is the bug."):
    """A multipart message: the text alternative plus attachment parts."""
    return {
        "id": msg_id,
        "payload": {
            "mimeType": "multipart/mixed",
            "filename": "",
            "headers": [
                {"name": "From", "value": sender},
                {"name": "Date", "value": "Tue, 28 Jul 2026 10:00:00 +0700"},
            ],
            "body": {},
            "parts": [
                {
                    "mimeType": "text/plain",
                    "filename": "",
                    "headers": [],
                    "body": {
                        "data": base64.urlsafe_b64encode(
                            text.encode(),
                        ).decode().rstrip("="),
                        "size": len(text),
                    },
                },
                *parts,
            ],
        },
    }


# ── The MIME walk ───────────────────────────────────────────────────


def test_the_walk_finds_files_and_ignores_the_text_alternatives():
    """The bug that lost 319 files: the walk looked for text and stopped.

    A part counts as a file when it has BOTH a filename and an
    ``attachmentId``. The text/plain alternative has neither — it carries
    its bytes inline in ``body.data`` — so it must not be collected.
    """
    found = mail_intake.collect_attachment_parts(_message(
        "m1", "researcher@example.com",
        [
            _part("poc.png", "image/png", PNG, attachment_id="a1"),
            _part("demo.mp4", "video/mp4", VIDEO, attachment_id="a2"),
        ],
    ))
    assert [f["filename"] for f in found] == ["poc.png", "demo.mp4"]
    assert [f["mime"] for f in found] == ["image/png", "video/mp4"]
    assert found[0]["message_id"] == "m1", "provenance must survive the walk"
    assert found[0]["attachment_id"] == "a1"
    assert all(f["size"] > 0 for f in found)


def test_the_walk_reaches_files_nested_below_the_first_level():
    """Real mail nests: multipart/mixed wrapping multipart/related. The
    original text extractor recursed and the attachment scan has to as
    well, or everything below an ``alternative`` wrapper is invisible."""
    message = _message("m2", "researcher@example.com", [{
        "mimeType": "multipart/related",
        "filename": "",
        "headers": [],
        "body": {},
        "parts": [_part("deep.png", "image/png", PNG, attachment_id="deep")],
    }])
    found = mail_intake.collect_attachment_parts(message)
    assert [f["filename"] for f in found] == ["deep.png"]


def test_a_content_id_alone_does_not_make_a_part_inline():
    """The misclassification that hid every PoC recording.

    The first version treated a ``Content-ID`` as evidence of inline mail
    furniture. Gmail sets one on ordinary attachments, so 103 of 125 real
    files were classified as furniture in production — including all 9
    proof-of-concept videos, which the UI then folded away. Only
    ``Content-Disposition: inline`` counts.
    """
    node = _part("PoC.mp4", "video/mp4", VIDEO, attachment_id="v")
    node["headers"].append({"name": "Content-ID", "value": "<abc123@mail>"})
    found = mail_intake.collect_attachment_parts(
        _message("m3", "researcher@example.com", [node]),
    )
    assert len(found) == 1
    assert found[0]["is_inline"] is False, (
        "a Content-ID is not a disposition — this is what buried the videos"
    )

    # And a genuine inline disposition still records as one. The flag is
    # metadata now; nothing hides on the strength of it.
    disposed = mail_intake.collect_attachment_parts(_message(
        "m4", "researcher@example.com",
        [_part("logo.png", "image/png", PNG, attachment_id="l", inline=True)],
    ))
    assert disposed[0]["is_inline"] is True


# ── The import ──────────────────────────────────────────────────────


def _fake_gmail(monkeypatch, messages, blobs):
    """Stand in for Gmail: a thread of messages and their attachment bytes.

    Enforces the wire contract the real API has — an attachment fetch is
    per (messageId, attachmentId), and asking for a pair that was never
    advertised raises, the way a 404 does. A double that answers anything
    would let a bug that mixes up message ids pass.
    """
    class _Attachments:
        def get(self, *, userId, messageId, id):  # noqa: N803, A002
            key = (messageId, id)
            if key not in blobs:
                raise KeyError(f"no such attachment: {key}")

            class _Exec:
                def execute(self):  # noqa: N805
                    return {"data": base64.urlsafe_b64encode(
                        blobs[key],
                    ).decode().rstrip("=")}
            return _Exec()

    class _Messages:
        def attachments(self):
            return _Attachments()

    class _Threads:
        def get(self, *, userId, id, format):  # noqa: N803, A002
            class _Exec:
                def execute(self):  # noqa: N805
                    return {"messages": messages}
            return _Exec()

    class _Users:
        def messages(self):
            return _Messages()

        def threads(self):
            return _Threads()

    class _Service:
        def users(self):
            return _Users()

    async def _fake_service(_db, _mailbox):
        return _Service()

    monkeypatch.setattr(mail_intake, "_service", _fake_service)


def _fake_s3(monkeypatch):
    """Record what would be uploaded; return the key the real one returns."""
    put: list[tuple[str, bytes]] = []

    async def _put(_db, *, relative_key, content, content_type):
        put.append((relative_key, content))
        return f"prefix/{relative_key}"

    from app.core import storage

    monkeypatch.setattr(storage, "put_object_bytes", _put)
    return put


async def _report_with_thread(db, *, thread_id="t1", email="wh@example.com"):
    report = await svc.create_report(
        db, title="Unauth admin API leaks PII",
        researcher_email=email, actor=None,
    )
    await svc.add_event(
        db, report=report, event_type="created", actor=None,
        actor_kind="system", meta={"backfill": f"gmail:{thread_id}"},
        visibility="internal",
    )
    await db.flush()
    return report


@pytest.mark.asyncio
async def test_the_researchers_files_land_in_s3_with_their_provenance(
    db_session, monkeypatch,
):
    report = await _report_with_thread(db_session)
    _fake_gmail(
        monkeypatch,
        [_message("m1", "wh@example.com", [
            _part("poc.png", "image/png", PNG, attachment_id="a1"),
            _part("demo.mp4", "video/mp4", VIDEO, attachment_id="a2"),
        ])],
        {("m1", "a1"): PNG, ("m1", "a2"): VIDEO},
    )
    put = _fake_s3(monkeypatch)

    stats = await mail_intake.import_attachments_for_report(db_session, report)
    assert stats == {"found": 2, "stored": 2, "duplicate": 0,
                     "too_large": 0, "failed": 0}
    assert len(put) == 2

    rows = await svc.list_attachments(db_session, report.id)
    assert {r.filename for r in rows} == {"poc.png", "demo.mp4"}
    png = next(r for r in rows if r.filename == "poc.png")
    assert png.size_bytes == len(PNG)
    assert png.sha256 == hashlib.sha256(PNG).hexdigest()
    assert png.content_type == "image/png"
    assert png.source == "researcher_mail"
    # Provenance: the pair is what makes a later re-fetch possible.
    assert png.gmail_message_id == "m1"
    assert png.gmail_attachment_id == "a1"
    assert png.object_key.endswith(png.sha256), "key derives from the digest"


@pytest.mark.asyncio
async def test_our_own_attachments_are_never_filed_as_the_researchers(
    db_session, monkeypatch,
):
    """The write-up import once stored OUR reply as the researcher's
    finding. The same mistake here would be quieter and worse: an internal
    screenshot filed as a white-hat's evidence, in a ledger that
    authorises payments.
    """
    report = await _report_with_thread(db_session)
    _fake_gmail(
        monkeypatch,
        [
            _message("ours", "security@pingcap.com", [
                _part("internal-notes.pdf", "application/pdf", PNG,
                      attachment_id="x1"),
            ]),
            _message("theirs", "wh@example.com", [
                _part("poc.png", "image/png", PNG, attachment_id="y1"),
            ]),
        ],
        {("ours", "x1"): PNG, ("theirs", "y1"): PNG},
    )
    _fake_s3(monkeypatch)

    await mail_intake.import_attachments_for_report(db_session, report)
    rows = await svc.list_attachments(db_session, report.id)
    assert [r.filename for r in rows] == ["poc.png"]
    assert all(r.gmail_message_id == "theirs" for r in rows)


@pytest.mark.asyncio
async def test_the_same_file_twice_in_a_thread_lands_once(
    db_session, monkeypatch,
):
    """Researchers resend threads, so the same PNG legitimately arrives
    more than once. Deduped on the bytes, which also makes a re-run of the
    importer a no-op instead of tripling the gallery."""
    report = await _report_with_thread(db_session)
    _fake_gmail(
        monkeypatch,
        [
            _message("m1", "wh@example.com",
                     [_part("poc.png", "image/png", PNG, attachment_id="a1")]),
            _message("m2", "wh@example.com",
                     [_part("poc-resend.png", "image/png", PNG,
                            attachment_id="a2")]),
        ],
        {("m1", "a1"): PNG, ("m2", "a2"): PNG},
    )
    put = _fake_s3(monkeypatch)

    stats = await mail_intake.import_attachments_for_report(db_session, report)
    assert stats["stored"] == 1
    assert stats["duplicate"] == 1
    assert len(put) == 1, "identical bytes must not occupy two objects"

    # And a second pass over the same thread stores nothing new.
    again = await mail_intake.import_attachments_for_report(db_session, report)
    assert again["stored"] == 0
    assert len(await svc.list_attachments(db_session, report.id)) == 1


@pytest.mark.asyncio
async def test_an_oversized_file_is_recorded_as_skipped_not_stored(
    db_session, monkeypatch,
):
    report = await _report_with_thread(db_session)
    huge = _part("huge.bin", "application/octet-stream", b"", attachment_id="h")
    huge["body"]["size"] = mail_intake.MAX_ATTACHMENT_BYTES + 1
    _fake_gmail(monkeypatch, [_message("m1", "wh@example.com", [huge])], {})
    put = _fake_s3(monkeypatch)

    stats = await mail_intake.import_attachments_for_report(db_session, report)
    assert stats["too_large"] == 1
    assert stats["stored"] == 0
    assert put == [], "the cap has to be checked before the fetch"


@pytest.mark.asyncio
async def test_real_files_sort_ahead_of_inline_noise(db_session, monkeypatch):
    """One production report carries 90 parts, mostly signature logos.
    Leading with those buries the proof-of-concept the researcher sent."""
    report = await _report_with_thread(db_session)
    _fake_gmail(
        monkeypatch,
        [_message("m1", "wh@example.com", [
            _part("logo.png", "image/png", PNG + b"1", attachment_id="l",
                  inline=True),
            _part("proof.png", "image/png", PNG + b"2", attachment_id="p"),
        ])],
        {("m1", "l"): PNG + b"1", ("m1", "p"): PNG + b"2"},
    )
    _fake_s3(monkeypatch)

    await mail_intake.import_attachments_for_report(db_session, report)
    rows = await svc.list_attachments(db_session, report.id)
    assert [r.filename for r in rows] == ["proof.png", "logo.png"]


# ── Who may download ────────────────────────────────────────────────


async def _stored_attachment(db, report, *, filename="poc.png"):
    row = BountyReportAttachment(
        report_id=report.id, object_key=f"prefix/bounty/{report.code}/abc",
        filename=filename, content_type="image/png", size_bytes=len(PNG),
        sha256=hashlib.sha256(PNG + filename.encode()).hexdigest(),
        source="researcher_mail",
    )
    db.add(row)
    await db.flush()
    return row


@pytest.mark.asyncio
async def test_the_owner_can_download_their_own_reports_attachment(
    client, db_session, employee_user, employee_headers, monkeypatch,
):
    from app.core import storage

    async def _sign(_db, *, object_key, expires_in=600):
        return f"https://s3.example/{object_key}?sig=1"

    monkeypatch.setattr(storage, "presign_get", _sign)

    owner, _token = employee_user
    report = await svc.create_report(
        db_session, title="assigned to me",
        researcher_email="wh@example.com", actor=None,
    )
    report.assignee_person_id = owner.id
    row = await _stored_attachment(db_session, report)
    await db_session.commit()

    resp = await client.get(
        f"/api/v1/me/bounty/{report.id}/attachments/{row.id}/url",
        headers=employee_headers,
    )
    assert resp.status_code == 200
    assert "s3.example" in resp.json()["data"]["url"]


@pytest.mark.asyncio
async def test_an_employee_cannot_download_someone_elses_attachment(
    client, db_session, employee_user, employee_headers, monkeypatch,
):
    """The ACL runs BEFORE the signature. A URL, once minted, is a bearer
    token for the object — so 「not assigned to you」 has to fail here and
    not at S3, which has no idea who asked."""
    from app.core import storage

    signed: list[str] = []

    async def _sign(_db, *, object_key, expires_in=600):
        signed.append(object_key)
        return "https://s3.example/leaked"

    monkeypatch.setattr(storage, "presign_get", _sign)

    stranger = IdentityPerson(
        source_system="test", hris_person_id=f"P{uuid.uuid4().hex[:8]}",
        employee_no=f"E{uuid.uuid4().hex[:8]}",
        work_email=f"u{uuid.uuid4().hex[:8]}@x.test",
        display_name="Someone Else", employment_status="active",
    )
    db_session.add(stranger)
    await db_session.flush()

    report = await svc.create_report(
        db_session, title="not mine", researcher_email="wh@example.com",
        actor=None,
    )
    report.assignee_person_id = stranger.id
    row = await _stored_attachment(db_session, report)
    await db_session.commit()

    resp = await client.get(
        f"/api/v1/me/bounty/{report.id}/attachments/{row.id}/url",
        headers=employee_headers,
    )
    assert resp.status_code == 403
    assert signed == [], "nothing may be signed for a report that is not yours"


@pytest.mark.asyncio
async def test_an_attachment_id_from_another_report_is_not_signed(
    client, db_session, employee_user, employee_headers, monkeypatch,
):
    """Both ids are checked together. Scoping the lookup to the report the
    caller was cleared for is what stops a valid attachment id from an
    unrelated report being signed through an authorised path."""
    from app.core import storage

    signed: list[str] = []

    async def _sign(_db, *, object_key, expires_in=600):
        signed.append(object_key)
        return "https://s3.example/leaked"

    monkeypatch.setattr(storage, "presign_get", _sign)

    owner, _token = employee_user
    mine = await svc.create_report(
        db_session, title="mine", researcher_email="a@example.com", actor=None,
    )
    mine.assignee_person_id = owner.id
    other = await svc.create_report(
        db_session, title="other", researcher_email="b@example.com", actor=None,
    )
    foreign = await _stored_attachment(db_session, other, filename="secret.png")
    await db_session.commit()

    resp = await client.get(
        f"/api/v1/me/bounty/{mine.id}/attachments/{foreign.id}/url",
        headers=employee_headers,
    )
    assert resp.status_code == 404
    assert signed == []


@pytest.mark.asyncio
async def test_the_payload_never_carries_the_object_key(
    client, db_session, employee_user, employee_headers,
):
    """The key is the single input to ``presign_get``. Shipping it beside a
    download endpoint would make the endpoint's scoping decorative."""
    owner, _token = employee_user
    report = await svc.create_report(
        db_session, title="mine", researcher_email="wh@example.com", actor=None,
    )
    report.assignee_person_id = owner.id
    await _stored_attachment(db_session, report)
    await db_session.commit()

    data = (
        await client.get(
            f"/api/v1/me/bounty/{report.id}", headers=employee_headers,
        )
    ).json()["data"]

    assert len(data["attachments"]) == 1
    assert data["attachments"][0]["filename"] == "poc.png"
    assert "object_key" not in data["attachments"][0]
    assert "prefix/bounty" not in str(data), "no key anywhere in the payload"


# ── Convergence ─────────────────────────────────────────────────────


@pytest.mark.asyncio
async def test_a_wholesale_failure_does_not_mark_the_report_swept(
    db_session, monkeypatch,
):
    """The defect the first production run exposed.

    Every upload was denied by a bucket policy that had never granted this
    feature's prefix. ``import_attachments_for_report`` counted the
    failures and returned normally, so the batch driver saw a clean
    outcome and marked 60 reports swept having stored nothing — retiring
    them from the queue over a systemic error a re-run would have fixed.

    The marker is a claim that a report has been dealt with, so it must
    not survive "found files, stored none of them".
    """
    from app.core import storage
    from app.domains.bounty.models import BountyReportEvent

    report = await _report_with_thread(db_session)
    _fake_gmail(
        monkeypatch,
        [_message("m1", "wh@example.com",
                  [_part("poc.png", "image/png", PNG, attachment_id="a1")])],
        {("m1", "a1"): PNG},
    )

    async def _denied(_db, *, relative_key, content, content_type):
        raise RuntimeError("AccessDenied: no identity-based policy allows this")

    monkeypatch.setattr(storage, "put_object_bytes", _denied)
    await db_session.commit()

    out = await mail_intake.backfill_attachments(db_session, limit=5)
    await db_session.commit()

    assert out["found"] >= 1
    assert out["stored"] == 0
    assert out["errored"] >= 1, "a wholesale failure has to count as errored"

    swept = (
        await db_session.execute(
            select(BountyReportEvent).where(
                BountyReportEvent.report_id == report.id,
                BountyReportEvent.event_type == "updated",
            ),
        )
    ).scalars().all()
    assert not [
        e for e in swept
        if (e.meta or {}).get("import") == mail_intake.ATTACH_SWEPT
    ], "must stay eligible for the next run"

    # And once the bucket accepts writes, the same report sweeps normally.
    stored: list[str] = []

    async def _ok(_db, *, relative_key, content, content_type):
        stored.append(relative_key)
        return f"prefix/{relative_key}"

    monkeypatch.setattr(storage, "put_object_bytes", _ok)
    again = await mail_intake.backfill_attachments(db_session, limit=5)
    await db_session.commit()
    assert again["stored"] >= 1, "the retry must not have been retired"
    assert stored and stored[0].startswith(
        f"{mail_intake.ATTACHMENT_PREFIX}/",
    ), f"key must use the granted prefix, got {stored[0]!r}"


@pytest.mark.asyncio
async def test_a_report_with_no_attachments_is_still_marked_swept(
    db_session, monkeypatch,
):
    """The other half: finding nothing IS a completed sweep, and must be
    recorded, or reports with no attachments sit at the head of the batch
    re-costing an API call every run — the exact starvation the write-up
    import hit."""
    from app.domains.bounty.models import BountyReportEvent

    report = await _report_with_thread(db_session)
    _fake_gmail(
        monkeypatch,
        [_message("m1", "wh@example.com", [])],
        {},
    )
    _fake_s3(monkeypatch)
    await db_session.commit()

    out = await mail_intake.backfill_attachments(db_session, limit=5)
    await db_session.commit()
    assert out["found"] == 0
    assert out["swept"] >= 1

    marks = [
        e for e in (
            await db_session.execute(
                select(BountyReportEvent).where(
                    BountyReportEvent.report_id == report.id,
                ),
            )
        ).scalars().all()
        if (e.meta or {}).get("import") == mail_intake.ATTACH_SWEPT
    ]
    assert marks, "a clean empty sweep must converge"