"""Unified finance spend ledger — rollup, dedup, FX, close, reconcile.

The production-grade correctness contract of PR1:

  * **dedup invariant** — AI-whitelisted cloud spend is counted ONCE
    under ``ai`` (excluded from ``cloud``); AI-category software seats
    are excluded from ``software``. ``cloud + software + ai`` partition
    the consumption universe with no overlap.
  * **USD normalisation** — native-currency software / procurement spend
    converts at the month-frozen FX rate; cloud / AI are native USD.
  * **basis separation** — procurement is a distinct ``basis`` and is
    NOT summed into the consumption grand total.
  * **internal reconciliation** — the ledger agrees with every source
    domain (the trust gate finance relies on).
  * **close / reopen / restatement** — a closed period freezes its
    facts non-provisional; reopen makes them provisional again; a
    recompute that finds drift on a closed period logs a restatement
    without moving the published number.
"""
from __future__ import annotations

import uuid
from datetime import UTC, date, datetime
from decimal import Decimal

import pytest
import pytest_asyncio
from sqlalchemy import func, select

from app.core.coordinators import finance_spend_coordinator as coord
from app.domains.asset.models import AssetSoftwareCatalog, SoftwareMonthlyBill
from app.domains.cmdb.models import CmdbCostFact
from app.domains.finance import service as finance_service
from app.domains.finance.models import (
    FinancePeriodClose,
    FinanceRestatement,
    FinanceSpendFact,
)
from app.domains.procurement.models import ProcurementOrder, ProcurementSupplier
from app.identity.person_models import IdentityPerson
from app.integrations.models import Integration, PlatformSetting

PERIOD = "2026-05"
# Test month FX: 1 CNY = 0.10 USD (within the plausibility band), chosen
# so native CNY amounts convert to clean USD figures.
CNY_RATE = Decimal("0.10")


@pytest_asyncio.fixture
async def seeded(db_session):
    """Seed one period of spend across all four streams.

    Cloud:        EC2 $100 (non-AI) + Bedrock $40 (AI-whitelisted)
    Software:     700 CNY (non-AI) → $70 at the pinned month rate
    AI:           = the Bedrock $40 (sourced via the AI service)
    Procurement:  $500 USD PO (procurement basis, not in consumption)

    Expected: cloud=100, software=70, ai=40, consumption total=210;
    procurement=500 separate.
    """
    # AI service whitelist (migration 149 seeds this in real envs).
    db_session.add(PlatformSetting(
        key="ai_spending.service_whitelist",
        value={"aws": ["Bedrock"]},
    ))
    # Live CNY/USD rate (some AI loaders consult it). Fresh updated_at via
    # server_default so the staleness guard passes.
    db_session.add(PlatformSetting(
        key="ai_spending.cny_to_usd_rate", value=0.14,
    ))

    # An owner person with a cost centre — exercises the derived
    # cost-centre axis on software / procurement rows.
    owner = IdentityPerson(
        id=uuid.uuid4(), hris_person_id="P0001",
        work_email="owner@example.com", display_name="Owner One",
        employment_status="active", cost_center="CC-ENG",
    )
    db_session.add(owner)
    await db_session.flush()
    owner_id = owner.id

    integ = Integration(
        id=uuid.uuid4(), kind="cloud", provider="aws",
        name="AWS Prod", slug="aws-prod",
    )
    db_session.add(integ)
    await db_session.flush()
    int_id = str(integ.id)

    def fact(service, usd, *, dept, key):
        return CmdbCostFact(
            fact_key=key, date=date(2026, 5, 15), integration_id=int_id,
            service=service, amount_usd=Decimal(usd), amount_native=Decimal(usd),
            currency="USD", cost_owner_department=dept,
            auto_owner_source="scope_owner" if dept else "none",
        )

    db_session.add(fact("Amazon EC2", "100", dept="Engineering", key="f-ec2"))
    db_session.add(fact("Amazon Bedrock", "40", dept="Engineering", key="f-bedrock"))

    # Software: one non-AI catalog + bill, one AI catalog + bill (the AI
    # bill must NOT leak into the software cost type).
    cat_non_ai = AssetSoftwareCatalog(id=uuid.uuid4(), name="Figma", category="design")
    cat_ai = AssetSoftwareCatalog(id=uuid.uuid4(), name="Copilot", category="ai")
    db_session.add_all([cat_non_ai, cat_ai])
    await db_session.flush()

    def bill(cat, amount, name):
        return SoftwareMonthlyBill(
            id=uuid.uuid4(), period=PERIOD, allocation_id=uuid.uuid4(),
            software_id=cat.id, plan_id=uuid.uuid4(), user_id=owner_id,
            software_name=name, plan_name=f"{name} Std", department="Design",
            price=Decimal(amount), billing_cycle="monthly", amount=Decimal(amount),
            currency="CNY", allocated_at=datetime(2026, 5, 1, tzinfo=UTC),
        )

    db_session.add(bill(cat_non_ai, "700", "Figma"))
    # AI-category bill — excluded from software (would be AI seats, but we
    # seed no seat allocation so it does not appear under AI either; its
    # only job here is to prove the software collector drops category=ai).
    db_session.add(bill(cat_ai, "999", "Copilot"))

    # Procurement: one completed USD PO recognised in the period.
    supplier = ProcurementSupplier(
        id=uuid.uuid4(), supplier_code="SUP1", name="Dell", status="active",
    )
    db_session.add(supplier)
    await db_session.flush()
    db_session.add(ProcurementOrder(
        id=uuid.uuid4(), order_number="PO-001", order_type="standard",
        status="completed", applicant_id=owner_id, business_owner_id=owner_id,
        supplier_id=supplier.id,
        currency="USD", total_amount=Decimal("500"), department="IT",
        cost_center="CC-IT", completed_at=datetime(2026, 5, 20, tzinfo=UTC),
    ))

    # Pin the month FX rate for CNY (frozen).
    from app.domains.finance.fx import upsert_monthly_rate
    await upsert_monthly_rate(
        db_session, "CNY", PERIOD, CNY_RATE, source="manual", frozen=True,
    )
    await db_session.flush()
    return db_session


