"""Cloud spend for the External API — query layer.

Serves the per-day and per-month cloud cost of the subscriptions an API
client is allowed to see. Lives in the cmdb domain because the data
(``cmdb_cost_facts`` / ``cmdb_cloud_billing_daily``) and every measure
rule are cmdb's; the router above it only adds auth + shaping.

Five contracts this module exists to keep, each one a scar:

**1. One basis, always ``list``.** Amounts are the business pre-discount
(研发口径) figure via :func:`app.domains.cmdb.cost_view.fact_amount_expr`.
There is deliberately no ``measure`` parameter: the invoiced net and the
pre-discount list figure for the same window divide out to the company's
contract discount rate, so the two must never be reachable by one
consumer. An invoice-basis surface is a separate scope and a separate
grant decision, not a flag here.

**2. Facts first, daily aggregate as the fallback.** ``cmdb_cost_facts``
is the line-item truth and the only source that can carry the list
basis per service. Its ingest cron and the per-day aggregate cron run
independently and can disagree for hours, so a scope-month with zero
facts falls back to ``cmdb_cloud_billing_daily`` (the same rule the
in-app scope billing page uses) and says so in ``source``. Without it
this API would report ``$0`` for a scope whose spend alert is firing.

**3. A missing FX rate drops the row, never zeroes it.** CN clouds write
native CNY into ``cmdb_cloud_billing_daily.total_cost_usd``. A row we
cannot convert is omitted and disclosed in
``meta.unconverted_by_currency`` (the dropped native amount per
currency) — emitting ``0`` would tell a consumer
"this day cost nothing", which is a lie an integrator cannot detect.

**4. Recent days are provisional and say so.** The facts cron re-ingests
a rolling 3-day window (late-arriving usage, retroactive re-rating), and
the current month is by definition partial. Every row carries
``is_provisional`` and every response carries ``data_through`` /
``provisional_from``, so a consumer's incremental load knows what to
overwrite instead of trusting a half-ingested day forever.

**5. Per-account sums are not the invoice.** Charges that land on no
sub-account (support, tax, credits, adjustments — ``scope_id IS NULL``)
are structurally absent from a per-subscription API, and line-item vs
daily-aggregate carry a ~1% caliber gap. ``meta.excludes_unallocatable``
states it in-band so nobody reconciles this against a provider invoice
and concludes Helix is wrong.

Scan discipline: every aggregate is chunked per calendar month
(:func:`app.domains.cmdb.cost_service._month_chunks`) — a single wide
``GROUP BY`` over a multi-month window blew TiDB's per-statement memory
quota in production (error 8175, 2026-07-28).
"""
from __future__ import annotations

import uuid as uuid_mod
from collections import defaultdict
from dataclasses import dataclass, field, replace
from datetime import UTC, date, datetime, timedelta
from decimal import Decimal
from typing import Any

from sqlalchemy import and_, func, or_, select
from sqlalchemy.ext.asyncio import AsyncSession

from app.core.exceptions import ValidationError
from app.domains.cmdb.cost_service import _month_chunks
from app.domains.cmdb.cost_view import (
    COST_MEASURE_LIST,
    daily_amount_expr,
    fact_amount_expr,
)
from app.domains.cmdb.models import (
    CmdbCloudBillingDaily,
    CmdbCloudScope,
    CmdbCostFact,
)
from app.domains.cmdb.open_visibility import ResolvedCloudVisibility
from app.identity.person_models import IdentityPerson
from app.integrations.models import Integration

#: The facts cron re-ingests a rolling 3-day window ending yesterday
#: (``sync_cost_facts_for_integration``), and today is not ingested at
#: all yet. Anything on or after ``today - 3d`` can still change.
PROVISIONAL_DAYS = 3

#: Upper bound on materialized (day × subscription × service) cells for
#: one request. The merge happens in Python (facts + fallback + two
#: scope-routing branches), so the cap is what stands between a
#: "give me everything" call and an OOM. Over it → 422 asking the caller
#: to narrow, never a silently truncated page that reads as complete.
MAX_RESULT_CELLS = 200_000

SOURCE_FACTS = "cost_facts"
SOURCE_DAILY_AGGREGATE = "billing_daily"

#: Scope types whose cost lives in ``cmdb_cost_facts`` keyed by
#: ``(integration_id, resource_group)`` rather than ``scope_id`` — cloud
#: providers don't report per-day cost at resource-group granularity, so
#: ``cmdb_cloud_billing_daily`` is structurally empty for them (hence no
#: fallback either). Imported lazily from the lifecycle router, which
#: owns the canonical set.


