from __future__ import annotations

import threading
import uuid
from concurrent.futures import ThreadPoolExecutor
from datetime import UTC, datetime
from decimal import Decimal

from savings_postgres_testkit import (  # noqa: F401 - 注册 pg_factory fixture
    _pg_factory_fixture,
)
from sqlalchemy import func, select, text
from sqlalchemy.exc import DBAPIError
from sqlalchemy.orm import Session, sessionmaker

from app.models.expense_case import BusinessEvent
from app.models.financial_connector import (
    FinancialConnectorConfig,
    FinancialConnectorConfigEvent,
    FinancialConnectorEvent,
    FinancialConnectorOperationalEvent,
)
from app.models.financial_record import ExpenseClaim
from app.models.tenant import Tenant
from app.schemas.financial_connector import (
    FinancialConnectorConfigLifecycleAction,
    FinancialEventEnvelope,
)
from app.services.expense_cases import ExpenseCaseService
from app.services.financial_connector_auth import (
    FinancialConnectorSecretResolver,
    sign_financial_event,
)
from app.services.financial_connector_config_lifecycle import (
    FinancialConnectorConfigConflictError,
    FinancialConnectorConfigLifecycleService,
)
from app.services.financial_connector_ingestion import (
    FinancialConnectorConflictError,
    FinancialConnectorIngestionService,
)
from app.services.financial_connector_operational_events import (
    FinancialConnectorOperationalContext,
    FinancialConnectorOperationalEventService,
    operational_event_candidate,
)


def test_concurrent_identical_settlement_has_one_fact_and_one_payment(
    pg_factory: sessionmaker[Session],
) -> None:
    seeded = _seed(pg_factory)
    timestamp = 1_800_000_000
    envelope = _envelope(seeded, event_id=f"settled-{uuid.uuid4().hex}")
    ready = threading.Barrier(2)

    def ingest_once() -> bool:
        with pg_factory() as db:
            ready.wait(timeout=5)
            result = _ingest(db, envelope, timestamp)
            db.commit()
            return result.replayed

    with ThreadPoolExecutor(max_workers=2) as pool:
        outcomes = [
            future.result(timeout=10)
            for future in (pool.submit(ingest_once), pool.submit(ingest_once))
        ]
    assert sorted(outcomes) == [False, True]
    with pg_factory() as db:
        claim = db.get(ExpenseClaim, seeded[1])
        assert claim is not None and claim.status == "paid"
        assert (
            db.scalar(
                select(func.count(FinancialConnectorEvent.id)).where(
                    FinancialConnectorEvent.tenant_id == seeded[0],
                    FinancialConnectorEvent.external_event_id == envelope.external_event_id,
                )
            )
            == 1
        )
        assert (
            db.scalar(
                select(func.count(BusinessEvent.id)).where(
                    BusinessEvent.tenant_id == seeded[0],
                    BusinessEvent.aggregate_id == seeded[1],
                    BusinessEvent.event_type == "payment_completed",
                )
            )
            == 1
        )

    refund = _envelope(
        seeded,
        event_id=f"refund-{uuid.uuid4().hex}",
        event_type="payment_refunded",
        origin_external_event_id=envelope.external_event_id,
    )
    with pg_factory() as db:
        _ingest(db, refund, timestamp + 1)
        db.commit()
        claim = db.get(ExpenseClaim, seeded[1])
        assert claim is not None and claim.status == "pending_payment"
    second_settlement = _envelope(
        seeded,
        event_id=f"settled-{uuid.uuid4().hex}",
    )
    with pg_factory() as db:
        _ingest(db, second_settlement, timestamp + 2)
        db.commit()
        claim = db.get(ExpenseClaim, seeded[1])
        assert claim is not None and claim.status == "paid"


