"""双口径摄取持久化契约（成本口径改造 PR2）。

Pins the write path end-to-end: driver-emitted ``CostFact`` dual
measures (list / public / bucket) survive ``ingest_cost_facts`` —
including the pre-aggregation collapse and the ON DUPLICATE KEY
re-ingest — and land on the ``cmdb_cost_facts`` columns with the
None≠0 contract intact.
"""
from __future__ import annotations

import uuid
from datetime import date
from decimal import Decimal

import pytest
import pytest_asyncio
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.domains.cmdb.cost_service import ingest_cost_facts
from app.domains.cmdb.models import CmdbCostFact
from app.integrations.driver_base import CloudScopeRef, CostFact
from app.integrations.models import Integration


@pytest_asyncio.fixture
async def ingest_integration(db_session: AsyncSession) -> Integration:
    integ = Integration(
        id=str(uuid.uuid4()),
        kind="cloud",
        provider="aws",
        name="AWS dual-measure ingest test",
        slug=f"aws-ingest-{uuid.uuid4().hex[:8]}",
        enabled=True,
    )
    db_session.add(integ)
    await db_session.commit()
    # No refresh — keep ``.id`` the assigned str (Uuid column refresh
    # would swap it for a uuid.UUID whose bind serialization mismatches
    # the String(36) fact columns).
    return integ


def _facts(integration_id: str) -> list[CostFact]:
    scope = CloudScopeRef(scope_type="aws_account", external_id="111122223333")
    day = date(2026, 6, 10)
    return [
        # spot usage: list at executed price, public at OD.
        CostFact(
            date=day, integration_id=integration_id,
            service="Amazon Elastic Compute Cloud",
            amount_usd=3.0, amount_native=3.0, currency="USD",
            amount_usd_list=3.0, amount_native_list=3.0,
            amount_usd_public=10.0, cost_bucket="usage",
            scope=scope, resource_id="i-spot1", usage_type="SpotUsage",
        ),
        # SP negation: whitelist-external — list None, bucket classified.
        CostFact(
            date=day, integration_id=integration_id,
            service="Savings Plans",
            amount_usd=-7.0, amount_native=-7.0, currency="USD",
            cost_bucket="commitment_discount",
            scope=scope,
            resource_id="aws-account:111122223333:ComputeSavingsPlans",
            usage_type=None,
        ),
    ]


@pytest.mark.asyncio
async def test_ingest_persists_dual_measures_and_reingest_converges(
    db_session: AsyncSession,
    ingest_integration: Integration,
) -> None:
    result = await ingest_cost_facts(
        db_session,
        integration_id=str(ingest_integration.id),
        facts=_facts(str(ingest_integration.id)),
    )
    assert result["upserted"] == 2

    rows = (await db_session.execute(
        select(CmdbCostFact)
        .where(CmdbCostFact.integration_id == str(ingest_integration.id)),
    )).scalars().all()
    by_service = {r.service: r for r in rows}
    ec2 = by_service["Amazon Elastic Compute Cloud"]
    assert ec2.amount_usd_list == Decimal("3.000000")
    assert ec2.amount_native_list == Decimal("3.000000")
    assert ec2.amount_usd_public == Decimal("10.000000")
    assert ec2.cost_bucket == "usage"
    sp = by_service["Savings Plans"]
    assert sp.amount_usd_list is None
    assert sp.amount_usd_public is None
    assert sp.cost_bucket == "commitment_discount"

    # Re-ingest with changed measures — ON DUPLICATE KEY must update
    # the dual-measure columns (not silently keep stale values).
    updated = _facts(str(ingest_integration.id))
    updated[0].amount_usd_list = 4.5
    updated[0].amount_native_list = 4.5
    updated[0].amount_usd_public = 11.0
    result2 = await ingest_cost_facts(
        db_session,
        integration_id=str(ingest_integration.id),
        facts=updated,
    )
    assert result2["upserted"] == 2
    await db_session.commit()
    # populate_existing：ON DUPLICATE KEY 的原生 UPDATE 不经过 ORM
    # identity map，重查须强制以 DB 行覆盖已加载实例。
    rows2 = (await db_session.execute(
        select(CmdbCostFact)
        .where(CmdbCostFact.integration_id == str(ingest_integration.id))
        .execution_options(populate_existing=True),
    )).scalars().all()
    assert len(rows2) == 2  # converged, no duplicates
    ec2_2 = {r.service: r for r in rows2}["Amazon Elastic Compute Cloud"]
    assert ec2_2.amount_usd_list == Decimal("4.500000")
    assert ec2_2.amount_usd_public == Decimal("11.000000")


