"""The business owner's confirmation surface.

Two things are being pinned here, and both are security properties rather
than conveniences:

1. **Self-scope authorization.** Every endpoint is reachable by any
   authenticated employee, so the ONLY thing standing between a colleague
   and someone else's report is the assignee check. Each mutation is
   tested against a non-assignee.
2. **Field narrowing.** The researcher's identity, the reward, the payment
   state and the verbatim researcher mail must not appear in an
   owner-facing payload, and `internal` timeline events must not either.
   These are asserted on the wire, not on the serializer.
"""
from __future__ import annotations

import uuid
from datetime import UTC, datetime, timedelta
from decimal import Decimal

import pytest

from app.auth.models import AuthUserRole
from app.core.security import create_access_token
from app.domains.bounty import constants as const
from app.domains.bounty import service as svc
from app.identity.person_models import IdentityPerson

BASE = "/api/v1/bounty"
ME = "/api/v1/me/bounty"


async def _employee(db_session, *, email: str, name: str):
    """An active employee with the ``employee`` role and a usable token."""
    from sqlalchemy import select

    from app.auth.models import AuthRole

    person_id = uuid.uuid4()
    db_session.add(IdentityPerson(
        id=person_id, work_email=email, display_name=name,
        employment_status="active", source_system="successfactors",
        hris_person_id=f"sf-{email.split('@')[0]}",
    ))
    await db_session.flush()
    role_id = (
        await db_session.execute(
            select(AuthRole.id).where(AuthRole.name == "employee"),
        )
    ).scalar_one()
    db_session.add(AuthUserRole(user_id=person_id, role_id=role_id))
    await db_session.commit()
    token = create_access_token(str(person_id), ["employee"])
    return person_id, {"Authorization": f"Bearer {token}"}


async def _assigned_report(db_session, owner_id: uuid.UUID, **kwargs):
    report = await svc.create_report(
        db_session,
        title=kwargs.pop("title", "docs 站反射型 XSS"),
        researcher_email=kwargs.pop("researcher_email", "wh@example.com"),
        researcher_name="Alex Chen",
        affected_system="docs.pingcap.com",
        raw_report="From: Alex Chen <wh@example.com>\nThere is an XSS...",
        summary="搜索参数没有转义,点开构造链接就会在 docs 域执行 JS。",
        reproduction="GET /search?q=<script>alert(1)</script>",
        actor=None,
        **kwargs,
    )
    person = await svc.load_assignable_person(db_session, owner_id)
    await svc.assign_report(
        db_session, report, person=person, note="你的服务", actor=None,
    )
    await db_session.commit()
    return report


@pytest.mark.asyncio
async def test_owner_sees_only_their_own_assignments(client, db_session):
    owner_id, owner_headers = await _employee(
        db_session, email="owner-a@test.com", name="Owner A",
    )
    other_id, other_headers = await _employee(
        db_session, email="owner-b@test.com", name="Owner B",
    )
    await _assigned_report(db_session, owner_id)

    resp = await client.get(ME, headers=owner_headers)
    assert resp.status_code == 200
    data = resp.json()["data"]
    assert len(data["items"]) == 1
    assert data["counts"]["awaiting_confirmation"] == 1
    assert data["counts"]["actionable"] == 1

    # The other employee's list is empty — not filtered client-side.
    resp = await client.get(ME, headers=other_headers)
    assert resp.json()["data"]["items"] == []
    assert other_id is not None