async def _type_total_usd(db, cost_type) -> Decimal:
    return Decimal(str((await db.execute(
        select(func.coalesce(func.sum(FinanceSpendFact.amount_usd), 0)).where(
            FinanceSpendFact.period == PERIOD,
            FinanceSpendFact.cost_type == cost_type,
        )
    )).scalar() or 0))


@pytest.mark.asyncio
async def test_rollup_totals_and_dedup(seeded):
    db = seeded
    await coord.materialize_period(db, PERIOD)
    await db.flush()

    assert await _type_total_usd(db, "cloud") == Decimal("100.000000")
    assert await _type_total_usd(db, "software") == Decimal("70.000000")
    assert await _type_total_usd(db, "ai") == Decimal("40.000000")
    assert await _type_total_usd(db, "procurement") == Decimal("500.000000")

    # Dedup: no cloud row carries the AI-whitelisted Bedrock service.
    cloud_subs = (await db.execute(
        select(FinanceSpendFact.sub_category).where(
            FinanceSpendFact.period == PERIOD,
            FinanceSpendFact.cost_type == "cloud",
        )
    )).scalars().all()
    assert not any("Bedrock" in (s or "") for s in cloud_subs)

    # Dedup: no software row for the AI-category Copilot bill.
    sw_vendors = (await db.execute(
        select(FinanceSpendFact.vendor).where(
            FinanceSpendFact.period == PERIOD,
            FinanceSpendFact.cost_type == "software",
        )
    )).scalars().all()
    assert "Copilot" not in sw_vendors
    assert "Figma" in sw_vendors


@pytest.mark.asyncio
async def test_basis_separation(seeded):
    db = seeded
    await coord.materialize_period(db, PERIOD)
    await db.flush()
    consumption = Decimal(str((await db.execute(
        select(func.coalesce(func.sum(FinanceSpendFact.amount_usd), 0)).where(
            FinanceSpendFact.period == PERIOD,
            FinanceSpendFact.basis == "consumption",
        )
    )).scalar()))
    procurement = Decimal(str((await db.execute(
        select(func.coalesce(func.sum(FinanceSpendFact.amount_usd), 0)).where(
            FinanceSpendFact.period == PERIOD,
            FinanceSpendFact.basis == "procurement",
        )
    )).scalar()))
    assert consumption == Decimal("210.000000")  # 100 + 70 + 40
    assert procurement == Decimal("500.000000")