def test_concurrent_conflicting_payload_has_one_winner_and_facts_are_append_only(
    pg_factory: sessionmaker[Session],
) -> None:
    seeded = _seed(pg_factory)
    timestamp = 1_800_000_000
    event_id = f"failed-{uuid.uuid4().hex}"
    first = _envelope(seeded, event_id=event_id, event_type="payment_failed")
    second = first.model_copy(
        update={"payload": {**first.payload, "failure_code": "different_failure"}}
    )
    ready = threading.Barrier(2)

    def ingest_once(envelope: FinancialEventEnvelope) -> str:
        with pg_factory() as db:
            ready.wait(timeout=5)
            try:
                _ingest(db, envelope, timestamp)
                db.commit()
                return "created"
            except FinancialConnectorConflictError as error:
                db.rollback()
                FinancialConnectorOperationalEventService(db).record(error.operational_event)
                db.commit()
                return "conflict"

    with ThreadPoolExecutor(max_workers=2) as pool:
        outcomes = [
            future.result(timeout=10)
            for future in (pool.submit(ingest_once, first), pool.submit(ingest_once, second))
        ]
    assert sorted(outcomes) == ["conflict", "created"]
    with pg_factory() as db:
        event = db.scalar(
            select(FinancialConnectorEvent).where(
                FinancialConnectorEvent.tenant_id == seeded[0],
                FinancialConnectorEvent.external_event_id == event_id,
            )
        )
        assert event is not None
        assert (
            db.scalar(
                select(func.count(FinancialConnectorOperationalEvent.id)).where(
                    FinancialConnectorOperationalEvent.tenant_id == seeded[0],
                    FinancialConnectorOperationalEvent.event_type == "payload_conflict",
                )
            )
            == 1
        )
        savepoint = db.begin_nested()
        try:
            try:
                db.execute(
                    text(
                        "UPDATE financial_connector_events SET error_code = 'tampered' "
                        "WHERE id = :event_id"
                    ),
                    {"event_id": event.id},
                )
            except DBAPIError:
                pass
            else:  # pragma: no cover - PostgreSQL trigger must reject
                raise AssertionError("append-only event unexpectedly accepted UPDATE")
        finally:
            savepoint.rollback()


def test_concurrent_operational_candidate_has_one_append_only_fact(
    pg_factory: sessionmaker[Session],
) -> None:
    tenant_id = f"tenant-operational-{uuid.uuid4().hex}"
    config_id = str(uuid.uuid4())
    with pg_factory() as db:
        db.add(
            FinancialConnectorConfig(
                id=config_id,
                tenant_id=tenant_id,
                provider="operational-bank",
                environment="production",
                key_version="v1",
                secret_ref=f"connector/{tenant_id}",
                allowed_event_types_json=["payment_settled"],
                clock_skew_seconds=300,
                status="active",
                created_by="postgres-test",
            )
        )
        db.commit()
    context = FinancialConnectorOperationalContext(
        tenant_id=tenant_id,
        config_id=config_id,
        provider="operational-bank",
        environment="production",
        request_fingerprint="hmac-sha256:" + "a" * 64,
        external_event_fingerprint="hmac-sha256:" + "b" * 64,
    )
    occurred_at = datetime.now(UTC)
    candidate = operational_event_candidate(
        context,
        event_type="replay",
        reason_code="duplicate_external_event",
        occurred_at=occurred_at,
    )
    ready = threading.Barrier(2)

    def record_once() -> None:
        with pg_factory() as db:
            ready.wait(timeout=5)
            FinancialConnectorOperationalEventService(db).record(candidate)
            db.commit()

    with ThreadPoolExecutor(max_workers=2) as pool:
        for future in (pool.submit(record_once), pool.submit(record_once)):
            future.result(timeout=10)

    with pg_factory() as db:
        rows = list(
            db.scalars(
                select(FinancialConnectorOperationalEvent).where(
                    FinancialConnectorOperationalEvent.tenant_id == tenant_id
                )
            ).all()
        )
        assert len(rows) == 1
        savepoint = db.begin_nested()
        try:
            try:
                db.execute(
                    text("DELETE FROM financial_connector_operational_events WHERE id = :event_id"),
                    {"event_id": rows[0].id},
                )
            except DBAPIError:
                pass
            else:  # pragma: no cover - PostgreSQL trigger must reject
                raise AssertionError("append-only operational event accepted DELETE")
        finally:
            savepoint.rollback()


def test_concurrent_config_activation_has_one_version_winner(
    pg_factory: sessionmaker[Session],
) -> None:
    tenant_id = f"tenant-config-{uuid.uuid4().hex}"
    config_id = str(uuid.uuid4())
    secret_ref = f"connector/{tenant_id}"
    secret = f"strong-secret-{tenant_id}"
    with pg_factory() as db:
        db.add(
            FinancialConnectorConfig(
                id=config_id,
                tenant_id=tenant_id,
                provider="versioned-bank",
                environment="production",
                key_version="v1",
                secret_ref=secret_ref,
                allowed_event_types_json=["payment_settled"],
                clock_skew_seconds=300,
                status="disabled",
                version=1,
                created_by="postgres-test",
            )
        )
        db.commit()
    ready = threading.Barrier(2)

    def activate_once(index: int) -> str:
        with pg_factory() as db:
            ready.wait(timeout=5)
            try:
                FinancialConnectorConfigLifecycleService(
                    db,
                    secrets=FinancialConnectorSecretResolver({secret_ref: secret}),
                ).activate(
                    tenant_id=tenant_id,
                    config_id=config_id,
                    payload=FinancialConnectorConfigLifecycleAction(
                        expected_version=1,
                        request_id=f"config-activate-{index:03d}",
                        reason="并发激活只能有一个版本胜者。",
                    ),
                    actor_id=f"postgres-operator-{index}",
                )
                db.commit()
                return "activated"
            except FinancialConnectorConfigConflictError:
                db.rollback()
                return "conflict"

    with ThreadPoolExecutor(max_workers=2) as pool:
        outcomes = [
            future.result(timeout=10)
            for future in (pool.submit(activate_once, 1), pool.submit(activate_once, 2))
        ]
    assert sorted(outcomes) == ["activated", "conflict"]
    with pg_factory() as db:
        config = db.get(FinancialConnectorConfig, config_id)
        assert config is not None
        assert config.status == "active" and config.version == 2
        assert (
            db.scalar(
                select(func.count(FinancialConnectorConfigEvent.id)).where(
                    FinancialConnectorConfigEvent.tenant_id == tenant_id,
                    FinancialConnectorConfigEvent.config_id == config_id,
                    FinancialConnectorConfigEvent.action == "activated",
                )
            )
            == 1
        )


