from __future__ import annotations

import threading
import uuid
from concurrent.futures import ThreadPoolExecutor

import pytest
from savings_postgres_testkit import (
    _pg_factory_fixture,  # noqa: F401 - 注册名为 pg_factory 的 pytest fixture
    _seed_opportunity,
    _seed_payment_case,
    _seed_pending_realization,
    _SeededRealization,
    _user,
)
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_record import ExpenseClaim
from app.models.savings import (
    SavingsEvent,
    SavingsEvidenceLink,
    SavingsOpportunity,
    SavingsRealization,
)
from app.schemas.savings import (
    SavingsOpportunityActionCreate,
    SavingsRealizationActionCreate,
)
from app.services.savings_access_policy import SavingsPermissionError
from app.services.savings_actions import SavingsActionService
from app.services.savings_realization import (
    SavingsRealizationError,
    SavingsRealizationService,
)


def test_concurrent_identical_request_replays_one_immutable_event(
    pg_factory: sessionmaker[Session],
) -> None:
    seed = _seed_opportunity(pg_factory, status="identified", owner_id="owner-idempotent")
    actor = _user("owner-idempotent", seed.tenant_id, employee_id="owner-idempotent")
    payload = SavingsOpportunityActionCreate(
        action="accept",
        request_id=f"accept-concurrent-{seed.suffix}",
        expected_version=1,
        comment="并发接受同一个节省机会",
    )
    ready = threading.Barrier(2)

    def accept_once():
        with pg_factory() as db:
            ready.wait(timeout=5)
            return SavingsActionService(db).execute(seed.opportunity_id, payload, actor).response

    with ThreadPoolExecutor(max_workers=2) as pool:
        results = [
            future.result(timeout=10)
            for future in (pool.submit(accept_once), pool.submit(accept_once))
        ]

    assert sorted(item.replayed for item in results) == [False, True]
    assert results[0].opportunity.model_dump(mode="json") == results[1].opportunity.model_dump(
        mode="json"
    )
    with pg_factory() as db:
        opportunity = db.get(SavingsOpportunity, seed.opportunity_id)
        assert opportunity is not None and opportunity.status == "accepted"
        assert opportunity.version == 2
        assert (
            db.scalar(
                select(func.count())
                .select_from(SavingsEvent)
                .where(
                    SavingsEvent.tenant_id == seed.tenant_id,
                    SavingsEvent.actor_id == "owner-idempotent",
                    SavingsEvent.request_id == payload.request_id,
                )
            )
            == 1
        )