@pytest.mark.asyncio
async def test_preaggregation_sums_list_measures_none_aware(
    db_session: AsyncSession,
    ingest_integration: Integration,
) -> None:
    """Two facts colliding on one natural key: list/public sum across
    members that expose a value; a colliding pair where NO member has
    list stays NULL (never a fabricated 0)."""
    scope = CloudScopeRef(scope_type="aws_account", external_id="111122223333")
    day = date(2026, 6, 11)
    colliding = [
        CostFact(
            date=day, integration_id=str(ingest_integration.id),
            service="Amazon S3", amount_usd=1.0, amount_native=1.0,
            currency="USD", amount_usd_list=1.0, amount_native_list=1.0,
            amount_usd_public=1.2, cost_bucket="usage",
            scope=scope, resource_id="bucket-a", usage_type="Requests",
        ),
        CostFact(
            date=day, integration_id=str(ingest_integration.id),
            service="Amazon S3", amount_usd=2.0, amount_native=2.0,
            currency="USD", amount_usd_list=2.0, amount_native_list=2.0,
            amount_usd_public=2.2, cost_bucket="usage",
            scope=scope, resource_id="bucket-a", usage_type="Requests",
        ),
        # Second natural key with NO list measures at all.
        CostFact(
            date=day, integration_id=str(ingest_integration.id),
            service="Tax", amount_usd=0.5, amount_native=0.5,
            currency="USD", cost_bucket="tax",
            scope=scope, resource_id=None, usage_type="Tax",
        ),
    ]
    await ingest_cost_facts(
        db_session,
        integration_id=str(ingest_integration.id),
        facts=colliding,
    )
    rows = (await db_session.execute(
        select(CmdbCostFact)
        .where(CmdbCostFact.integration_id == str(ingest_integration.id))
        .where(CmdbCostFact.date == day),
    )).scalars().all()
    by_service = {r.service: r for r in rows}
    s3 = by_service["Amazon S3"]
    assert s3.amount_usd == Decimal("3.000000")
    assert s3.amount_usd_list == Decimal("3.000000")
    assert s3.amount_usd_public == Decimal("3.400000")
    tax = by_service["Tax"]
    assert tax.amount_usd_list is None
    assert tax.amount_usd_public is None
    assert tax.cost_bucket == "tax"


