"""Durable Feishu-create regressions for identity consumers and domains."""

from __future__ import annotations

import asyncio
import uuid
from contextlib import asynccontextmanager
from datetime import UTC, datetime
from unittest.mock import AsyncMock

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

from app.approval.models import ApprovalRequest, ApprovalSubmissionIntent
from app.audit.models import AuditLog
from app.core.exceptions import AppException
from app.domains.domain_registration import service as domain_service
from app.domains.domain_registration.models import DomainRegistration
from app.identity.consumers import request_service as consumer_service
from app.identity.consumers.models import IdentityConsumerRequest
from app.identity.consumers.request_schemas import ConsumerRequestCreate
from app.identity.person_models import IdentityPerson
from tests.support.approval_intents import make_submission_intent_retries_due


class _Flow:
    external_code = "approval-code"


class _Provider:
    def __init__(
        self,
        *,
        lose_first_response: bool = False,
        cancel_fails: bool = False,
    ) -> None:
        self.lose_first_response = lose_first_response
        self.cancel_fails = cancel_fails
        self.created = 0
        self.cancelled = 0
        self.uuids: list[str] = []
        self.accepted: dict[str, str] = {}

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

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

    async def create_instance(self, *, instance_uuid=None, **kwargs) -> dict:
        assert instance_uuid
        self.created += 1
        self.uuids.append(instance_uuid)
        if instance_uuid in self.accepted:
            raise AppException(
                code="FEISHU_API_ERROR",
                message="Feishu API error 60012: duplicate uuid",
                status_code=502,
            )
        instance_code = f"instance-{len(self.accepted) + 1}"
        self.accepted[instance_uuid] = instance_code
        if self.lose_first_response and len(self.accepted) == 1:
            raise httpx.ReadTimeout("accepted; app_secret=must-not-persist")
        return {"instance_code": instance_code, "uuid": instance_uuid}

    async def find_instance_by_uuid(self, *, instance_uuid, **kwargs) -> dict:
        return {
            "instance_code": self.accepted[instance_uuid],
            "uuid": instance_uuid,
            "status": "PENDING",
        }

    async def get_instance_detail(self, instance_code: str) -> dict:
        return {"instance_code": instance_code, "status": "PENDING"}

    async def cancel_instance(self, **kwargs) -> None:
        self.cancelled += 1
        if self.cancel_fails:
            raise AppException(
                code="FEISHU_API_ERROR",
                message="Feishu API error 60009: withdraw disabled",
                status_code=502,
            )


async def _flow(*args, **kwargs):
    return _Flow()


def _wire(kind: str, monkeypatch, provider: _Provider) -> None:
    module = domain_service if kind == "domain" else consumer_service
    monkeypatch.setattr(module, "get_approval_flow", _flow)
    if kind == "domain":
        monkeypatch.setattr(
            domain_service, "_build_domain_form_payload", lambda *args: "[]",
        )


def _service(kind: str, db: AsyncSession, provider: _Provider):
    service = (
        domain_service.DomainRegistrationService(db)
        if kind == "domain"
        else consumer_service.ConsumerRequestService(db)
    )
    service._feishu_client = provider
    return service


async def _submit(
    kind: str,
    service,
    requester: IdentityPerson,
    trace: str,
    logical_key: str,
):
    if kind == "domain":
        return await service.submit(
            requester=requester,
            domain_name=f"{logical_key}.example.com",
            purpose="Product documentation",
            trace_id=trace,
        )
    return await service.submit(
        requester=requester,
        body=ConsumerRequestCreate(
            name=f"Consumer {logical_key}",
            slug=logical_key,
            scopes=["directory.persons:read"],
            description="Directory synchronization",
        ),
        trace_id=trace,
    )


async def _requester(db: AsyncSession) -> IdentityPerson:
    person = IdentityPerson(
        id=uuid.uuid4(),
        source_system="successfactors",
        hris_person_id=f"sf-{uuid.uuid4().hex[:8]}",
        display_name="Durable intent requester",
        work_email=f"intent-{uuid.uuid4().hex[:8]}@test.com",
        employment_status="active",
    )
    db.add(person)
    await db.commit()
    return person


@pytest.mark.parametrize("kind", ["domain", "consumer"])
@pytest.mark.asyncio
async def test_response_loss_retry_reuses_uuid_and_finalizes_once(
    kind: str, db_session: AsyncSession, monkeypatch,
) -> None:
    requester = await _requester(db_session)
    provider = _Provider(lose_first_response=True)
    _wire(kind, monkeypatch, provider)
    service = _service(kind, db_session, provider)

    with pytest.raises(AppException) as exc_info:
        await _submit(kind, service, requester, "trace-loss-1", "loss-key")
    assert exc_info.value.code == "APPROVAL_SUBMISSION_RESULT_UNKNOWN"
    intent = await db_session.scalar(
        select(ApprovalSubmissionIntent).where(
            ApprovalSubmissionIntent.applicant_id == requester.id,
            ApprovalSubmissionIntent.scope == (
                "domain_registration_request"
                if kind == "domain" else "identity_consumer_request"
            ),
        )
    )
    assert intent is not None
    assert "must-not-persist" not in (intent.last_error or "")

    row = await _submit(kind, service, requester, "trace-loss-2", "loss-key")
    assert row.feishu_instance_code == next(iter(provider.accepted.values()))
    assert provider.uuids == [intent.provider_request_uuid] * 2
    assert await db_session.scalar(select(func.count()).select_from(ApprovalRequest)) == 1