def _seed(factory: sessionmaker[Session]) -> tuple[str, str, str, str]:
    tenant_id = f"tenant-connector-{uuid.uuid4().hex}"
    claim_id = str(uuid.uuid4())
    config_id = str(uuid.uuid4())
    with factory() as db:
        db.add(
            Tenant(
                tenant_id=tenant_id,
                tenant_code=tenant_id,
                name="连接器并发探针租户",
                status="active",
            )
        )
        db.flush()
        claim = ExpenseClaim(
            id=claim_id,
            tenant_id=tenant_id,
            claim_no=f"BX-PG-{uuid.uuid4().hex[:12]}",
            employee_name="PostgreSQL 测试员工",
            department_name="测试部",
            expense_type="travel",
            reason="连接器并发验证",
            location="上海",
            amount=Decimal("66.00"),
            currency="CNY",
            invoice_count=1,
            occurred_at=datetime.now(UTC),
            submitted_at=datetime.now(UTC),
            status="pending_payment",
            approval_stage="待付款",
            risk_flags_json=[],
        )
        db.add(claim)
        db.flush()
        ExpenseCaseService(db).ensure_case_for_claim(claim, tenant_id=tenant_id)
        db.add(
            FinancialConnectorConfig(
                id=config_id,
                tenant_id=tenant_id,
                provider="postgres-bank",
                environment="production",
                key_version="v1",
                secret_ref=f"connector/{tenant_id}",
                allowed_event_types_json=[
                    "payment_settled",
                    "payment_failed",
                    "payment_refunded",
                ],
                clock_skew_seconds=300,
                status="active",
                created_by="postgres-test",
            )
        )
        db.commit()
    return tenant_id, claim_id, config_id, claim.claim_no


def _envelope(
    seeded: tuple[str, str, str, str],
    *,
    event_id: str,
    event_type: str = "payment_settled",
    origin_external_event_id: str | None = None,
) -> FinancialEventEnvelope:
    tenant_id, claim_id, _, claim_no = seeded
    return FinancialEventEnvelope(
        tenant_id=tenant_id,
        external_event_id=event_id,
        event_type=event_type,
        occurred_at=datetime.now(UTC),
        correlation_id=f"correlation-{event_id}"[:64],
        payload={
            "claim_id": claim_id,
            "claim_reference": claim_no,
            "amount": "66.00",
            "currency": "CNY",
            "external_payment_reference": f"PAY-{event_id}",
            "failure_code": "provider_rejected",
            **(
                {"origin_external_event_id": origin_external_event_id}
                if origin_external_event_id
                else {}
            ),
        },
    )


def _ingest(
    db: Session,
    envelope: FinancialEventEnvelope,
    timestamp: int,
):
    claim = db.get(ExpenseClaim, str(envelope.payload["claim_id"]))
    assert claim is not None
    envelope = envelope.model_copy(
        update={"payload": {**envelope.payload, "claim_reference": claim.claim_no}}
    )
    secret_ref = f"connector/{envelope.tenant_id}"
    secret = f"secret-{envelope.tenant_id}"
    signature = sign_financial_event(
        envelope,
        timestamp=timestamp,
        secret=secret,
        tenant_id=envelope.tenant_id,
        provider="postgres-bank",
        key_version="v1",
    )
    return FinancialConnectorIngestionService(
        db,
        secrets=FinancialConnectorSecretResolver({secret_ref: secret}),
        now_epoch=timestamp,
    ).ingest(
        envelope,
        tenant_header=envelope.tenant_id,
        provider_header="postgres-bank",
        key_version_header="v1",
        timestamp_header=str(timestamp),
        signature_header=signature,
    )