"""Announcement submission deduplication and durable-provider regressions.

The suite covers sequential and true dual-session contention, response loss,
process death after provider acceptance, scheduled recovery, credential-safe
intent persistence, and the closed inventory of Feishu callers still awaiting
migration to stable provider UUIDs.
"""
from __future__ import annotations

import ast
import asyncio
import uuid
from contextlib import asynccontextmanager
from datetime import UTC, datetime, timedelta
from pathlib import Path

import httpx
import pytest
from sqlalchemy import func, select
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker

from app.approval.models import ApprovalSubmissionIntent
from app.core.exceptions import AppException
from app.domains.announcement_request import service as ann_service
from app.domains.announcement_request.models import (
    ANNOUNCEMENT_REQ_STATUS_PENDING,
    AnnouncementRequest,
)
from app.identity.person_models import IdentityPerson
from tests.support.approval_intents import make_submission_intent_retries_due


class _FakeFeishuClient:
    """Counts how many Feishu approval instances were created."""

    def __init__(self) -> None:
        self.created = 0

    async def get_approval_definition(self, approval_code: str) -> dict:
        return {"form": "[]"}

    async def resolve_user_id_by_email(self, email: str) -> str:
        return "fake-feishu-user-id"

    async def create_instance(
        self, *, approval_code, user_id, form, instance_uuid=None, **kwargs,
    ) -> dict:
        self.created += 1
        return {"instance_code": f"ann-inst-{uuid.uuid4().hex[:8]}"}

    async def cancel_instance(self, *, instance_code, user_id, approval_code):
        return None


class _FakeFlow:
    external_code = "fake-approval-code"


def test_uuid_less_nonasset_callers_are_a_closed_migration_inventory() -> None:
    """No new production call may silently join the remaining rollout."""
    app_root = Path(__file__).parents[2] / "app"
    remaining: set[tuple[str, str]] = set()
    found: set[tuple[str, str]] = set()
    for path in app_root.rglob("*.py"):
        tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
        for function in ast.walk(tree):
            if not isinstance(function, (ast.FunctionDef, ast.AsyncFunctionDef)):
                continue
            for node in ast.walk(function):
                if not (
                    isinstance(node, ast.Call)
                    and isinstance(node.func, ast.Attribute)
                    and node.func.attr == "create_instance"
                ):
                    continue
                if not any(kw.arg == "instance_uuid" for kw in node.keywords):
                    found.add((str(path.relative_to(app_root)), function.name))
    assert found == remaining
    assert ("domains/announcement_request/service.py", "submit") not in found


class _ResponseLossFeishuClient(_FakeFeishuClient):
    """Accept once, lose the response, then expose the UUID via lookup."""

    def __init__(self) -> None:
        super().__init__()
        self.instance_uuids: list[str] = []
        self.instance_code = "ann-response-loss-instance"

    async def create_instance(
        self, *, approval_code, user_id, form, instance_uuid=None, **kwargs,
    ) -> dict:
        assert instance_uuid
        self.created += 1
        self.instance_uuids.append(instance_uuid)
        if self.created == 1:
            raise httpx.ReadTimeout(
                "provider accepted; response lost; app_secret=never-persist-me"
            )
        raise AppException(
            code="FEISHU_API_ERROR",
            message="Feishu API error 60012: duplicate uuid",
            status_code=502,
        )

    async def find_instance_by_uuid(
        self, *, approval_code, instance_uuid, start_time, end_time,
        max_pages=2,
    ) -> dict:
        assert instance_uuid == self.instance_uuids[0]
        assert datetime.now(UTC) - timedelta(hours=1) < end_time
        return {
            "instance_code": self.instance_code,
            "uuid": instance_uuid,
            "status": "PENDING",
        }


@pytest.mark.asyncio
async def test_double_submit_creates_one_pending_one_feishu_instance(
    db_session: AsyncSession, monkeypatch,
) -> None:
    requester = IdentityPerson(
        id=uuid.uuid4(),
        source_system="successfactors",
        hris_person_id=f"sf-{uuid.uuid4().hex[:8]}",
        display_name="Announcement Requester",
        work_email=f"ann-{uuid.uuid4().hex[:6]}@test.com",
        employment_status="active",
    )
    db_session.add(requester)
    await db_session.commit()

    # Stub the Feishu side so submit() doesn't hit the network.
    fake_client = _FakeFeishuClient()

    async def _fake_flow(*args, **kwargs):
        return _FakeFlow()

    monkeypatch.setattr(ann_service, "get_approval_flow", _fake_flow)
    monkeypatch.setattr(
        ann_service, "_build_announcement_form_payload",
        lambda widgets, field_values: "[]",
    )

    svc = ann_service.AnnouncementRequestService(db_session)
    svc._feishu_client = fake_client  # inject fake, skip _client() config

    # First submit succeeds.
    row = await svc.submit(
        requester=requester,
        description="Please announce the new VPN node to all-hands.",
        trace_id="trace-1",
    )
    assert row.status == ANNOUNCEMENT_REQ_STATUS_PENDING

    # Immediate retry of the same intent → rejected as duplicate-pending.
    with pytest.raises(AppException) as ei:
        await svc.submit(
            requester=requester,
            description="Please announce the new VPN node to all-hands.",
            trace_id="trace-2",
        )
    assert ei.value.code == "ANNOUNCEMENT_ALREADY_PENDING"
    assert ei.value.status_code == 409

    # Exactly ONE pending row for this requester.
    pending_count = await db_session.scalar(
        select(func.count())
        .select_from(AnnouncementRequest)
        .where(
            AnnouncementRequest.person_id == str(requester.id),
            AnnouncementRequest.status == ANNOUNCEMENT_REQ_STATUS_PENDING,
        )
    )
    assert pending_count == 1

    # Exactly ONE Feishu approval instance created — the second submit
    # must not have fired a parallel approval card.
    assert fake_client.created == 1