@pytest.mark.parametrize("kind", ["domain", "consumer"])
@pytest.mark.asyncio
async def test_process_exit_after_provider_commit_finalizes_without_resubmit(
    kind: str, db_session: AsyncSession, monkeypatch,
) -> None:
    requester = await _requester(db_session)
    provider = _Provider()
    _wire(kind, monkeypatch, provider)
    service_class = (
        domain_service.DomainRegistrationService
        if kind == "domain" else consumer_service.ConsumerRequestService
    )
    original = service_class._finalize_submission_intent
    crashed = False

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

    monkeypatch.setattr(service_class, "_finalize_submission_intent", _crash_once)
    service = _service(kind, db_session, provider)
    with pytest.raises(SystemExit):
        await _submit(kind, service, requester, "trace-crash-1", "crash-key")
    row = await _submit(kind, service, requester, "trace-crash-2", "crash-key")
    assert row.feishu_instance_code
    assert provider.created == 1


@pytest.mark.parametrize("kind", ["domain", "consumer"])
@pytest.mark.asyncio
async def test_cron_recovers_unknown_without_user_retry(
    kind: str, db_session: AsyncSession, monkeypatch,
) -> None:
    requester = await _requester(db_session)
    provider = _Provider(lose_first_response=True)
    _wire(kind, monkeypatch, provider)
    service = _service(kind, db_session, provider)
    with pytest.raises(AppException):
        await _submit(kind, service, requester, "trace-cron", "cron-key")

    await make_submission_intent_retries_due(db_session)
    summary = await service.reconcile_submission_intents(min_age_minutes=0)
    assert summary["checked"] == 1
    assert summary["reconciled"] == 1
    assert summary["failed"] == 0
    assert provider.uuids[0] == provider.uuids[1]


@pytest.mark.parametrize("kind", ["domain", "consumer"])
@pytest.mark.asyncio
async def test_dual_session_submit_creates_one_provider_instance(
    kind: str, db_session: AsyncSession, monkeypatch,
) -> None:
    requester = await _requester(db_session)
    requester_id = requester.id
    provider = _Provider()
    _wire(kind, monkeypatch, provider)
    process_lock = asyncio.Lock()
    factory = async_sessionmaker(
        db_session.bind, expire_on_commit=False, autoflush=False,
    )

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

    monkeypatch.setattr("app.core.distributed_lock.distributed_lock", _lock)

    async def _one(trace: str):
        async with factory() as session:
            person = await session.get(IdentityPerson, requester_id)
            assert person is not None
            return await _submit(
                kind, _service(kind, session, provider), person, trace, "same-key",
            )

    results = await asyncio.gather(
        _one("trace-a"), _one("trace-b"), return_exceptions=True,
    )
    successful = [item for item in results if not isinstance(item, Exception)]
    assert successful, results
    assert len({str(item.id) for item in successful}) == 1
    assert sum(isinstance(item, AppException) for item in results) <= 1, results
    assert provider.created == 1
    model = DomainRegistration if kind == "domain" else IdentityConsumerRequest
    assert await db_session.scalar(select(func.count()).select_from(model)) == 1


@pytest.mark.parametrize("kind", ["domain", "consumer"])
@pytest.mark.asyncio
async def test_finalization_is_atomic_and_retryable(
    kind: str, db_session: AsyncSession, monkeypatch,
) -> None:
    requester = await _requester(db_session)
    requester_id = requester.id
    provider = _Provider()
    _wire(kind, monkeypatch, provider)
    service = _service(kind, db_session, provider)
    original_log = service.audit.log
    failed = False

    async def _fail_once(**kwargs):
        nonlocal failed
        if not failed:
            failed = True
            raise RuntimeError("audit insert failed")
        return await original_log(**kwargs)

    service.audit.log = _fail_once
    with pytest.raises(RuntimeError, match="audit insert failed"):
        await _submit(kind, service, requester, "trace-atomic-1", "atomic-key")
    await db_session.rollback()
    model = DomainRegistration if kind == "domain" else IdentityConsumerRequest
    assert await db_session.scalar(select(func.count()).select_from(model)) == 0
    assert await db_session.scalar(select(func.count()).select_from(ApprovalRequest)) == 0

    requester = await db_session.get(IdentityPerson, requester_id)
    assert requester is not None
    row = await _submit(kind, service, requester, "trace-atomic-2", "atomic-key")
    assert row.feishu_instance_code
    assert provider.created == 1


