"""Tests for ``feishu.approval_drift_reconciler`` (2026-07-03).

Fast-path companion to the daily ``approval_pending_reaper``: every
15 minutes it scans pending ``approval_requests`` older than
``min_age_minutes`` and replays the missed callback when Feishu's side
is already terminal. Born from the prod incident where terminal
webhooks for asset-linked approvals crashed on the form-string bug and
rows sat 待处理 until an admin opened the detail drawer — after the
crash fix, this cron closes the remaining at-most-once delivery window
to ≤15 minutes instead of the reaper's 14 days.
"""

from __future__ import annotations

import logging
import uuid
from contextlib import ExitStack
from contextvars import Context
from datetime import UTC, datetime, timedelta
from unittest.mock import AsyncMock, patch

import pytest
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.approval.models import ApprovalRequest, ApprovalSubmissionIntent
from app.approval.submission_intents import compact_terminal_intent_payload
from app.scheduled_jobs.feishu_handlers import run_approval_drift_reconciler

_FEISHU_CFG = {"app_id": "cli_test", "app_secret": "secret_test"}
_INTENT_RECONCILER_METHOD_PATHS = (
    "app.approval.service.ApprovalService.reconcile_submission_intents",
    "app.domains.announcement_request.service."
    "AnnouncementRequestService.reconcile_submission_intents",
    "app.domains.vpn_application.service.VpnApplicationService.reconcile_submission_intents",
    "app.domains.ai_tool_application.service.AiToolApplicationService.reconcile_submission_intents",
    "app.domains.domain_registration.service."
    "DomainRegistrationService.reconcile_submission_intents",
    "app.domains.sso.service.SsoGrantService.reconcile_submission_intents",
    "app.identity.consumers.request_service.ConsumerRequestService.reconcile_submission_intents",
    "app.domains.data_access.service.DataAccessService.reconcile_submission_intents",
    "app.domains.apikey.service.ApiKeyService.reconcile_submission_intents",
    "app.domains.cmdb.scope_request_service.CloudScopeRequestService.reconcile_submission_intents",
    "app.domains.cmdb.dns_request_service.DnsRequestService.reconcile_submission_intents",
)


@pytest.fixture(autouse=True)
def _patch_feishu_cfg(monkeypatch):
    async def _fake_cfg(**_):
        return dict(_FEISHU_CFG)

    monkeypatch.setattr("app.integrations.compat.get_feishu_config", _fake_cfg)
    monkeypatch.setattr("app.approval.service.get_feishu_config", _fake_cfg)

    async def _fake_token(self):
        return "fake-tenant-token"

    monkeypatch.setattr(
        "app.approval.feishu_client.FeishuApprovalClient.get_tenant_access_token",
        _fake_token,
    )


async def _seed_pending(
    db: AsyncSession,
    instance_code: str,
    *,
    age_minutes: int,
    approval_type: str = "sso_grant_request",
) -> ApprovalRequest:
    req = ApprovalRequest(
        id=uuid.uuid4(),
        external_id=instance_code,
        approval_type=approval_type,
        applicant_id=None,
        status="pending",
        source_payload={},
        created_at=datetime.now(UTC) - timedelta(minutes=age_minutes),
    )
    db.add(req)
    await db.commit()
    return req


@pytest.mark.asyncio
async def test_reconciler_aggregates_each_submission_domain_once(db_session):
    """The unified rollout must neither omit nor double-count a domain."""
    with ExitStack() as stack:
        reconcilers = [
            stack.enter_context(patch(path, new_callable=AsyncMock))
            for path in _INTENT_RECONCILER_METHOD_PATHS
        ]
        for reconciler in reconcilers:
            reconciler.return_value = {"recovered": 1, "failed": 2}

        result = await run_approval_drift_reconciler(
            db_session,
            {"intent_min_age_minutes": 7, "intent_batch_limit": 13},
        )

    assert result["submission_intents"] == {"recovered": 11, "failed": 22}
    for reconciler in reconcilers:
        reconciler.assert_awaited_once()
        assert reconciler.await_args.kwargs == {
            "min_age_minutes": 7,
            "batch_limit": 13,
        }