def _rg_scope_types() -> frozenset[str]:
    from app.domains.cmdb.lifecycle_router import _RG_SCOPE_TYPES

    return frozenset(_RG_SCOPE_TYPES)


@dataclass(frozen=True)
class SubscriptionRef:
    """The subscription metadata every billing row is labelled with."""

    id: str
    external_id: str
    name: str | None
    provider: str | None
    environment: str | None
    integration_id: str
    scope_type: str
    owner_email: str | None

    @property
    def is_resource_group(self) -> bool:
        return self.scope_type in _rg_scope_types()

    def label(self) -> dict[str, Any]:
        return {
            "subscription_id": self.id,
            "external_id": self.external_id,
            "name": self.name,
            "provider": self.provider,
            "environment": self.environment,
        }


@dataclass
class BillingQuery:
    """Everything a billing read is scoped by, already validated."""

    start: date
    end: date
    subscriptions: tuple[str, ...] = ()
    providers: tuple[str, ...] = ()
    environments: tuple[str, ...] = ()
    owner_email: str | None = None
    include_services: bool = False


@dataclass
class BillingResult:
    rows: list[dict[str, Any]] = field(default_factory=list)
    data_through: date | None = None
    unconverted_by_currency: dict[str, float] = field(default_factory=dict)
    sources: set[str] = field(default_factory=set)


def provisional_from(today: date | None = None) -> date:
    """First date whose amount can still change on the next ingest."""
    return (today or datetime.now(UTC).date()) - timedelta(days=PROVISIONAL_DAYS)


async def resolve_subscriptions(
    db: AsyncSession,
    *,
    visibility: ResolvedCloudVisibility,
    query: BillingQuery,
) -> list[SubscriptionRef]:
    """The subscription set a request may aggregate over.

    Whitelist first, then the caller's filters — so a filter can only
    ever narrow what the client was already entitled to. Deleted
    subscriptions are excluded; ``closed`` (decommissioned) ones are NOT:
    a subscription shut down mid-window still spent money that month, and
    dropping it would silently shrink a historical total.
    """
    conds: list[Any] = [
        CmdbCloudScope.deleted_at.is_(None),
        *visibility.scope_conditions(),
    ]

    if query.environments:
        conds.append(CmdbCloudScope.environment.in_(query.environments))

    if query.providers:
        integ_ids = (await db.execute(
            select(Integration.id).where(Integration.provider.in_(query.providers)),
        )).scalars().all()
        if not integ_ids:
            return []
        conds.append(CmdbCloudScope.integration_id.in_([str(i) for i in integ_ids]))

    if query.owner_email:
        email = query.owner_email.strip().lower()
        person_ids = (await db.execute(
            select(IdentityPerson.id).where(
                func.lower(IdentityPerson.work_email) == email,
            ),
        )).scalars().all()
        if not person_ids:
            return []
        conds.append(
            CmdbCloudScope.owner_person_id.in_([str(i) for i in person_ids]),
        )

    if query.subscriptions:
        # Accept the Helix UUID or the cloud-native external id, matching
        # ``/open/cloud/subscriptions/{key}``. An id the client cannot
        # see simply matches nothing — same silence as a 404 there.
        uuids: list[uuid_mod.UUID] = []
        externals: list[str] = []
        for raw in query.subscriptions:
            value = raw.strip()
            if not value:
                continue
            try:
                uuids.append(uuid_mod.UUID(value))
            except (ValueError, TypeError):
                externals.append(value)
        terms: list[Any] = []
        if uuids:
            terms.append(CmdbCloudScope.id.in_(uuids))
        if externals:
            terms.append(CmdbCloudScope.external_id.in_(externals))
        if not terms:
            return []
        conds.append(or_(*terms) if len(terms) > 1 else terms[0])

    rows = (await db.execute(
        select(CmdbCloudScope).where(*conds).order_by(
            CmdbCloudScope.external_id.asc(), CmdbCloudScope.id.asc(),
        ),
    )).scalars().all()
    if not rows:
        return []

    integ_by_id: dict[str, Integration] = {}
    # Distinct name from the provider-filter lookup above: same concept,
    # different type (str ids off the scope rows vs UUIDs off the query),
    # and reusing one name made both of them ambiguous to the type checker.
    scope_integration_ids = {s.integration_id for s in rows if s.integration_id}
    if scope_integration_ids:
        integ_rows = (await db.execute(
            select(Integration).where(
                Integration.id.in_([
                    uuid_mod.UUID(i) for i in scope_integration_ids
                ]),
            ),
        )).scalars().all()
        integ_by_id = {str(r.id): r for r in integ_rows}

    owner_by_id: dict[str, str | None] = {}
    owner_ids = {s.owner_person_id for s in rows if s.owner_person_id}
    if owner_ids:
        owner_rows = (await db.execute(
            select(IdentityPerson.id, IdentityPerson.work_email).where(
                IdentityPerson.id.in_([uuid_mod.UUID(i) for i in owner_ids]),
            ),
        )).all()
        owner_by_id = {str(pid): email for pid, email in owner_rows}

    return [
        SubscriptionRef(
            id=str(s.id),
            external_id=s.external_id,
            name=s.display_name,
            provider=(
                integ_by_id[str(s.integration_id)].provider
                if str(s.integration_id) in integ_by_id else None
            ),
            environment=s.environment,
            integration_id=str(s.integration_id),
            scope_type=s.scope_type,
            owner_email=(
                owner_by_id.get(s.owner_person_id) if s.owner_person_id else None
            ),
        )
        for s in rows
    ]