@pytest.mark.asyncio
async def test_identity_unknown_slug_does_not_absorb_another_slug(
    db_session: AsyncSession, monkeypatch,
) -> None:
    requester = await _requester(db_session)
    provider = _Provider(lose_first_response=True)
    _wire("consumer", monkeypatch, provider)
    service = _service("consumer", db_session, provider)
    with pytest.raises(AppException):
        await _submit("consumer", service, requester, "trace-a", "slug-a")

    row = await _submit("consumer", service, requester, "trace-b", "slug-b")
    assert row.slug == "slug-b"
    intents = list((await db_session.scalars(
        select(ApprovalSubmissionIntent).where(
            ApprovalSubmissionIntent.applicant_id == requester.id,
            ApprovalSubmissionIntent.scope == "identity_consumer_request",
        )
    )).all())
    assert len(intents) == 2
    assert len({intent.provider_request_uuid for intent in intents}) == 2


@pytest.mark.parametrize("kind", ["domain", "consumer"])
@pytest.mark.asyncio
async def test_leaver_after_provider_create_is_correlated_and_compensated(
    kind: str, db_session: AsyncSession, monkeypatch,
) -> None:
    requester = await _requester(db_session)
    requester_id = requester.id
    provider = _Provider()
    _wire(kind, monkeypatch, provider)
    service_class = (
        domain_service.DomainRegistrationService
        if kind == "domain" else consumer_service.ConsumerRequestService
    )
    original = service_class._finalize_submission_intent
    factory = async_sessionmaker(
        db_session.bind, expire_on_commit=False, autoflush=False,
    )
    changed = False

    async def _mark_leaver(self, intent_id, *, recovered):
        nonlocal changed
        if not changed:
            changed = True
            async with factory() as session:
                person = await session.get(IdentityPerson, requester_id)
                assert person is not None
                person.employment_status = "leaver_pending"
                await session.commit()
        return await original(self, intent_id, recovered=recovered)

    monkeypatch.setattr(service_class, "_finalize_submission_intent", _mark_leaver)
    service = _service(kind, db_session, provider)
    with pytest.raises(AppException):
        await _submit(kind, service, requester, "trace-leaver", "leaver-key")
    await db_session.rollback()
    intent = await db_session.scalar(
        select(ApprovalSubmissionIntent).where(
            ApprovalSubmissionIntent.applicant_id == requester_id,
        )
    )
    assert intent is not None and intent.state == "compensated"
    assert intent.approval_request_id is not None
    assert provider.cancelled == 1


@pytest.mark.parametrize("kind", ["domain", "consumer"])
@pytest.mark.asyncio
async def test_failed_compensation_is_scheduled_then_reconciled(
    kind: str, db_session: AsyncSession, monkeypatch,
) -> None:
    requester = await _requester(db_session)
    requester_id = requester.id
    provider = _Provider(cancel_fails=True)
    _wire(kind, monkeypatch, provider)
    module = domain_service if kind == "domain" else consumer_service
    notify = AsyncMock(return_value=True)
    monkeypatch.setattr(module, "notify_feishu_cancel_desync", notify)
    service_class = (
        domain_service.DomainRegistrationService
        if kind == "domain" else consumer_service.ConsumerRequestService
    )
    original = service_class._finalize_submission_intent
    factory = async_sessionmaker(
        db_session.bind, expire_on_commit=False, autoflush=False,
    )
    changed = False

    async def _mark_leaver(self, intent_id, *, recovered):
        nonlocal changed
        if not changed:
            changed = True
            async with factory() as session:
                person = await session.get(IdentityPerson, requester_id)
                assert person is not None
                person.employment_status = "leaver_pending"
                await session.commit()
        return await original(self, intent_id, recovered=recovered)

    monkeypatch.setattr(service_class, "_finalize_submission_intent", _mark_leaver)
    service = _service(kind, db_session, provider)
    with pytest.raises(AppException):
        await _submit(kind, service, requester, "trace-comp-fail", "comp-key")

    intent = await db_session.scalar(select(ApprovalSubmissionIntent).where(
        ApprovalSubmissionIntent.applicant_id == requester_id,
    ))
    assert intent is not None and intent.state == "compensation_pending"
    assert intent.next_attempt_at is not None
    scheduled = intent.next_attempt_at
    if scheduled.tzinfo is None:
        scheduled = scheduled.replace(tzinfo=UTC)
    assert scheduled > datetime.now(UTC)
    assert provider.cancelled == 1
    assert notify.await_count == 1
    assert await db_session.scalar(
        select(func.count()).select_from(AuditLog).where(
            AuditLog.action == "approval.submission_intent.compensation_pending",
        )
    ) == 1

    provider.cancel_fails = False
    await make_submission_intent_retries_due(db_session)
    summary = await service.reconcile_submission_intents(min_age_minutes=0)
    assert summary["compensated"] == 1
    assert summary["still_unknown"] == 0
    await db_session.refresh(intent)
    assert intent.state == "compensated"
    assert provider.cancelled == 2