def test_same_benefit_concurrent_confirmation_has_one_canonical_winner(
    pg_factory: sessionmaker[Session],
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    benefit_key = f"benefit-shared-{uuid.uuid4().hex}"
    first = _seed_pending_realization(
        pg_factory,
        benefit_key=benefit_key,
        owner_id="owner-canonical-a",
        recorder_id="payer-canonical-a",
    )
    second = _seed_pending_realization(
        pg_factory,
        benefit_key=benefit_key,
        owner_id="owner-canonical-b",
        recorder_id="payer-canonical-b",
    )
    ready = threading.Barrier(2)
    original_confirm = SavingsRealizationService._confirm

    def confirm_after_duplicate_probe(self, *args, **kwargs):
        original_confirm(self, *args, **kwargs)
        ready.wait(timeout=5)

    monkeypatch.setattr(SavingsRealizationService, "_confirm", confirm_after_duplicate_probe)

    def confirm_once(seed: _SeededRealization, actor_id: str):
        payload = SavingsRealizationActionCreate(
            action="confirm",
            request_id=f"confirm-race-{seed.suffix}",
            expected_version=1,
            comment="并发确认相同经济收益",
        )
        with pg_factory() as db:
            try:
                return (
                    SavingsRealizationService(db)
                    .execute_action(
                        seed.realization_id,
                        payload,
                        _user(actor_id, seed.tenant_id, employee_id=actor_id, roles=["finance"]),
                    )
                    .response.realization.id
                )
            except SavingsRealizationError as error:
                return error

    with ThreadPoolExecutor(max_workers=2) as pool:
        outcomes = [
            future.result(timeout=10)
            for future in (
                pool.submit(confirm_once, first, "finance-canonical-a"),
                pool.submit(confirm_once, second, "finance-canonical-b"),
            )
        ]

    assert sum(isinstance(item, str) for item in outcomes) == 1
    assert sum(isinstance(item, SavingsRealizationError) for item in outcomes) == 1
    with pg_factory() as db:
        realizations = list(
            db.scalars(
                select(SavingsRealization).where(
                    SavingsRealization.tenant_id == first.tenant_id,
                    SavingsRealization.benefit_key == benefit_key,
                )
            )
        )
        assert sorted((row.status, row.dedupe_status) for row in realizations) == [
            ("finance_confirmed", "canonical"),
            ("pending_confirmation", "pending_review"),
        ]
        winner = next(row for row in realizations if row.status == "finance_confirmed")
        loser = next(row for row in realizations if row.status == "pending_confirmation")
        loser_evidence = list(
            db.scalars(
                select(SavingsEvidenceLink).where(
                    SavingsEvidenceLink.realization_id == loser.id,
                )
            )
        )
        assert loser_evidence and {item.verification_status for item in loser_evidence} == {
            "unverified"
        }
        assert (
            db.scalar(
                select(func.count())
                .select_from(SavingsEvent)
                .where(
                    SavingsEvent.realization_id.in_([first.realization_id, second.realization_id]),
                    SavingsEvent.action == "confirm",
                )
            )
            == 1
        )
        assert (
            db.scalar(
                select(func.count())
                .select_from(BusinessEvent)
                .where(
                    BusinessEvent.aggregate_type == "savings_realization",
                    BusinessEvent.aggregate_id == winner.id,
                    BusinessEvent.event_type == "saving_confirmed",
                )
            )
            == 1
        )
        assert (
            db.scalar(
                select(func.count())
                .select_from(SavingsRealization)
                .where(
                    SavingsRealization.tenant_id == first.tenant_id,
                    SavingsRealization.benefit_key == benefit_key,
                    SavingsRealization.realization_type == "actual",
                    SavingsRealization.dedupe_status == "canonical",
                    SavingsRealization.status == "finance_confirmed",
                )
            )
            == 1
        )


def test_cross_tenant_action_is_rejected_without_side_effect(
    pg_factory: sessionmaker[Session],
) -> None:
    seed = _seed_opportunity(
        pg_factory, tenant_id="tenant-savings-private", owner_id="owner-private"
    )
    realization_seed = _seed_pending_realization(
        pg_factory,
        tenant_id="tenant-savings-private",
        owner_id="owner-private-realization",
        recorder_id="payer-private-realization",
    )
    request_id = f"cross-tenant-{seed.suffix}"
    realization_request_id = f"cross-tenant-confirm-{realization_seed.suffix}"
    with pg_factory() as db:
        with pytest.raises(LookupError, match="不存在"):
            SavingsActionService(db).execute(
                seed.opportunity_id,
                SavingsOpportunityActionCreate(
                    action="accept",
                    request_id=request_id,
                    expected_version=1,
                    comment="尝试跨租户改变机会",
                ),
                _user("finance-outsider", "tenant-savings-outsider", roles=["finance"]),
            )
        with pytest.raises(LookupError, match="不存在"):
            SavingsRealizationService(db).execute_action(
                realization_seed.realization_id,
                SavingsRealizationActionCreate(
                    action="confirm",
                    request_id=realization_request_id,
                    expected_version=1,
                    comment="尝试跨租户确认实际节省",
                ),
                _user("finance-outsider", "tenant-savings-outsider", roles=["finance"]),
            )

    with pg_factory() as db:
        opportunity = db.get(SavingsOpportunity, seed.opportunity_id)
        assert opportunity is not None and opportunity.status == "identified"
        assert opportunity.version == 1
        realization = db.get(SavingsRealization, realization_seed.realization_id)
        assert realization is not None and realization.status == "pending_confirmation"
        assert realization.version == 1
        assert (
            db.scalar(
                select(func.count())
                .select_from(SavingsEvent)
                .where(SavingsEvent.request_id.in_([request_id, realization_request_id]))
            )
            == 0
        )


def test_confirmation_requires_actor_independent_from_owner_and_recorder(
    pg_factory: sessionmaker[Session],
) -> None:
    seed = _seed_pending_realization(
        pg_factory,
        owner_id="finance-owner-independent",
        recorder_id="finance-recorder-independent",
    )

    for actor_id in ("finance-owner-independent", "finance-recorder-independent"):
        with pg_factory() as db:
            with pytest.raises(SavingsPermissionError, match="不能确认自己"):
                SavingsRealizationService(db).execute_action(
                    seed.realization_id,
                    SavingsRealizationActionCreate(
                        action="confirm",
                        request_id=f"self-confirm-{actor_id}-{seed.suffix}",
                        expected_version=1,
                        comment="不应允许自我确认",
                    ),
                    _user(actor_id, seed.tenant_id, employee_id=actor_id, roles=["finance"]),
                )

    confirmer_id = "finance-independent-reviewer"
    with pg_factory() as db:
        result = SavingsRealizationService(db).execute_action(
            seed.realization_id,
            SavingsRealizationActionCreate(
                action="confirm",
                request_id=f"independent-confirm-{seed.suffix}",
                expected_version=1,
                comment="独立复核付款与政策证据",
            ),
            _user(confirmer_id, seed.tenant_id, employee_id=confirmer_id, roles=["finance"]),
        )
        assert result.response.realization.status == "finance_confirmed"
        assert result.response.realization.finance_confirmer_id == confirmer_id
        assert result.response.realization.recorded_by_id != confirmer_id
        evidence_statuses = set(
            db.scalars(
                select(SavingsEvidenceLink.verification_status).where(
                    SavingsEvidenceLink.realization_id == seed.realization_id,
                )
            )
        )
        assert evidence_statuses == {"verified"}


def test_concurrent_same_payment_event_persists_one_realization(
    pg_factory: sessionmaker[Session],
) -> None:
    seed = _seed_payment_case(pg_factory)
    ready = threading.Barrier(2)

    def realize_once():
        with pg_factory() as db:
            claim = db.get(ExpenseClaim, seed.claim_id)
            event = db.get(BusinessEvent, seed.payment_event_id)
            assert claim is not None and event is not None
            ready.wait(timeout=5)
            created = SavingsRealizationService(db).realize_paid_claim(
                claim,
                event,
                _user(
                    "payment-actor", seed.tenant_id, employee_id="payment-actor", roles=["finance"]
                ),
            )
            db.commit()
            return len(created)

    with ThreadPoolExecutor(max_workers=2) as pool:
        outcomes = [
            future.result(timeout=10)
            for future in (pool.submit(realize_once), pool.submit(realize_once))
        ]

    assert sorted(outcomes) == [0, 1]
    with pg_factory() as db:
        rows = list(
            db.scalars(
                select(SavingsRealization).where(
                    SavingsRealization.tenant_id == seed.tenant_id,
                    SavingsRealization.opportunity_id == seed.opportunity_id,
                    SavingsRealization.business_event_id == seed.payment_event_id,
                )
            )
        )
        assert len(rows) == 1
        assert rows[0].realization_key == (f"payment:{seed.payment_event_id}:{seed.opportunity_id}")
        opportunity = db.get(SavingsOpportunity, seed.opportunity_id)
        assert opportunity is not None and opportunity.status == "realized"
        assert opportunity.version == 2
        assert (
            db.scalar(
                select(func.count())
                .select_from(BusinessEvent)
                .where(
                    BusinessEvent.tenant_id == seed.tenant_id,
                    BusinessEvent.aggregate_type == "savings_realization",
                    BusinessEvent.aggregate_id == rows[0].id,
                    BusinessEvent.event_type == "saving_action_completed",
                    BusinessEvent.causation_id == seed.payment_event_id,
                )
            )
            == 1
        )


def test_savings_event_is_append_only_at_database_boundary(
    pg_factory: sessionmaker[Session],
) -> None:
    seed = _seed_opportunity(pg_factory, owner_id="owner-append-only")
    request_id = f"append-only-{seed.suffix}"
    with pg_factory() as db:
        response = (
            SavingsActionService(db)
            .execute(
                seed.opportunity_id,
                SavingsOpportunityActionCreate(
                    action="accept",
                    request_id=request_id,
                    expected_version=1,
                    comment="创建不可变审计事件",
                ),
                _user("owner-append-only", seed.tenant_id, employee_id="owner-append-only"),
            )
            .response
        )
        event_id = response.event.id

    for statement in (
        "UPDATE savings_events SET action = 'tampered' WHERE id = :event_id",
        "DELETE FROM savings_events WHERE id = :event_id",
    ):
        with pg_factory() as db:
            with pytest.raises(DBAPIError, match="append-only"):
                db.execute(text(statement), {"event_id": event_id})
                db.commit()
            db.rollback()

    with pg_factory() as db:
        event = db.get(SavingsEvent, event_id)
        assert event is not None and event.action == "accept"
        assert event.response_json["event"]["id"] == event_id