# ── Aggregation ───────────────────────────────────────────────────────


async def _facts_by_day(
    db: AsyncSession,
    *,
    subs: list[SubscriptionRef],
    start: date,
    end: date,
    include_services: bool,
) -> dict[tuple[str, date, str | None], Decimal]:
    """``(subscription_id, date, service) → amount`` from cost_facts.

    Two queries because the fact table is keyed two different ways: a
    tenant-level subscription by ``scope_id``, a resource-group-typed one
    by ``(integration_id, resource_group)``. Merging them in SQL would
    mean a UNION whose GROUP BY keys differ per branch; merging in Python
    keeps each query on its own index.
    """
    amount = fact_amount_expr(COST_MEASURE_LIST)
    out: dict[tuple[str, date, str | None], Decimal] = defaultdict(Decimal)

    tenant_subs = [s for s in subs if not s.is_resource_group]
    rg_subs = [s for s in subs if s.is_resource_group]

    for chunk_start, chunk_end in _month_chunks(start, end):
        if tenant_subs:
            cols: list[Any] = [
                CmdbCostFact.scope_id,
                CmdbCostFact.date,
                func.coalesce(func.sum(amount), 0).label("amount"),
            ]
            group: list[Any] = [CmdbCostFact.scope_id, CmdbCostFact.date]
            if include_services:
                cols.insert(2, CmdbCostFact.service)
                group.append(CmdbCostFact.service)
            rows = (await db.execute(
                select(*cols).where(
                    CmdbCostFact.scope_id.in_([s.id for s in tenant_subs]),
                    CmdbCostFact.date >= chunk_start,
                    CmdbCostFact.date <= chunk_end,
                ).group_by(*group),
            )).all()
            for row in rows:
                service = row.service if include_services else None
                out[(str(row.scope_id), row.date, service)] += Decimal(
                    str(row.amount or 0),
                )

        if rg_subs:
            by_key = {(s.integration_id, s.external_id): s for s in rg_subs}
            cols = [
                CmdbCostFact.integration_id,
                CmdbCostFact.resource_group,
                CmdbCostFact.date,
                func.coalesce(func.sum(amount), 0).label("amount"),
            ]
            group = [
                CmdbCostFact.integration_id,
                CmdbCostFact.resource_group,
                CmdbCostFact.date,
            ]
            if include_services:
                cols.insert(3, CmdbCostFact.service)
                group.append(CmdbCostFact.service)
            rows = (await db.execute(
                select(*cols).where(
                    or_(*[
                        and_(
                            CmdbCostFact.integration_id == integ_id,
                            CmdbCostFact.resource_group == rg_name,
                        )
                        for integ_id, rg_name in by_key
                    ]),
                    CmdbCostFact.date >= chunk_start,
                    CmdbCostFact.date <= chunk_end,
                ).group_by(*group),
            )).all()
            for row in rows:
                sub = by_key.get((str(row.integration_id), row.resource_group))
                if sub is None:
                    continue
                service = row.service if include_services else None
                out[(sub.id, row.date, service)] += Decimal(str(row.amount or 0))

    return dict(out)