@pytest.mark.asyncio
async def test_owner_payload_hides_researcher_identity_and_money(
    client, db_session,
):
    owner_id, owner_headers = await _employee(
        db_session, email="owner-c@test.com", name="Owner C",
    )
    report = await _assigned_report(db_session, owner_id)
    # Give the report a severity + reward so the fields would exist if
    # the serializer leaked them.
    await svc.set_severity(
        db_session, report, severity="medium", note=None,
        apply_default_reward=True, actor=None,
    )
    await db_session.commit()

    resp = await client.get(f"{ME}/{report.id}", headers=owner_headers)
    assert resp.status_code == 200
    payload = resp.json()["data"]

    # Present: the technical substance the owner needs. The write-up is
    # part of that and deliberately so — it is the only field that says
    # WHERE the bug is, and an owner asked to confirm a finding they
    # cannot read has been handed a title and a deadline.
    assert payload["title"]
    assert payload["summary"]
    assert payload["reproduction"]
    assert payload["raw_report"], "the owner cannot judge what they cannot read"
    assert payload["affected_system"] == "docs.pingcap.com"
    assert payload["severity"] == "medium"

    # Absent: everything that is the security team's business. The line is
    # commercial, not technical — who the researcher is, how to reach
    # them, and what they are owed.
    for leaked in (
        "researcher", "researcher_id", "reward_amount_usd",
        "payment_status", "payment_reference", "fiscal_year",
        "tier_default_reward_usd", "last_replied_at",
    ):
        assert leaked not in payload, f"{leaked} leaked to the business owner"

    # The severity options carry no money either — the owner grades
    # impact, not price.
    assert payload["severity_options"]
    for option in payload["severity_options"]:
        assert "default_reward_usd" not in option
        assert "max_reward_usd" not in option


@pytest.mark.asyncio
async def test_owner_timeline_excludes_internal_events(client, db_session):
    owner_id, owner_headers = await _employee(
        db_session, email="owner-d@test.com", name="Owner D",
    )
    report = await _assigned_report(db_session, owner_id)
    await svc.set_severity(
        db_session, report, severity="low", note=None,
        apply_default_reward=True, actor=None,
    )
    await svc.log_reply(
        db_session, report, template_key=None,
        subject="Re: your report", body="a reward of USD 100 ...", actor=None,
    )
    await svc.add_comment(
        db_session, report, body="仅安全团队可见的内部讨论",
        visibility=const.VISIBILITY_INTERNAL, actor=None,
    )
    await svc.add_comment(
        db_session, report, body="给责任人看的说明",
        visibility=const.VISIBILITY_SHARED, actor=None,
    )
    await db_session.commit()

    payload = (
        await client.get(f"{ME}/{report.id}", headers=owner_headers)
    ).json()["data"]
    bodies = [e["body"] for e in payload["events"]]
    types = {e["event_type"] for e in payload["events"]}
    assert all(e["visibility"] == "shared" for e in payload["events"])
    assert "给责任人看的说明" in bodies
    assert "仅安全团队可见的内部讨论" not in bodies
    # The reply we sent the researcher — and its reward figure — is
    # internal.
    assert "reply_logged" not in types
    assert not any("USD 100" in (b or "") for b in bodies)


@pytest.mark.asyncio
async def test_non_assignee_is_refused_on_every_mutation(client, db_session):
    owner_id, _ = await _employee(
        db_session, email="owner-e@test.com", name="Owner E",
    )
    _, intruder_headers = await _employee(
        db_session, email="intruder@test.com", name="Intruder",
    )
    report = await _assigned_report(db_session, owner_id)

    for method, path, body in (
        ("get", f"{ME}/{report.id}", None),
        ("post", f"{ME}/{report.id}/respond", {"decision": "valid"}),
        ("post", f"{ME}/{report.id}/remediation",
         {"remediation_status": "in_progress"}),
        ("post", f"{ME}/{report.id}/comments", {"body": "hello"}),
    ):
        resp = await getattr(client, method)(
            path, headers=intruder_headers,
            **({"json": body} if body is not None else {}),
        )
        assert resp.status_code == 403, f"{method} {path} → {resp.status_code}"
        assert resp.json()["errors"][0]["code"] == "BOUNTY_NOT_ASSIGNEE"


