"""The shared Feishu callback must fork assistant events without disturbing approval.

The assistant subscribes on the SAME app-level event URL the approval flow has
used in production for months. The two properties that matter are therefore:
an assistant event never reaches the approval state machine, and an approval
event never reaches the assistant.
"""
from __future__ import annotations

import time
import uuid
from unittest.mock import AsyncMock

import pytest
from httpx import AsyncClient
from sqlalchemy.ext.asyncio import AsyncSession

from tests.test_assistant.conftest import message_event

pytestmark = pytest.mark.asyncio

CALLBACK_URL = "/api/v1/approval/feishu/callback"

FEISHU_CONFIG = {
    "app_id": "cli_test",
    "app_secret": "secret_test",
    "event_verification_token": "verification-token",
}


@pytest.fixture(autouse=True)
def feishu_integration(monkeypatch):
    """The callback reads the Feishu integration before it verifies anything."""
    import app.approval.router as approval_router

    monkeypatch.setattr(
        approval_router, "get_feishu_config", AsyncMock(return_value=FEISHU_CONFIG),
    )


def push_headers() -> dict[str, str]:
    """Feishu stamps every state-changing push; the router rejects it otherwise."""
    return {"X-Lark-Request-Timestamp": str(int(time.time()))}


@pytest.fixture
def approval_spy(monkeypatch) -> list[dict]:
    """Record approval-callback invocations instead of running the state machine."""
    from app.approval.service import ApprovalService

    seen: list[dict] = []

    async def _fake_handle(self, payload):
        seen.append(payload)

    async def _accept_token(self, payload):
        return None

    monkeypatch.setattr(ApprovalService, "handle_feishu_callback", _fake_handle)
    monkeypatch.setattr(ApprovalService, "verify_callback_token", _accept_token)
    return seen


@pytest.fixture
def assistant_spy(monkeypatch) -> list[dict]:
    """Record assistant turns without spawning background work."""
    from app.assistant import feishu_events

    seen: list[dict] = []

    def _fake_claim(payload, *, trace_id):
        if not feishu_events.claims_event(payload):
            return False
        seen.append({"payload": payload, "trace_id": trace_id})
        return True

    monkeypatch.setattr(feishu_events, "try_claim_event", _fake_claim)
    return seen


async def test_assistant_event_is_claimed_and_never_reaches_approval(
    client: AsyncClient, db_session: AsyncSession, approval_spy, assistant_spy,
):
    response = await client.post(
        CALLBACK_URL, json=message_event(), headers=push_headers(),
    )

    assert response.status_code == 200
    assert response.json()["code"] == 0
    assert len(assistant_spy) == 1
    assert approval_spy == []


async def test_approval_event_is_untouched_by_the_fork(
    client: AsyncClient, db_session: AsyncSession, approval_spy, assistant_spy,
):
    payload = {
        "header": {"event_type": "approval_instance", "event_id": "evt-approval"},
        "event": {"instance_code": "inst-1", "approval_code": "appr-1"},
        "token": "verification-token",
    }

    response = await client.post(CALLBACK_URL, json=payload, headers=push_headers())

    assert response.status_code == 200
    assert assistant_spy == []
    assert len(approval_spy) == 1


async def test_url_verification_handshake_still_answers_the_challenge(
    client: AsyncClient, db_session: AsyncSession, approval_spy, assistant_spy,
):
    response = await client.post(
        CALLBACK_URL,
        json={"type": "url_verification", "challenge": "abc123", "token": "t"},
    )

    assert response.status_code == 200
    assert response.json()["challenge"] == "abc123"
    assert assistant_spy == []
    assert approval_spy == []


async def test_unverified_assistant_event_is_rejected_before_the_fork(
    client: AsyncClient, db_session: AsyncSession, monkeypatch, assistant_spy,
):
    """Verification runs first: an unsigned push must never reach the assistant."""
    from app.approval.service import ApprovalService

    async def _reject(self, payload):
        raise ValueError("bad token")

    monkeypatch.setattr(ApprovalService, "verify_callback_token", _reject)

    response = await client.post(
        CALLBACK_URL, json=message_event(), headers=push_headers(),
    )

    assert response.status_code == 401
    assert assistant_spy == []