async def daily_aggregate_fallback(
    db: AsyncSession,
    *,
    subs: list[SubscriptionRef],
    start: date,
    end: date,
    covered: set[tuple[str, date]],
) -> tuple[dict[tuple[str, date, str | None], Decimal], dict[str, float]]:
    """Fill scope-months the line-item table has no rows for.

    Granularity of the decision is the **scope-month**, matching the
    in-app scope billing page: a per-day decision would interleave two
    calibers inside one month (the two tables differ by ~1%), and a
    per-window decision would let one stale month hide behind a fresh
    one. Resource-group subscriptions are skipped — the aggregate table
    cannot answer ``(integration, resource_group)``.

    Returns the extra cells plus, per currency, the native amount that
    was dropped for want of an FX rate (never converted to ``0``).
    """
    from app.core.fx_rate import CloudCostUsdConverter

    eligible = [s for s in subs if not s.is_resource_group]
    if not eligible:
        return {}, {}

    out: dict[tuple[str, date, str | None], Decimal] = {}
    converter = CloudCostUsdConverter(db)

    for chunk_start, chunk_end in _month_chunks(start, end):
        missing = [
            s for s in eligible
            if not any(
                (s.id, day) in covered
                for day in _days(chunk_start, chunk_end)
            )
        ]
        if not missing:
            continue
        rows = (await db.execute(
            select(
                CmdbCloudBillingDaily.scope_id,
                CmdbCloudBillingDaily.date,
                CmdbCloudBillingDaily.integration_id,
                CmdbCloudBillingDaily.currency,
                daily_amount_expr(COST_MEASURE_LIST).label("amount"),
            ).where(
                CmdbCloudBillingDaily.scope_id.in_([s.id for s in missing]),
                CmdbCloudBillingDaily.date >= chunk_start,
                CmdbCloudBillingDaily.date <= chunk_end,
            ),
        )).all()
        for row in rows:
            usd, converted = await converter.to_usd_checked(
                row.amount, row.currency, str(row.integration_id),
            )
            if not converted:
                # Disclosed in meta; emitting 0 would read as "no spend".
                continue
            key = (str(row.scope_id), row.date, None)
            out[key] = out.get(key, Decimal("0")) + Decimal(str(usd))

    return out, dict(converter.unconverted_by_currency or {})


def _days(start: date, end: date):
    cur = start
    while cur <= end:
        yield cur
        cur += timedelta(days=1)


async def daily_costs(
    db: AsyncSession,
    *,
    subs: list[SubscriptionRef],
    query: BillingQuery,
) -> BillingResult:
    """Per-day rows for ``subs``, facts-first with aggregate fallback."""
    result = BillingResult()
    if not subs:
        return result

    facts = await _facts_by_day(
        db, subs=subs, start=query.start, end=query.end,
        include_services=query.include_services,
    )
    covered = {(sub_id, day) for (sub_id, day, _svc) in facts}
    if facts:
        result.sources.add(SOURCE_FACTS)

    fallback, unconverted = await daily_aggregate_fallback(
        db, subs=subs, start=query.start, end=query.end, covered=covered,
    )
    if fallback:
        result.sources.add(SOURCE_DAILY_AGGREGATE)
    result.unconverted_by_currency = unconverted

    by_ref = {s.id: s for s in subs}
    cutoff = provisional_from()
    cells = len(facts) + len(fallback)
    if cells > MAX_RESULT_CELLS:
        raise ValidationError(
            f"Result too large: {cells} rows for the requested window and "
            f"filters (cap {MAX_RESULT_CELLS}). Narrow the date range, the "
            "subscription filter, or drop the service breakdown.",
        )

    rows: list[dict[str, Any]] = []
    for source, cells_map in (
        (SOURCE_FACTS, facts), (SOURCE_DAILY_AGGREGATE, fallback),
    ):
        for (sub_id, day, service), amount in cells_map.items():
            ref = by_ref.get(sub_id)
            if ref is None:
                continue
            row = {
                "date": day.isoformat(),
                **ref.label(),
                "amount_usd": float(amount),
                "currency": "USD",
                "is_provisional": day >= cutoff,
                "source": source,
            }
            if query.include_services:
                row["service"] = service or ""
            rows.append(row)
            if result.data_through is None or day > result.data_through:
                result.data_through = day

    rows.sort(key=lambda r: (
        r["date"], r["external_id"], r.get("service") or "",
    ))
    result.rows = rows
    return result