@pytest.mark.asyncio
async def test_fx_normalisation_uses_month_rate(seeded):
    db = seeded
    await coord.materialize_period(db, PERIOD)
    await db.flush()
    row = (await db.execute(
        select(FinanceSpendFact).where(
            FinanceSpendFact.cost_type == "software",
            FinanceSpendFact.vendor == "Figma",
        )
    )).scalar_one()
    assert row.currency_native == "CNY"
    assert row.amount_native == Decimal("700.000000")
    assert row.amount_usd == Decimal("70.000000")  # 700 * 0.10
    assert row.fx_rate_used == CNY_RATE
    assert row.fx_basis == "month_frozen"


@pytest.mark.asyncio
async def test_reconcile_internal_all_match(seeded):
    db = seeded
    await coord.materialize_period(db, PERIOD)
    await db.flush()
    result = await coord.reconcile_internal(db, PERIOD)
    assert result["all_match"] is True, result
    assert result["cloud"]["match"] is True
    assert result["ai"]["match"] is True
    assert result["software"]["match"] is True
    assert result["procurement"]["match"] is True


@pytest.mark.asyncio
async def test_idempotent_rollup(seeded):
    db = seeded
    await coord.materialize_period(db, PERIOD)
    await db.flush()
    n1 = (await db.execute(
        select(func.count()).select_from(FinanceSpendFact).where(
            FinanceSpendFact.period == PERIOD
        )
    )).scalar()
    await coord.materialize_period(db, PERIOD)
    await db.flush()
    n2 = (await db.execute(
        select(func.count()).select_from(FinanceSpendFact).where(
            FinanceSpendFact.period == PERIOD
        )
    )).scalar()
    assert n1 == n2
    # Total unchanged after re-run.
    assert await _type_total_usd(db, "cloud") == Decimal("100.000000")


@pytest.mark.asyncio
async def test_close_freezes_provisional_and_snapshots_total(seeded):
    db = seeded
    await coord.materialize_period(db, PERIOD)
    await db.flush()
    # Open period → provisional.
    provs = (await db.execute(
        select(FinanceSpendFact.is_provisional).where(FinanceSpendFact.period == PERIOD)
    )).scalars().all()
    assert all(p is True for p in provs)

    await finance_service.close_period(db, PERIOD, actor_id=None, note="m-end")
    await db.flush()
    provs = (await db.execute(
        select(FinanceSpendFact.is_provisional).where(FinanceSpendFact.period == PERIOD)
    )).scalars().all()
    assert all(p is False for p in provs)
    close = (await db.execute(
        select(FinancePeriodClose).where(FinancePeriodClose.period == PERIOD)
    )).scalar_one()
    assert close.status == "closed"
    assert close.closed_total_usd == Decimal("210.000000")