@pytest.mark.parametrize("failed_indexes", ((1,), (1, 7, 10)))
@pytest.mark.asyncio
async def test_intent_domain_failures_do_not_starve_other_domains_or_drift(
    db_session,
    failed_indexes: tuple[int, ...],
):
    await _seed_pending(
        db_session,
        f"drift-after-{len(failed_indexes)}-intent-failures",
        age_minutes=30,
    )
    with ExitStack() as stack:
        reconcilers = [
            stack.enter_context(patch(path, new_callable=AsyncMock))
            for path in _INTENT_RECONCILER_METHOD_PATHS
        ]
        for index, reconciler in enumerate(reconcilers):
            if index in failed_indexes:
                reconciler.side_effect = RuntimeError(f"domain-{index}-failed")
            else:
                reconciler.return_value = {"failed": 0}
        stack.enter_context(
            patch(
                "app.approval.feishu_client.FeishuApprovalClient.get_instance_detail",
                new=AsyncMock(return_value={"status": "PENDING"}),
            )
        )

        result = await run_approval_drift_reconciler(db_session, {})

    assert result["checked"] == 1
    assert result["still_pending"] == 1
    assert result["submission_intents"] == {
        "failed": len(failed_indexes),
        "domain_failures": len(failed_indexes),
    }
    for reconciler in reconcilers:
        reconciler.assert_awaited_once()


@pytest.mark.asyncio
async def test_all_intent_domains_fail_only_after_every_domain_and_drift_run(
    db_session,
):
    await _seed_pending(db_session, "drift-after-all-intent-failures", age_minutes=30)
    fetch = AsyncMock(return_value={"status": "PENDING"})
    with ExitStack() as stack:
        reconcilers = [
            stack.enter_context(patch(path, new_callable=AsyncMock))
            for path in _INTENT_RECONCILER_METHOD_PATHS
        ]
        for index, reconciler in enumerate(reconcilers):
            reconciler.side_effect = RuntimeError(f"domain-{index}-failed")
        stack.enter_context(
            patch(
                "app.approval.feishu_client.FeishuApprovalClient.get_instance_detail",
                new=fetch,
            )
        )

        with pytest.raises(RuntimeError, match="all 11 intent domains failed"):
            await run_approval_drift_reconciler(db_session, {})

    for reconciler in reconcilers:
        reconciler.assert_awaited_once()
    fetch.assert_awaited_once_with("drift-after-all-intent-failures")


@pytest.mark.asyncio
async def test_all_checked_intents_failed_marks_run_degraded(db_session):
    with ExitStack() as stack:
        reconcilers = [
            stack.enter_context(patch(path, new_callable=AsyncMock))
            for path in _INTENT_RECONCILER_METHOD_PATHS
        ]
        for reconciler in reconcilers:
            reconciler.return_value = {
                "checked": 0,
                "reconciled": 0,
                "still_unknown": 0,
                "failed": 0,
            }
        reconcilers[4].return_value = {
            "checked": 2,
            "reconciled": 0,
            "still_unknown": 0,
            "failed": 2,
        }

        with pytest.raises(RuntimeError, match="all 2 checked intents failed"):
            await run_approval_drift_reconciler(db_session, {})

    for reconciler in reconcilers:
        reconciler.assert_awaited_once()


@pytest.mark.asyncio
async def test_partial_checked_intent_failure_keeps_run_successful(db_session):
    with ExitStack() as stack:
        reconcilers = [
            stack.enter_context(patch(path, new_callable=AsyncMock))
            for path in _INTENT_RECONCILER_METHOD_PATHS
        ]
        for reconciler in reconcilers:
            reconciler.return_value = {
                "checked": 0,
                "reconciled": 0,
                "still_unknown": 0,
                "failed": 0,
            }
        reconcilers[4].return_value = {
            "checked": 2,
            "reconciled": 1,
            "still_unknown": 0,
            "failed": 1,
        }

        result = await run_approval_drift_reconciler(db_session, {})

    assert result["submission_intents"]["checked"] == 2
    assert result["submission_intents"]["reconciled"] == 1
    assert result["submission_intents"]["failed"] == 1


@pytest.mark.asyncio
async def test_drift_fetch_log_redacts_provider_credentials(
    db_session,
    caplog,
):
    await _seed_pending(db_session, "drift-secret-log", age_minutes=30)
    error = RuntimeError(
        "Authorization: Bearer bearer-secret "
        "SecretKey=provider-secret-key "
        "https://provider.invalid/state?token=query-secret"
    )

    with (
        patch(
            "app.approval.feishu_client.FeishuApprovalClient.get_instance_detail",
            new=AsyncMock(side_effect=error),
        ),
        caplog.at_level(logging.WARNING),
        pytest.raises(RuntimeError, match="all 1 rows failed"),
    ):
        await run_approval_drift_reconciler(db_session, {})

    for secret in ("bearer-secret", "provider-secret-key", "query-secret"):
        assert secret not in caplog.text
    assert "<redacted>" in caplog.text
    assert "<query-redacted>" in caplog.text