# ── the card callback carries no X-Lark-* headers at all ─────────────
#
# Production, 2026-08-18 16:17: three taps on an answer card, three
# ``approval.callback.rejected`` audit rows reading
# ``token_reason=token_path_missing_timestamp``, and the Feishu client showing
# 「Something went wrong … code: 200671」. A ``card.action.trigger`` post ships
# only the v2 envelope — no signature, no timestamp header, no nonce — so the
# shared callback's mandatory-timestamp rule rejected every button the assistant
# has ever sent. ``assistant_message_feedback`` held zero rows to prove it.


def card_tap_payload(
    *,
    create_time_ms: int | None = None,
    event_id: str = "evt-card-0001",
    action: str = "assistant.feedback_helpful",
) -> dict:
    """The exact envelope Feishu posts for a tap, per the production audit row.

    ``header`` keys, and only these: event_id, token, create_time, event_type,
    tenant_key, app_id. ``event`` keys: operator, token, action, host, context.
    """
    return {
        "schema": "2.0",
        "header": {
            "event_id": event_id,
            "token": "verification-token",
            "create_time": str(
                create_time_ms
                if create_time_ms is not None
                else int(time.time() * 1000),
            ),
            "event_type": "card.action.trigger",
            "tenant_key": "tk_test",
            "app_id": "cli_test",
        },
        "event": {
            "operator": {
                "tenant_key": "tk_test",
                "open_id": "ou_assistant_open_0001",
                "union_id": "on_assistant_union_0001",
            },
            "token": "c-callback-token",
            "action": {
                "value": {"action": action, "message": "not-a-uuid"},
                "tag": "button",
            },
            "host": "im_message",
            "context": {"open_message_id": "om-1", "open_chat_id": "oc-1"},
        },
    }


@pytest.fixture
def card_spy(monkeypatch) -> list[dict]:
    """Record card taps that got past verification, without running them."""
    import app.approval.router as approval_router
    from app.assistant import card_actions

    seen: list[dict] = []

    async def _fake_handle(payload, *, trace_id):
        seen.append({"payload": payload, "trace_id": trace_id})
        return {"toast": {"type": "success", "content": "ok"}}

    monkeypatch.setattr(card_actions, "handle_card_action", _fake_handle)
    # The router imports the symbol inside the request, so patching the module
    # attribute is what the running code will look up.
    monkeypatch.setattr(
        approval_router, "get_feishu_config", AsyncMock(return_value=FEISHU_CONFIG),
    )
    return seen


async def test_a_card_tap_is_accepted_without_any_lark_headers(
    client: AsyncClient, db_session: AsyncSession, approval_spy, card_spy,
):
    """The 200671 regression. No headers at all — exactly what Feishu sends."""
    response = await client.post(CALLBACK_URL, json=card_tap_payload())

    assert response.status_code == 200, "a card tap was rejected before it ran"
    assert len(card_spy) == 1
    assert approval_spy == []


async def test_a_stale_card_tap_is_handled_idempotently(
    client: AsyncClient, db_session: AsyncSession, approval_spy, card_spy,
):
    """A card can be tapped from scrollback long after it was rendered.

    The event time is not reliably the tap time, so card branches use their
    own idempotent writes rather than rejecting a legitimate employee action.
    """
    ten_minutes_ago = int((time.time() - 600) * 1000)
    response = await client.post(
        CALLBACK_URL, json=card_tap_payload(create_time_ms=ten_minutes_ago),
    )

    assert response.status_code == 200
    assert len(card_spy) == 1


async def test_a_tap_with_an_unreadable_envelope_time_is_refused(
    client: AsyncClient, db_session: AsyncSession, approval_spy, card_spy,
):
    payload = card_tap_payload()
    payload["header"]["create_time"] = "not-a-number"

    response = await client.post(CALLBACK_URL, json=payload)

    assert response.status_code == 401
    assert card_spy == []


async def test_feishus_own_retry_of_a_slow_tap_is_not_dropped_as_a_replay(
    client: AsyncClient, db_session: AsyncSession, approval_spy, card_spy,
):
    """Feishu re-posts a callback it judged slow under the SAME event_id.

    Dropping it would show the employee a failure for a submission that went
    through — the defect ``_already_done_card`` exists to prevent. Every card
    branch is idempotent instead.
    """
    payload = card_tap_payload(event_id=f"evt-retry-{uuid.uuid4()}")

    first = await client.post(CALLBACK_URL, json=payload)
    second = await client.post(CALLBACK_URL, json=payload)

    assert first.status_code == 200
    assert second.status_code == 200, "Feishu's retry was dropped as a replay"
    assert len(card_spy) == 2