async def monthly_costs(
    db: AsyncSession,
    *,
    subs: list[SubscriptionRef],
    query: BillingQuery,
) -> BillingResult:
    """Per-month rows — the daily cells folded up, **one month at a time**.

    Deliberately derived from the same cells the daily endpoint returns
    rather than from a separate monthly aggregate: two independent
    queries over two tables is exactly how "the month total disagrees
    with the sum of its days" happens.

    Folded per calendar-month chunk rather than over the whole window,
    because the daily path materializes its cells and guards them with
    :data:`MAX_RESULT_CELLS`. A 36-month pull (this endpoint's own
    documented limit) across a few hundred subscriptions is ~200k daily
    cells — it tripped that guard and 422'd a request the API says it
    supports. Folding per chunk keeps live memory at one month's cells
    while the arithmetic stays identical.
    """
    today = datetime.now(UTC).date()
    folded: dict[tuple[str, str, str | None], dict[str, Any]] = {}
    merged = BillingResult()

    for chunk_start, chunk_end in _month_chunks(query.start, query.end):
        chunk_query = replace(query, start=chunk_start, end=chunk_end)
        daily = await daily_costs(db, subs=subs, query=chunk_query)
        merged.sources |= daily.sources
        for currency, amount in daily.unconverted_by_currency.items():
            merged.unconverted_by_currency[currency] = (
                merged.unconverted_by_currency.get(currency, 0.0) + amount
            )
        if daily.data_through and (
            merged.data_through is None or daily.data_through > merged.data_through
        ):
            merged.data_through = daily.data_through

        for row in daily.rows:
            day = date.fromisoformat(row["date"])
            month = f"{day.year:04d}-{day.month:02d}"
            key = (month, row["subscription_id"], row.get("service"))
            entry = folded.get(key)
            if entry is None:
                entry = {
                    "month": month,
                    **{k: row[k] for k in (
                        "subscription_id", "external_id", "name", "provider",
                        "environment",
                    )},
                    "amount_usd": 0.0,
                    "currency": "USD",
                    "month_is_partial": False,
                    "sources": set(),
                }
                if query.include_services:
                    entry["service"] = row.get("service") or ""
                folded[key] = entry
            entry["amount_usd"] += row["amount_usd"]
            entry["sources"].add(row["source"])
            # Partial when the month is still running OR the window itself
            # only covers part of it OR any day in it is still provisional —
            # a consumer must not treat any of the three as a closed month.
            if (
                row["is_provisional"]
                or (day.year, day.month) == (today.year, today.month)
            ):
                entry["month_is_partial"] = True

    month_first = date(query.start.year, query.start.month, 1)
    last_month_end = _month_end(query.end)
    for entry in folded.values():
        year, month_num = (int(p) for p in entry["month"].split("-"))
        first = date(year, month_num, 1)
        if first < month_first or _month_end(first) > last_month_end:
            entry["month_is_partial"] = True
        entry["amount_usd"] = round(entry["amount_usd"], 6)
        entry["source"] = (
            SOURCE_FACTS if entry["sources"] == {SOURCE_FACTS}
            else SOURCE_DAILY_AGGREGATE if entry["sources"] == {
                SOURCE_DAILY_AGGREGATE,
            }
            else "mixed"
        )
        entry.pop("sources")

    merged.rows = sorted(
        folded.values(),
        key=lambda r: (r["month"], r["external_id"], r.get("service") or ""),
    )
    return merged


def _month_end(day: date) -> date:
    if day.month == 12:
        return date(day.year, 12, 31)
    return date(day.year, day.month + 1, 1) - timedelta(days=1)


def response_meta(result: BillingResult, *, query: BillingQuery) -> dict[str, Any]:
    """The in-band honesty block every billing response carries."""
    meta: dict[str, Any] = {
        # Named explicitly so a consumer never has to infer which cost
        # basis it is holding.
        "measure": COST_MEASURE_LIST,
        "measure_note": (
            "Business pre-discount (list) cost. Not the invoiced net "
            "amount — do not reconcile against a provider invoice."
        ),
        "currency": "USD",
        "window": {"start": query.start.isoformat(), "end": query.end.isoformat()},
        "data_through": (
            result.data_through.isoformat() if result.data_through else None
        ),
        "provisional_from": provisional_from().isoformat(),
        # Charges that land on no sub-account (support / tax / credits /
        # adjustments) cannot appear in a per-subscription view.
        "excludes_unallocatable": True,
        "sources": sorted(result.sources),
    }
    if result.unconverted_by_currency:
        meta["unconverted_by_currency"] = result.unconverted_by_currency
    return meta


__all__ = [
    "MAX_RESULT_CELLS",
    "daily_aggregate_fallback",
    "PROVISIONAL_DAYS",
    "BillingQuery",
    "BillingResult",
    "SubscriptionRef",
    "daily_costs",
    "monthly_costs",
    "provisional_from",
    "resolve_subscriptions",
    "response_meta",
]