@pytest.mark.asyncio
async def test_reconciler_replays_terminal_and_skips_young(db_session):
    """A pending row past min-age whose Feishu state is APPROVED gets
    replayed; a row younger than min-age is not even fetched (young
    rows are overwhelmingly genuinely-pending — skipping them keeps
    Feishu API volume proportional to rows humans actually wait on).
    """
    old = await _seed_pending(db_session, "drift_old_approved", age_minutes=30)
    old_id = old.id
    await _seed_pending(db_session, "drift_too_young", age_minutes=2)

    fetched: list[str] = []

    async def _fake_detail(self, code):
        fetched.append(code)
        return {"status": "APPROVED"}

    with patch(
        "app.approval.feishu_client.FeishuApprovalClient.get_instance_detail",
        new=_fake_detail,
    ):
        result = await run_approval_drift_reconciler(db_session, {})

    assert result["checked"] == 1
    assert result["replayed"] == 1
    assert fetched == ["drift_old_approved"]

    db_session.expire_all()
    row = (
        await db_session.execute(
            select(ApprovalRequest).where(ApprovalRequest.id == old_id),
        )
    ).scalar_one()
    assert row.status == "approved"


@pytest.mark.parametrize(
    ("provider_status", "local_status"),
    (("APPROVED", "approved"), ("REJECTED", "rejected")),
)
@pytest.mark.asyncio
async def test_recovered_nonasset_terminal_replays_same_tick_only(
    db_session: AsyncSession,
    provider_status: str,
    local_status: str,
) -> None:
    """Only a row materialized by this run bypasses the young-row cutoff.

    The second row models an ordinary HTTP submission finalized during the
    same wall-clock window.  Its compaction runs in an independent Context,
    as a sibling request task would, so it must not enter this run's collector.
    A second tick also proves the terminal row cannot repeat callback effects.
    """
    recovered_code = f"same-tick-{provider_status.lower()}"
    ordinary_code = f"ordinary-young-{provider_status.lower()}"
    recovered_id: uuid.UUID | None = None
    ordinary_id: uuid.UUID | None = None

    async def _materialize_current_run(**_kwargs: int) -> dict[str, int]:
        nonlocal recovered_id, ordinary_id
        if recovered_id is None:
            recovered = await _seed_pending(
                db_session,
                recovered_code,
                age_minutes=0,
            )
            ordinary = await _seed_pending(
                db_session,
                ordinary_code,
                age_minutes=0,
            )
            recovered_id = recovered.id
            ordinary_id = ordinary.id

            recovered_intent = ApprovalSubmissionIntent(
                scope="sso_grant_request",
                applicant_id=uuid.uuid4(),
                idempotency_key=f"recovered-{provider_status}",
                payload_fingerprint="r" * 64,
                provider_request_uuid=str(uuid.uuid4()),
                state="completed",
                approval_request_id=recovered.id,
                payload={},
            )
            compact_terminal_intent_payload(
                recovered_intent,
                resource_type="sso_grant",
                resource_id=uuid.uuid4(),
            )

            ordinary_intent = ApprovalSubmissionIntent(
                scope="sso_grant_request",
                applicant_id=uuid.uuid4(),
                idempotency_key=f"ordinary-{provider_status}",
                payload_fingerprint="o" * 64,
                provider_request_uuid=str(uuid.uuid4()),
                state="completed",
                approval_request_id=ordinary.id,
                payload={},
            )
            Context().run(
                compact_terminal_intent_payload,
                ordinary_intent,
                resource_type="sso_grant",
                resource_id=uuid.uuid4(),
            )
        else:
            # A reconciler may idempotently encounter the same completed
            # receipt again.  Collection is harmless because the linked row
            # is no longer pending after the first canonical replay.
            repeated_intent = ApprovalSubmissionIntent(
                scope="sso_grant_request",
                applicant_id=uuid.uuid4(),
                idempotency_key=f"repeat-{provider_status}",
                payload_fingerprint="x" * 64,
                provider_request_uuid=str(uuid.uuid4()),
                state="completed",
                approval_request_id=recovered_id,
                payload={},
            )
            compact_terminal_intent_payload(
                repeated_intent,
                resource_type="sso_grant",
                resource_id=uuid.uuid4(),
            )
        return {"checked": 1, "reconciled": 1, "failed": 0}

    fetched: list[str] = []

    async def _terminal_detail(self, code: str) -> dict[str, str]:
        fetched.append(code)
        return {"status": provider_status}

    with ExitStack() as stack:
        reconcilers = [
            stack.enter_context(patch(path, new_callable=AsyncMock))
            for path in _INTENT_RECONCILER_METHOD_PATHS
        ]
        for reconciler in reconcilers:
            reconciler.return_value = {"checked": 0, "failed": 0}
        reconcilers[1].side_effect = _materialize_current_run
        stack.enter_context(
            patch(
                "app.approval.feishu_client.FeishuApprovalClient.get_instance_detail",
                new=_terminal_detail,
            )
        )

        first = await run_approval_drift_reconciler(db_session, {})
        second = await run_approval_drift_reconciler(db_session, {})

    assert first["checked"] == 1
    assert first["replayed"] == 1
    assert second["checked"] == 0
    assert second["replayed"] == 0
    assert fetched == [recovered_code]

    assert recovered_id is not None
    assert ordinary_id is not None
    db_session.expire_all()
    recovered_row = await db_session.get(ApprovalRequest, recovered_id)
    ordinary_row = await db_session.get(ApprovalRequest, ordinary_id)
    assert recovered_row is not None and recovered_row.status == local_status
    assert ordinary_row is not None and ordinary_row.status == "pending"