@pytest.mark.asyncio
async def test_response_loss_retry_reuses_durable_feishu_uuid(
    db_session: AsyncSession, monkeypatch,
) -> None:
    requester = IdentityPerson(
        id=uuid.uuid4(),
        source_system="successfactors",
        hris_person_id=f"sf-{uuid.uuid4().hex[:8]}",
        display_name="Announcement response-loss requester",
        work_email=f"ann-loss-{uuid.uuid4().hex[:6]}@test.com",
        employment_status="active",
    )
    db_session.add(requester)
    await db_session.commit()

    fake_client = _ResponseLossFeishuClient()

    async def _fake_flow(*args, **kwargs):
        return _FakeFlow()

    monkeypatch.setattr(ann_service, "get_approval_flow", _fake_flow)
    monkeypatch.setattr(
        ann_service, "_build_announcement_form_payload",
        lambda widgets, field_values: "[]",
    )
    svc = ann_service.AnnouncementRequestService(db_session)
    svc._feishu_client = fake_client

    with pytest.raises(AppException) as exc_info:
        await svc.submit(
            requester=requester,
            description="One logical announcement",
            trace_id="trace-response-loss-1",
        )
    assert exc_info.value.code == "APPROVAL_SUBMISSION_RESULT_UNKNOWN"

    intent = await db_session.scalar(select(ApprovalSubmissionIntent).where(
        ApprovalSubmissionIntent.scope == "announcement_request",
        ApprovalSubmissionIntent.applicant_id == requester.id,
    ))
    assert intent is not None
    assert intent.state == "unknown"
    assert intent.provider_request_uuid == fake_client.instance_uuids[0]
    assert "never-persist-me" not in (intent.last_error or "")
    assert "app_secret" not in str(intent.payload).lower()
    assert "tenant_access_token" not in str(intent.payload).lower()

    row = await svc.submit(
        requester=requester,
        description="One logical announcement",
        trace_id="trace-response-loss-2",
    )

    assert row.feishu_instance_code == fake_client.instance_code
    assert fake_client.created == 2
    assert fake_client.instance_uuids == [
        intent.provider_request_uuid,
        intent.provider_request_uuid,
    ]
    await db_session.refresh(intent)
    assert intent.state == "completed"
    assert intent.external_id == fake_client.instance_code
    assert intent.approval_request_id is not None


@pytest.mark.asyncio
async def test_process_exit_after_provider_success_retries_local_finalize_only(
    db_session: AsyncSession, monkeypatch,
) -> None:
    requester = IdentityPerson(
        id=uuid.uuid4(),
        source_system="successfactors",
        hris_person_id=f"sf-{uuid.uuid4().hex[:8]}",
        display_name="Announcement crash requester",
        work_email=f"ann-crash-{uuid.uuid4().hex[:6]}@test.com",
        employment_status="active",
    )
    db_session.add(requester)
    await db_session.commit()
    fake_client = _FakeFeishuClient()

    async def _fake_flow(*args, **kwargs):
        return _FakeFlow()

    monkeypatch.setattr(ann_service, "get_approval_flow", _fake_flow)
    monkeypatch.setattr(
        ann_service, "_build_announcement_form_payload",
        lambda widgets, field_values: "[]",
    )
    original_finalize = (
        ann_service.AnnouncementRequestService._finalize_submission_intent
    )
    crashed = False

    async def _crash_once(self, intent_id, *, recovered):
        nonlocal crashed
        if not crashed:
            crashed = True
            raise SystemExit("simulated process exit")
        return await original_finalize(
            self, intent_id, recovered=recovered,
        )

    monkeypatch.setattr(
        ann_service.AnnouncementRequestService,
        "_finalize_submission_intent",
        _crash_once,
    )
    svc = ann_service.AnnouncementRequestService(db_session)
    svc._feishu_client = fake_client

    with pytest.raises(SystemExit):
        await svc.submit(
            requester=requester,
            description="Crash after provider accepted",
            trace_id="trace-process-crash-1",
        )
    intent = await db_session.scalar(select(ApprovalSubmissionIntent).where(
        ApprovalSubmissionIntent.scope == "announcement_request",
        ApprovalSubmissionIntent.applicant_id == requester.id,
    ))
    assert intent is not None
    assert intent.state == "provider_created"
    assert fake_client.created == 1

    row = await svc.submit(
        requester=requester,
        description="Crash after provider accepted",
        trace_id="trace-process-crash-2",
    )

    assert fake_client.created == 1
    assert row.feishu_instance_code == intent.external_id
    await db_session.refresh(intent)
    assert intent.state == "completed"