@pytest.mark.asyncio
async def test_close_blocked_when_foreign_currency_unresolvable(seeded, monkeypatch):
    """Regression: closing must REFUSE when a non-USD/CNY currency carries
    native spend but has no usable FX rate (FX source unreachable, no frozen
    /platform rate). Otherwise the unconverted 1.0 sentinel freezes a wildly
    overstated USD figure (e.g. ¥150,000 booked as $150,000) into
    ``closed_total_usd`` as the authoritative restatement baseline.
    """
    db = seeded
    owner_id = (await db.execute(select(IdentityPerson.id))).scalars().first()
    cat_id = (await db.execute(select(AssetSoftwareCatalog.id))).scalars().first()
    db.add(SoftwareMonthlyBill(
        id=uuid.uuid4(), period=PERIOD, allocation_id=uuid.uuid4(),
        software_id=cat_id, plan_id=uuid.uuid4(), user_id=owner_id,
        software_name="Foreign SaaS", plan_name="Std", department="IT",
        price=Decimal("150000"), billing_cycle="monthly",
        amount=Decimal("150000"), currency="JPY",
        allocated_at=datetime(2026, 5, 1, tzinfo=UTC),
    ))
    await db.flush()

    # FX source unreachable for the foreign currency → no real rate resolvable.
    from app.domains.finance import fx as finance_fx

    async def _no_rate(currency, period):  # noqa: ANN001
        return None

    monkeypatch.setattr(finance_fx, "fetch_rate_from_frankfurter", _no_rate)

    from app.core.exceptions import AppException
    with pytest.raises(AppException) as ei:
        await finance_service.close_period(db, PERIOD, actor_id=None)
    assert ei.value.code == "PERIOD_FX_UNRESOLVED"
    assert "JPY" in ei.value.message

    # The period must NOT be left closed.
    close = (await db.execute(
        select(FinancePeriodClose).where(FinancePeriodClose.period == PERIOD)
    )).scalar_one_or_none()
    assert close is None or close.status != "closed"


@pytest.mark.asyncio
async def test_reopen_makes_provisional_again(seeded):
    db = seeded
    await coord.materialize_period(db, PERIOD)
    await db.flush()
    await finance_service.close_period(db, PERIOD, actor_id=None)
    await db.flush()
    await finance_service.reopen_period(db, PERIOD, actor_id=None)
    await db.flush()
    provs = (await db.execute(
        select(FinanceSpendFact.is_provisional).where(FinanceSpendFact.period == PERIOD)
    )).scalars().all()
    assert all(p is True for p in provs)


@pytest.mark.asyncio
async def test_closed_period_drift_logs_restatement_without_moving(seeded):
    db = seeded
    await coord.materialize_period(db, PERIOD)
    await db.flush()
    await finance_service.close_period(db, PERIOD, actor_id=None)
    await db.flush()
    cloud_before = await _type_total_usd(db, "cloud")

    # A late-arriving cloud fact lands after close.
    db.add(CmdbCostFact(
        fact_key="f-late", date=date(2026, 5, 28), integration_id=(
            str((await db.execute(select(Integration.id))).scalars().first())
        ),
        service="Amazon S3", amount_usd=Decimal("25"), amount_native=Decimal("25"),
        currency="USD", cost_owner_department="Engineering",
        auto_owner_source="scope_owner",
    ))
    await db.flush()

    result = await coord.materialize_period(db, PERIOD)  # closed, not forced
    await db.flush()
    assert result["replaced"] is False
    assert result["restatement"] is not None
    # Published facts unchanged.
    assert await _type_total_usd(db, "cloud") == cloud_before
    # Restatement row written.
    restate = (await db.execute(
        select(FinanceRestatement).where(FinanceRestatement.period == PERIOD)
    )).scalars().all()
    assert len(restate) == 1
    assert restate[0].delta_usd == Decimal("25.000000")


@pytest.mark.asyncio
async def test_closed_period_standing_drift_does_not_duplicate_restatement(seeded):
    """The rollup recomputes closed periods daily. A standing drift (a late
    bill that permanently shifts the total) must log ONE restatement, not a
    fresh identical row on every recompute — the table has no unique key.
    """
    db = seeded
    await coord.materialize_period(db, PERIOD)
    await db.flush()
    await finance_service.close_period(db, PERIOD, actor_id=None)
    await db.flush()

    db.add(CmdbCostFact(
        fact_key="f-late-standing", date=date(2026, 5, 28), integration_id=(
            str((await db.execute(select(Integration.id))).scalars().first())
        ),
        service="Amazon S3", amount_usd=Decimal("25"), amount_native=Decimal("25"),
        currency="USD", cost_owner_department="Engineering",
        auto_owner_source="scope_owner",
    ))
    await db.flush()

    # Three recomputes of the SAME standing drift (simulating three daily
    # cron ticks). Only the first should log a restatement.
    for _ in range(3):
        result = await coord.materialize_period(db, PERIOD)
        await db.flush()
        assert result["restatement"] is not None  # drift still surfaced

    restate = (await db.execute(
        select(FinanceRestatement).where(FinanceRestatement.period == PERIOD)
    )).scalars().all()
    assert len(restate) == 1, (
        f"expected a single restatement for a standing drift, got {len(restate)}"
    )