@pytest.mark.asyncio
async def test_current_run_rows_share_total_batch_and_precede_aged_rows(
    db_session: AsyncSession,
) -> None:
    """Current-run priority must not enlarge the configured total batch."""
    aged = await _seed_pending(db_session, "aged-outside-capacity", age_minutes=30)
    current_codes = [f"current-run-{index}" for index in range(3)]
    materialized = False

    async def _materialize_over_batch(**_kwargs: int) -> dict[str, int]:
        nonlocal materialized
        if materialized:
            return {"checked": 0, "failed": 0}

        materialized = True
        base_time = datetime.now(UTC)
        request_ids: list[uuid.UUID] = []
        for index, code in enumerate(current_codes):
            request_id = uuid.uuid4()
            request_ids.append(request_id)
            db_session.add(
                ApprovalRequest(
                    id=request_id,
                    external_id=code,
                    approval_type="sso_grant_request",
                    applicant_id=None,
                    status="pending",
                    source_payload={},
                    created_at=base_time + timedelta(seconds=index),
                )
            )
        await db_session.commit()

        for index, request_id in enumerate(request_ids):
            intent = ApprovalSubmissionIntent(
                scope="sso_grant_request",
                applicant_id=uuid.uuid4(),
                idempotency_key=f"over-batch-{index}",
                payload_fingerprint=str(index) * 64,
                provider_request_uuid=str(uuid.uuid4()),
                state="completed",
                approval_request_id=request_id,
                payload={},
            )
            compact_terminal_intent_payload(
                intent,
                resource_type="sso_grant",
                resource_id=uuid.uuid4(),
            )
        return {"checked": 3, "reconciled": 3, "failed": 0}

    fetched: list[str] = []

    async def _pending_detail(self, code: str) -> dict[str, str]:
        fetched.append(code)
        return {"status": "PENDING"}

    with ExitStack() as stack:
        reconcilers = [
            stack.enter_context(patch(path, new_callable=AsyncMock))
            for path in _INTENT_RECONCILER_METHOD_PATHS
        ]
        for reconciler in reconcilers:
            reconciler.return_value = {"checked": 0, "failed": 0}
        reconcilers[1].side_effect = _materialize_over_batch
        stack.enter_context(
            patch(
                "app.approval.feishu_client.FeishuApprovalClient.get_instance_detail",
                new=_pending_detail,
            )
        )

        result = await run_approval_drift_reconciler(
            db_session,
            {"batch_limit": 2},
        )

    assert result["checked"] == 2
    assert result["still_pending"] == 2
    assert fetched == current_codes[:2]
    assert aged.external_id not in fetched
    assert current_codes[2] not in fetched


@pytest.mark.asyncio
async def test_reconciler_leaves_feishu_pending_untouched(db_session):
    """Feishu still PENDING → no replay, no status change, counted as
    still_pending so the run summary shows the cron did look."""
    req = await _seed_pending(db_session, "drift_still_pending", age_minutes=30)
    req_id = req.id

    with patch(
        "app.approval.feishu_client.FeishuApprovalClient.get_instance_detail",
        new=AsyncMock(return_value={"status": "PENDING"}),
    ):
        result = await run_approval_drift_reconciler(db_session, {})

    assert result["checked"] == 1
    assert result["replayed"] == 0
    assert result["still_pending"] == 1

    db_session.expire_all()
    row = (
        await db_session.execute(
            select(ApprovalRequest).where(ApprovalRequest.id == req_id),
        )
    ).scalar_one()
    assert row.status == "pending"