@pytest.mark.asyncio
async def test_owner_verdict_does_not_close_the_report(client, db_session):
    owner_id, owner_headers = await _employee(
        db_session, email="owner-f@test.com", name="Owner F",
    )
    report = await _assigned_report(db_session, owner_id)

    resp = await client.post(
        f"{ME}/{report.id}/respond",
        json={
            "decision": "false_positive",
            "note": "这是设计如此,搜索页输出已转义",
        },
        headers=owner_headers,
    )
    assert resp.status_code == 200, resp.text
    payload = resp.json()["data"]
    # The owner said "false positive" — the report is NOT filed as one.
    # Scope and the reply to the researcher stay with the security team.
    assert payload["status"] == "in_discussion"
    assert payload["owner_decision"] == "false_positive"


@pytest.mark.asyncio
async def test_owner_valid_verdict_starts_the_fix_clock(client, db_session):
    owner_id, owner_headers = await _employee(
        db_session, email="owner-g@test.com", name="Owner G",
    )
    report = await _assigned_report(db_session, owner_id)
    eta = (datetime.now(UTC) + timedelta(days=14)).date().isoformat()

    resp = await client.post(
        f"{ME}/{report.id}/respond",
        json={
            "decision": "valid",
            "note": "确实存在,影响 docs 站访客",
            "suggested_severity": "medium",
            "fix_eta": eta,
        },
        headers=owner_headers,
    )
    assert resp.status_code == 200
    payload = resp.json()["data"]
    assert payload["remediation_status"] == "pending"
    assert payload["fix_eta"] == eta
    assert payload["suggested_severity"] == "medium"


@pytest.mark.asyncio
async def test_owner_cannot_set_a_risk_decision(client, db_session):
    owner_id, owner_headers = await _employee(
        db_session, email="owner-h@test.com", name="Owner H",
    )
    report = await _assigned_report(db_session, owner_id)

    for forbidden in ("wont_fix", "risk_accepted"):
        resp = await client.post(
            f"{ME}/{report.id}/remediation",
            json={"remediation_status": forbidden}, headers=owner_headers,
        )
        # Rejected by the schema before it reaches the service — accepting
        # a security risk is not the owner's call.
        assert resp.status_code == 422, forbidden

    resp = await client.post(
        f"{ME}/{report.id}/remediation",
        json={"remediation_status": "fixed"}, headers=owner_headers,
    )
    assert resp.status_code == 200
    assert resp.json()["data"]["remediation_status"] == "fixed"


@pytest.mark.asyncio
async def test_owner_comment_is_always_shared(client, db_session):
    owner_id, owner_headers = await _employee(
        db_session, email="owner-i@test.com", name="Owner I",
    )
    report = await _assigned_report(db_session, owner_id)
    resp = await client.post(
        f"{ME}/{report.id}/comments",
        json={"body": "这块是别的组的,建议找 X"}, headers=owner_headers,
    )
    assert resp.status_code == 200
    # An owner has no way to write into the security team's internal-only
    # channel — the endpoint takes no visibility argument at all.
    assert resp.json()["data"]["visibility"] == "shared"
    resp = await client.post(
        f"{ME}/{report.id}/comments",
        json={"body": "x", "visibility": "internal"}, headers=owner_headers,
    )
    assert resp.status_code == 422


@pytest.mark.asyncio
async def test_reassignment_makes_the_stale_link_say_so(client, db_session):
    """A Feishu card outlives a reassignment. The old owner must get a
    clear "not yours any more", not a 404 that reads like a broken link."""
    first_id, first_headers = await _employee(
        db_session, email="owner-j@test.com", name="Owner J",
    )
    second_id, _ = await _employee(
        db_session, email="owner-k@test.com", name="Owner K",
    )
    report = await _assigned_report(db_session, first_id)
    second = await svc.load_assignable_person(db_session, second_id)
    await svc.assign_report(
        db_session, report, person=second, note=None, actor=None,
    )
    await db_session.commit()

    resp = await client.get(f"{ME}/{report.id}", headers=first_headers)
    assert resp.status_code == 403
    assert resp.json()["errors"][0]["code"] == "BOUNTY_NOT_ASSIGNEE"