@pytest.mark.asyncio
async def test_reconciler_recovers_response_loss_without_user_retry(
    db_session: AsyncSession, monkeypatch,
) -> None:
    requester = IdentityPerson(
        id=uuid.uuid4(),
        source_system="successfactors",
        hris_person_id=f"sf-{uuid.uuid4().hex[:8]}",
        display_name="Announcement reconciler requester",
        work_email=f"ann-reconcile-{uuid.uuid4().hex[:6]}@test.com",
        employment_status="active",
    )
    db_session.add(requester)
    await db_session.commit()
    fake_client = _ResponseLossFeishuClient()

    async def _fake_flow(*args, **kwargs):
        return _FakeFlow()

    monkeypatch.setattr(ann_service, "get_approval_flow", _fake_flow)
    monkeypatch.setattr(
        ann_service, "_build_announcement_form_payload",
        lambda widgets, field_values: "[]",
    )
    svc = ann_service.AnnouncementRequestService(db_session)
    svc._feishu_client = fake_client
    with pytest.raises(AppException):
        await svc.submit(
            requester=requester,
            description="Recover me in the scheduled job",
            trace_id="trace-reconciler-loss",
        )

    reconciler = ann_service.AnnouncementRequestService(db_session)
    reconciler._feishu_client = fake_client
    await make_submission_intent_retries_due(db_session)
    summary = await reconciler.reconcile_submission_intents(
        min_age_minutes=0,
    )

    assert summary == {
        "checked": 1,
        "reconciled": 1,
        "still_unknown": 0,
        "failed": 0,
    }
    row = await db_session.scalar(select(AnnouncementRequest).where(
        AnnouncementRequest.person_id == str(requester.id),
    ))
    assert row is not None
    assert row.feishu_instance_code == fake_client.instance_code
    assert fake_client.instance_uuids[0] == fake_client.instance_uuids[1]


@pytest.mark.asyncio
async def test_concurrent_submits_share_one_provider_create(
    db_session: AsyncSession, monkeypatch,
) -> None:
    requester_id = uuid.uuid4()
    requester = IdentityPerson(
        id=requester_id,
        source_system="successfactors",
        hris_person_id=f"sf-{uuid.uuid4().hex[:8]}",
        display_name="Announcement concurrent requester",
        work_email=f"ann-concurrent-{uuid.uuid4().hex[:6]}@test.com",
        employment_status="active",
    )
    db_session.add(requester)
    await db_session.commit()
    fake_client = _FakeFeishuClient()
    process_lock = asyncio.Lock()
    session_factory = async_sessionmaker(
        db_session.bind,
        expire_on_commit=False,
        autoflush=False,
    )

    @asynccontextmanager
    async def _serialized_lock(*args, **kwargs):
        async with process_lock:
            yield "test-lock-token"

    async def _fake_flow(*args, **kwargs):
        return _FakeFlow()

    monkeypatch.setattr(
        "app.core.distributed_lock.distributed_lock",
        _serialized_lock,
    )
    monkeypatch.setattr(ann_service, "get_approval_flow", _fake_flow)
    monkeypatch.setattr(
        ann_service, "_build_announcement_form_payload",
        lambda widgets, field_values: "[]",
    )

    async def _submit(trace_id: str):
        async with session_factory() as session:
            person = await session.get(IdentityPerson, requester_id)
            assert person is not None
            service = ann_service.AnnouncementRequestService(session)
            service._feishu_client = fake_client
            return await service.submit(
                requester=person,
                description="Concurrent logical announcement",
                trace_id=trace_id,
            )

    results = await asyncio.gather(
        _submit("trace-concurrent-a"),
        _submit("trace-concurrent-b"),
        return_exceptions=True,
    )

    assert (
        sum(isinstance(result, AnnouncementRequest) for result in results) == 1
    ), repr(results)
    errors = [result for result in results if isinstance(result, AppException)]
    assert len(errors) == 1
    assert errors[0].code == "ANNOUNCEMENT_ALREADY_PENDING"
    assert fake_client.created == 1