@pytest.mark.asyncio
async def test_reconciler_all_failed_raises_for_failure_notifier(db_session):
    """100% fetch failure must raise — a silent all-fail tick would let
    a Feishu credential rotation rot unnoticed (same contract as the
    daily reaper)."""
    await _seed_pending(db_session, "drift_fetch_fail", age_minutes=30)

    with patch(
        "app.approval.feishu_client.FeishuApprovalClient.get_instance_detail",
        new=AsyncMock(side_effect=RuntimeError("feishu down")),
    ):
        with pytest.raises(RuntimeError, match="all 1 rows failed"):
            await run_approval_drift_reconciler(db_session, {})


@pytest.mark.asyncio
async def test_reconciler_isolates_replay_failure_per_row(db_session):
    """One row's replay crash must not poison the batch: the other
    row's replay still lands (SAVEPOINT isolation, mirroring the
    reaper). This is exactly the failure class the 2026-07-03 form-
    string bug exposed — a systematic dispatcher crash must degrade to
    per-row failure counts, not an aborted batch."""
    ok_row = await _seed_pending(db_session, "drift_ok", age_minutes=30)
    ok_row_id = ok_row.id
    await _seed_pending(db_session, "drift_crash", age_minutes=40)

    real_dispatch = None

    async def _fake_detail(self, code):
        return {"status": "APPROVED"}

    from app.approval.service import ApprovalService

    real_dispatch = ApprovalService.handle_feishu_callback

    async def _crash_on_marked(self, payload, **kwargs):
        code = payload.get("event", {}).get("instance_code")
        if code == "drift_crash":
            raise RuntimeError("boom in domain handler")
        return await real_dispatch(self, payload, **kwargs)

    with (
        patch(
            "app.approval.feishu_client.FeishuApprovalClient.get_instance_detail",
            new=_fake_detail,
        ),
        patch.object(ApprovalService, "handle_feishu_callback", _crash_on_marked),
    ):
        result = await run_approval_drift_reconciler(db_session, {})

    assert result["checked"] == 2
    assert result["replayed"] == 1
    assert result["replay_failures"] == 1

    db_session.expire_all()
    ok = (
        await db_session.execute(
            select(ApprovalRequest).where(ApprovalRequest.id == ok_row_id),
        )
    ).scalar_one()
    assert ok.status == "approved"
    crashed = (
        await db_session.execute(
            select(ApprovalRequest).where(ApprovalRequest.external_id == "drift_crash"),
        )
    ).scalar_one()
    assert crashed.status == "pending"


@pytest.mark.asyncio
async def test_reconciler_raises_when_no_terminal_row_makes_progress(db_session):
    """Regression (2026-08-06 Zoom seat incident): a single terminal row
    that fails replay every tick must fail the run, even when the rest of
    the batch is legitimately still pending at Feishu.

    The old all-rows-failed guard only tripped when *every* checked row
    failed, so ``replay_failures=1, still_pending=7`` reported
    ``last_status=ok``. A Feishu-approved software-seat request sat
    unprovisioned for 20 hours behind that green light.
    """
    await _seed_pending(db_session, "drift_terminal_broken", age_minutes=30)
    await _seed_pending(db_session, "drift_still_pending_a", age_minutes=30)
    await _seed_pending(db_session, "drift_still_pending_b", age_minutes=40)

    async def _fake_detail(self, code):
        if code == "drift_terminal_broken":
            return {"status": "APPROVED"}
        return {"status": "PENDING"}

    from app.approval.service import ApprovalService

    async def _crash(self, payload, **kwargs):
        raise RuntimeError("dispatcher blew up")

    with (
        patch(
            "app.approval.feishu_client.FeishuApprovalClient.get_instance_detail",
            new=_fake_detail,
        ),
        patch.object(ApprovalService, "handle_feishu_callback", _crash),
        pytest.raises(RuntimeError, match="no terminal row could be replayed"),
    ):
        await run_approval_drift_reconciler(db_session, {})


@pytest.mark.asyncio
async def test_reconciler_stays_green_when_only_pending_rows(db_session):
    """Counterpart: nothing terminal to replay is a perfectly healthy
    tick — the new guard must not turn a quiet batch red."""
    await _seed_pending(db_session, "drift_quiet_a", age_minutes=30)
    await _seed_pending(db_session, "drift_quiet_b", age_minutes=30)

    async def _fake_detail(self, code):
        return {"status": "PENDING"}

    with patch(
        "app.approval.feishu_client.FeishuApprovalClient.get_instance_detail",
        new=_fake_detail,
    ):
        result = await run_approval_drift_reconciler(db_session, {})

    assert result["checked"] == 2
    assert result["still_pending"] == 2
    assert result["replay_failures"] == 0