@pytest.mark.asyncio
async def test_closed_report_drops_out_of_the_open_list(client, db_session):
    owner_id, owner_headers = await _employee(
        db_session, email="owner-l@test.com", name="Owner L",
    )
    report = await _assigned_report(db_session, owner_id)
    await svc.set_status(
        db_session, report, target=const.STATUS_OUT_OF_SCOPE, note=None,
        actor=None,
    )
    await db_session.commit()

    assert (
        await client.get(ME, headers=owner_headers)
    ).json()["data"]["items"] == []
    # …but is still readable with include_closed, so a card that arrives
    # after the security team closed it does not 404.
    resp = await client.get(f"{ME}?include_closed=true", headers=owner_headers)
    items = resp.json()["data"]["items"]
    assert len(items) == 1
    detail = (
        await client.get(f"{ME}/{report.id}", headers=owner_headers)
    ).json()["data"]
    assert detail["can_respond"] is False


@pytest.mark.asyncio
async def test_console_does_not_claim_notified_when_nothing_was_delivered(
    client, admin_headers, db_session,
):
    """``notified`` must mean "the owner was actually told", not "we wrote
    an outbox row".

    ``dispatch`` persists the row and dead-letters a send that every
    channel rejected — without raising. With no Feishu/email integration
    configured (the test environment, and any environment mid-outage) that
    is exactly what happens, so a naive ``return True`` after ``dispatch``
    would render 已通知 on the console for a card nobody received.
    """
    owner_id, _ = await _employee(
        db_session, email="owner-m@test.com", name="Owner M",
    )
    created = await client.post(
        f"{BASE}/reports",
        json={
            "title": "x", "researcher_email": "wh2@example.com",
            "affected_system": "docs.pingcap.com",
        },
        headers=admin_headers,
    )
    assert created.status_code == 200
    report_id = created.json()["data"]["id"]

    resp = await client.post(
        f"{BASE}/reports/{report_id}/assign",
        json={"person_id": str(owner_id)}, headers=admin_headers,
    )
    assert resp.status_code == 200
    payload = resp.json()["data"]
    assert payload["status"] == "awaiting_owner"
    # No integration is configured in tests, so no channel delivered.
    assert payload["notified"] is False

    # …and the attempt is still on the record, so an operator can see the
    # card was tried and dead-lettered rather than never generated.
    from sqlalchemy import select

    from app.notifications.models import NotificationOutbox

    rows = (
        await db_session.execute(
            select(NotificationOutbox).where(
                NotificationOutbox.template_key
                == "bounty.owner_confirm_requested",
            ),
        )
    ).scalars().all()
    assert len(rows) == 1
    assert rows[0].recipient_person_id == owner_id


@pytest.mark.asyncio
async def test_counts_split_verdict_work_from_fix_work(client, db_session):
    owner_id, owner_headers = await _employee(
        db_session, email="owner-n@test.com", name="Owner N",
    )
    awaiting = await _assigned_report(
        db_session, owner_id, researcher_email="wh3@example.com",
    )
    confirmed = await _assigned_report(
        db_session, owner_id, title="第二条", researcher_email="wh4@example.com",
    )
    await svc.record_owner_response(
        db_session, confirmed, decision=const.OWNER_DECISION_VALID, note=None,
        suggested_severity=None, fix_eta=None, remediation_status=None,
        actor=None,
    )
    await svc.set_severity(
        db_session, confirmed, severity="low", note=None,
        apply_default_reward=True, actor=None,
    )
    await svc.set_status(
        db_session, confirmed, target=const.STATUS_CONFIRMED, note=None,
        actor=None,
    )
    await db_session.commit()

    counts = (
        await client.get(ME, headers=owner_headers)
    ).json()["data"]["counts"]
    assert counts["total"] == 2
    assert counts["awaiting_confirmation"] == 1
    assert counts["fix_outstanding"] == 1
    # The badge counts unanswered work only — an in-progress fix is
    # tracked, not nagged.
    assert counts["actionable"] == 1
    assert awaiting is not None
    assert Decimal(str(confirmed.reward_amount_usd)) == Decimal("100.00")