@pytest.mark.asyncio
async def test_cross_bucket_fact_key_collision_persists_two_facts(
    db_session: AsyncSession,
    ingest_integration: Integration,
) -> None:
    """同 (date, scope, instance, item) 但不同 ``cost_bucket`` 的行
    **绝不合并**(阿里云同日 Refund/Adjustment 与 usage 共享自然键的
    形状)——合并会让净额挂在 |金额| 最大成员的桶下,污染财务瀑布与
    list 白名单。usage 桶保留历史(裸)fact_key,其余桶用
    bucket_disambiguator 落成独立行;重灌收敛、不重复。"""
    from app.domains.cmdb.cost_service import compute_fact_key

    integ_id = str(ingest_integration.id)
    scope = CloudScopeRef(scope_type="aws_account", external_id="999900001111")
    day = date(2026, 6, 12)

    def _pair() -> list[CostFact]:
        return [
            CostFact(
                date=day, integration_id=integ_id,
                service="云服务器 ECS",
                amount_usd=100.0, amount_native=100.0, currency="USD",
                amount_usd_list=120.0, amount_native_list=120.0,
                cost_bucket="usage",
                scope=scope, resource_id="i-shared", usage_type="cpu",
            ),
            CostFact(
                date=day, integration_id=integ_id,
                service="云服务器 ECS",
                amount_usd=-30.0, amount_native=-30.0, currency="USD",
                cost_bucket="adjustment",
                scope=scope, resource_id="i-shared", usage_type="cpu",
            ),
        ]

    result = await ingest_cost_facts(
        db_session, integration_id=integ_id, facts=_pair(),
    )
    assert result["upserted"] == 2

    rows = (await db_session.execute(
        select(CmdbCostFact).where(
            CmdbCostFact.integration_id == integ_id,
            CmdbCostFact.resource_id == "i-shared",
        ),
    )).scalars().all()
    assert len(rows) == 2
    by_bucket = {r.cost_bucket: r for r in rows}
    assert set(by_bucket) == {"usage", "adjustment"}
    assert float(by_bucket["usage"].amount_usd) == pytest.approx(100.0)
    assert float(by_bucket["usage"].amount_usd_list) == pytest.approx(120.0)
    assert float(by_bucket["adjustment"].amount_usd) == pytest.approx(-30.0)
    assert by_bucket["adjustment"].amount_usd_list is None

    # usage 成员保留历史裸键(重灌已存量窗口时覆盖 legacy 合并行,
    # 而非另立新行);adjustment 成员的键被 bucket 消歧。
    plain_key = compute_fact_key(
        integration_id=integ_id, fact_date=day,
        scope_external_id="999900001111",
        resource_id="i-shared", usage_type="cpu",
    )
    assert by_bucket["usage"].fact_key == plain_key
    assert by_bucket["adjustment"].fact_key != plain_key

    # 幂等:同窗口重灌收敛到同两行,金额不漂移、不重复。
    result2 = await ingest_cost_facts(
        db_session, integration_id=integ_id, facts=_pair(),
    )
    assert result2["upserted"] == 2
    await db_session.commit()
    rows2 = (await db_session.execute(
        select(CmdbCostFact)
        .where(
            CmdbCostFact.integration_id == integ_id,
            CmdbCostFact.resource_id == "i-shared",
        )
        .execution_options(populate_existing=True),
    )).scalars().all()
    assert len(rows2) == 2
    assert sorted(float(r.amount_usd) for r in rows2) == [
        pytest.approx(-30.0), pytest.approx(100.0),
    ]