# ── #1682: concurrent FIRST close must not 500 ──


@pytest.mark.asyncio
async def test_close_period_concurrent_first_close_returns_409_not_500(
    db_session, monkeypatch,
):
    """Two near-simultaneous FIRST closes of a period: both see
    ``get_close()==None`` and both INSERT ``finance_period_close``; the second
    trips ``uq_finance_period_close_period``. The SAVEPOINT must confine the
    IntegrityError and re-select the surviving winner → a clean 409, not a
    poisoned session / 500. Bare period (no seed) — the race is on the close
    row alone.
    """
    from app.core.exceptions import AppException

    # Isolate the test to the close-row INSERT race: the FX freeze/gate and
    # the downstream materialise are exercised elsewhere and here only add
    # network flakiness / seed dependencies unrelated to the race.
    async def _noop_freeze(db, p):
        return None
    async def _fake_materialize(db, p, *, allow_closed=False):
        return {"consumption_total_usd": Decimal("0")}
    monkeypatch.setattr(finance_service, "_freeze_period_fx", _noop_freeze)
    monkeypatch.setattr(coord, "materialize_period", _fake_materialize)

    period = "2026-09"
    # Winner closes the period first.
    await finance_service.close_period(db_session, period, actor_id=None)
    await db_session.flush()

    # Force the race for the loser: the first ``get_close`` returns None (as a
    # concurrent caller would see before the winner's row is visible), so the
    # INSERT path runs against the already-present unique row.
    real_get_close = finance_service.get_close
    calls = {"n": 0}

    async def _get_close_race(db, p):
        calls["n"] += 1
        if calls["n"] == 1 and p == period:
            return None
        return await real_get_close(db, p)

    monkeypatch.setattr(finance_service, "get_close", _get_close_race)

    with pytest.raises(AppException) as ei:
        await finance_service.close_period(db_session, period, actor_id=None)
    assert ei.value.status_code == 409
    assert ei.value.code == "PERIOD_ALREADY_CLOSED"
    # Session still usable — a follow-up commit succeeds (not poisoned).
    await db_session.commit()


@pytest.mark.asyncio
async def test_close_period_recovers_cross_transaction_winner(
    db_session, db_engine, monkeypatch,
):
    """The IntegrityError recovery must use a LOCKING current-read, not a
    snapshot read. Under REPEATABLE READ session-1's snapshot is fixed at the
    first ``get_close`` (None, before the winner commits); a plain re-select
    would still see None → re-raise → the very 500 the guard prevents. A
    ``FOR UPDATE`` current read sees the committed winner → clean 409.
    """
    from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker

    from app.core.exceptions import AppException

    # Isolate to the close-row INSERT race (see sibling test).
    async def _noop_freeze(db, p):
        return None
    async def _fake_materialize(db, p, *, allow_closed=False):
        return {"consumption_total_usd": Decimal("0")}
    monkeypatch.setattr(finance_service, "_freeze_period_fx", _noop_freeze)
    monkeypatch.setattr(coord, "materialize_period", _fake_materialize)

    period = "2026-11"
    # 1. Fix session-1's snapshot NOW (no close row yet).
    assert await finance_service.get_close(db_session, period) is None

    # 2. A DIFFERENT transaction closes + commits the winner.
    other = async_sessionmaker(db_engine, class_=AsyncSession, expire_on_commit=False)
    async with other() as s2:
        await finance_service.close_period(s2, period, actor_id=None)
        await s2.commit()

    # 3. Session-1 close: snapshot get_close still None → INSERT collides with
    #    the committed row → locking re-select finds the winner → 409.
    with pytest.raises(AppException) as ei:
        await finance_service.close_period(db_session, period, actor_id=None)
    assert ei.value.status_code == 409
    # Session is not poisoned — a follow-up commit succeeds.
    await db_session.commit()