async def test_a_replayed_non_card_v2_event_is_dropped(
    client: AsyncClient, db_session: AsyncSession, approval_spy, card_spy,
):
    """Loosening the timestamp rule must not loosen replay protection."""
    payload = {
        "schema": "2.0",
        "header": {
            "event_id": f"evt-appr-{uuid.uuid4()}",
            "event_type": "approval_instance",
            "token": "verification-token",
            "create_time": str(int(time.time() * 1000)),
            "app_id": "cli_test",
        },
        "event": {"instance_code": "inst-replay", "approval_code": "appr-1"},
    }

    first = await client.post(CALLBACK_URL, json=payload)
    second = await client.post(CALLBACK_URL, json=payload)

    assert first.status_code == 200
    assert second.status_code == 401
    assert len(approval_spy) == 1


async def test_an_event_with_neither_a_header_nor_an_envelope_time_is_refused(
    client: AsyncClient, db_session: AsyncSession, approval_spy, card_spy,
):
    payload = {
        "header": {"event_type": "approval_instance", "event_id": "evt-no-time"},
        "event": {"instance_code": "inst-1"},
        "token": "verification-token",
    }

    response = await client.post(CALLBACK_URL, json=payload)

    assert response.status_code == 401
    assert approval_spy == []


async def test_a_nan_envelope_time_cannot_bypass_the_freshness_window(
    client: AsyncClient, db_session: AsyncSession, approval_spy, card_spy,
):
    """``float("nan")`` passes every comparison it is put through.

    ``nan <= 0`` is False and ``nan >= 1e12`` is False, so it came back as the
    event time — and ``abs(now - nan) > 300`` is False too, so the freshness check
    silently passed. Three characters in ``create_time`` made a captured callback
    replayable forever, which is the bound the card-tap dedup exemption rests on.
    """
    for hostile in ("nan", "NaN", "-nan", "inf", "-inf", "1e400"):
        payload = card_tap_payload()
        payload["header"]["create_time"] = hostile

        response = await client.post(CALLBACK_URL, json=payload)

        assert response.status_code == 401, f"{hostile!r} was accepted"
    assert card_spy == []


async def test_an_envelope_without_an_event_id_is_refused(
    client: AsyncClient, db_session: AsyncSession, approval_spy, card_spy,
):
    """No id, nothing to dedup on.

    Accepting the push anyway would leave a whole class of events with no replay
    protection at all. Feishu stamps every v2 push with one.
    """
    payload = card_tap_payload()
    payload["header"].pop("event_id")

    response = await client.post(CALLBACK_URL, json=payload)

    assert response.status_code == 401
    assert card_spy == []


async def test_a_split_event_type_cannot_dodge_the_replay_guard(
    client: AsyncClient, db_session: AsyncSession, approval_spy, card_spy,
):
    """The exemption and the routing must answer the same question.

    ``event_type_hint`` prefers the v1 top-level ``payload["type"]`` while
    ``claims_card_action`` prefers ``header["event_type"]`` — so a body carrying
    different values in the two places was exempted from the replay guard as "a
    card tap" and then routed to the approval state machine anyway.
    """
    payload = {
        "schema": "2.0",
        # Says "card tap" where the old exemption looked …
        "type": "card.action.trigger",
        "header": {
            "event_id": f"evt-split-{uuid.uuid4()}",
            # … and "approval" where the routing looks.
            "event_type": "approval_instance",
            "token": "verification-token",
            "create_time": str(int(time.time() * 1000)),
            "app_id": "cli_test",
        },
        "event": {"instance_code": "inst-split", "approval_code": "appr-1"},
    }

    first = await client.post(CALLBACK_URL, json=payload)
    second = await client.post(CALLBACK_URL, json=payload)

    assert first.status_code == 200
    assert second.status_code == 401, "a split event type dodged the replay guard"
    assert card_spy == []
    assert len(approval_spy) == 1