@pytest.mark.asyncio
async def test_bucket_key_converges_across_batch_composition_changes(
    db_session: AsyncSession,
    ingest_integration: Integration,
) -> None:
    """桶→fact_key 只取决于桶本身，与本批次里出现了哪些桶无关。

    上游批次组成会缩小：阿里两条取数路径都对 ``native == 0`` 的行
    ``continue``，所以某天 usage 行净额归零就整行消失，该自然键的桶集合
    从 {usage, adjustment} 退回 {adjustment}。按批次选「保留裸键的桶」时，
    退单行会从消歧键搬回裸键、另写一行并孤立自己上一行 —— 谁都不扫，
    −30 的退款永久记成 −60（进的是财务瀑布与全部分摊的底表）。
    """
    from app.domains.cmdb.cost_service import compute_fact_key

    integ_id = str(ingest_integration.id)
    scope = CloudScopeRef(scope_type="aliyun_account",
                          external_id="5000000000000001")
    day = date(2026, 6, 13)

    def _usage() -> CostFact:
        return CostFact(
            date=day, integration_id=integ_id, service="ECS",
            amount_usd=100.0, amount_native=100.0, currency="USD",
            amount_usd_list=120.0, amount_native_list=120.0,
            cost_bucket="usage", scope=scope,
            resource_id="i-shrink", usage_type="cpu",
        )

    def _adjustment() -> CostFact:
        return CostFact(
            date=day, integration_id=integ_id, service="ECS",
            amount_usd=-30.0, amount_native=-30.0, currency="USD",
            cost_bucket="adjustment", scope=scope,
            resource_id="i-shrink", usage_type="cpu",
        )

    async def _rows() -> list[CmdbCostFact]:
        await db_session.commit()
        return list((await db_session.execute(
            select(CmdbCostFact)
            .where(
                CmdbCostFact.integration_id == integ_id,
                CmdbCostFact.resource_id == "i-shrink",
            )
            .execution_options(populate_existing=True),
        )).scalars().all())

    # 同步 1：只有退单行落地（usage 当日净额为 0，上游整行不返回）。
    await ingest_cost_facts(
        db_session, integration_id=integ_id, facts=[_adjustment()],
    )
    rows1 = await _rows()
    assert [r.cost_bucket for r in rows1] == ["adjustment"]
    adj_key = rows1[0].fact_key

    # 同步 2：usage 行出现 —— 桶集合增长。退单行的键不得改变。
    await ingest_cost_facts(
        db_session, integration_id=integ_id,
        facts=[_usage(), _adjustment()],
    )
    rows2 = await _rows()
    assert len(rows2) == 2, [r.cost_bucket for r in rows2]
    by_bucket2 = {r.cost_bucket: r for r in rows2}
    assert by_bucket2["adjustment"].fact_key == adj_key

    # 同步 3：usage 行再次消失 —— 桶集合缩小回 {adjustment}。退单行仍
    # 写同一行；不得凭空多出第二条退单。
    await ingest_cost_facts(
        db_session, integration_id=integ_id, facts=[_adjustment()],
    )
    rows3 = await _rows()
    adj_rows = [r for r in rows3 if r.cost_bucket == "adjustment"]
    assert len(adj_rows) == 1, [(r.cost_bucket, r.fact_key) for r in rows3]
    assert adj_rows[0].fact_key == adj_key
    assert sum(
        float(r.amount_usd) for r in rows3 if r.cost_bucket == "adjustment"
    ) == pytest.approx(-30.0)

    # 键的形状：usage 裸键、非 usage 恒带桶分量（与批次无关）。
    plain = compute_fact_key(
        integration_id=integ_id, fact_date=day,
        scope_external_id="5000000000000001",
        resource_id="i-shrink", usage_type="cpu",
    )
    assert adj_key == compute_fact_key(
        integration_id=integ_id, fact_date=day,
        scope_external_id="5000000000000001",
        resource_id="i-shrink", usage_type="cpu",
        bucket_disambiguator="adjustment",
    )
    assert adj_key != plain
    assert by_bucket2["usage"].fact_key == plain


def test_bucket_disambiguator_is_pure_function_of_bucket() -> None:
    """消歧分量的取值只看桶：usage → None（裸键），其余桶 → 桶名。
    非 usage 单桶批次也必须带桶分量 —— 否则「本批只有这一个桶」的行
    与 usage 行争同一个唯一键。"""
    from app.domains.cmdb.cost_service import (
        _bucket_disambiguator,
        _preaggregate_facts_by_fact_key,
    )

    assert _bucket_disambiguator("usage") is None
    for bucket in ("adjustment", "credit", "tax", "support", "other"):
        assert _bucket_disambiguator(bucket) == bucket
    # None / "" 都不得折叠到 usage 的裸键，也不得互相折叠。
    assert _bucket_disambiguator(None) == "\x00"
    assert _bucket_disambiguator("") == ""

    scope = CloudScopeRef(scope_type="aliyun_account", external_id="500")
    day = date(2026, 6, 14)

    def _fact(bucket: str) -> CostFact:
        return CostFact(
            date=day, integration_id="i", service="ECS",
            amount_usd=1.0, amount_native=1.0, currency="USD",
            cost_bucket=bucket, scope=scope,
            resource_id="r", usage_type="u",
        )

    for batch in (
        [_fact("credit")],
        [_fact("credit"), _fact("usage")],
        [_fact("credit"), _fact("adjustment")],
        [_fact("credit"), _fact("adjustment"), _fact("usage")],
    ):
        collapsed, _ = _preaggregate_facts_by_fact_key(batch)
        got = {
            f.cost_bucket: disambiguator for f, disambiguator in collapsed
        }
        assert got["credit"] == "credit", batch
        if "usage" in got:
            assert got["usage"] is None