"""Window-total cloud spend grouped by one dimension, for the External API.

Companion to :mod:`app.domains.cmdb.open_billing_service` (which answers
"how much per day"); this answers "how much per *what*" over a window:
subscription / service / provider / environment / department / owner /
resource.

Everything the daily module contracts still holds — one ``list`` basis,
whitelist-scoped, USD only, unallocatable charges excluded, month-chunked
scans. Two contracts are specific to grouping and both exist because the
alternative is a silently wrong total:

**Dimensions the daily-aggregate fallback cannot answer.** A
subscription-month with no line items falls back to
``cmdb_cloud_billing_daily``, which carries no service / owner /
department / resource columns (and its per-service JSON is the *invoice*
basis, so it may not be emitted here at all). Those rows can be folded
into a ``subscription`` / ``provider`` / ``environment`` grouping, but
not into the other four. Rather than let that spend evaporate from the
breakdown, it is reported in ``meta.not_broken_down`` — amount plus the
subscriptions responsible — so a consumer can see that the parts do not
add up to the whole, and why.

**Rows nobody owns.** Facts whose allocation ladder resolved to nothing
(``auto_owner_source='none'``) are a first-class ``unallocated`` bucket
in the ``department`` / ``owner`` groupings, never dropped and never
folded into an arbitrary owner. Likewise ``resource`` grouping reports
``meta.without_resource_id`` — GCP's standard billing export carries no
resource column at all, so a Top-N that silently covered 3% of a
provider's spend would be worse than no answer.
"""
from __future__ import annotations

import json
import re
import uuid as uuid_mod
from collections import defaultdict
from dataclasses import dataclass, field
from datetime import date
from decimal import Decimal
from typing import Any

from sqlalchemy import and_, case, 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, fact_amount_expr
from app.domains.cmdb.models import CmdbCloudBillingDaily, CmdbCostFact
from app.domains.cmdb.open_billing_service import (
    BillingQuery,
    SubscriptionRef,
    daily_aggregate_fallback,
    provisional_from,
)
from app.identity.person_models import IdentityPerson

GROUP_SUBSCRIPTION = "subscription"
GROUP_SERVICE = "service"
GROUP_PROVIDER = "provider"
GROUP_ENVIRONMENT = "environment"
GROUP_DEPARTMENT = "department"
GROUP_OWNER = "owner"
GROUP_RESOURCE = "resource"
GROUP_REGION = "region"
GROUP_RESOURCE_TYPE = "resource_type"
GROUP_COST_CENTER = "cost_center"

#: ``tag:<key>`` — the resource label the cloud provider carries on the
#: billing line (AWS cost-allocation tags, GCP labels, Aliyun tags).
#: The one dimension that can answer "what did team X spend", because org
#: structure does not match cloud-account structure.
TAG_PREFIX = "tag:"

#: Groupings the per-subscription aggregate fallback can be folded into
#: (their key is derivable from the subscription row itself).
_SUBSCRIPTION_DERIVED = frozenset({
    GROUP_SUBSCRIPTION, GROUP_PROVIDER, GROUP_ENVIRONMENT,
})

VALID_GROUP_BY = frozenset({
    GROUP_SUBSCRIPTION, GROUP_SERVICE, GROUP_PROVIDER, GROUP_ENVIRONMENT,
    GROUP_DEPARTMENT, GROUP_OWNER, GROUP_RESOURCE,
    GROUP_REGION, GROUP_RESOURCE_TYPE, GROUP_COST_CENTER,
})

#: How many dimensions one request may cross. Two covers the FinOps
#: questions people actually ask (team × environment, team × service) and
#: keeps the key cardinality — which multiplies — from exploding.
MAX_DIMENSIONS = 2

#: Key used for facts the allocation ladder could not attribute.
UNALLOCATED_KEY = "unallocated"

#: Key for facts that do not carry the requested tag at all. A
#: first-class row, never dropped: measured on production 2026-08-15,
#: ~21% of charges carry no tag, and three providers (Azure / UCloud /
#: Tencent) contribute none structurally — a tag chart that silently
#: omitted that fifth of the money would be worse than no chart.
UNTAGGED_KEY = "untagged"

#: Display label for a group key that is genuinely empty in the data
#: (e.g. a driver that reported no service name). The key itself stays
#: empty — this only keeps a blank slice from reaching a chart.
UNKNOWN_LABEL = "unknown"

DEFAULT_TOP = 20
MAX_TOP = 100

#: Upper bound on distinct group keys held in memory while chunks are
#: merged. Only ``resource`` can realistically approach it (a busy tenant
#: emits a new resource id per ephemeral instance), and the merge must
#: happen on full per-chunk sets — a resource that is large in month 2 and
#: absent in month 1 would be mis-ranked by per-chunk truncation. Over the
#: cap → 422 asking the caller to narrow, never a silently partial ranking.
MAX_GROUP_KEYS = 100_000


@dataclass
class BreakdownResult:
    rows: list[dict[str, Any]] = field(default_factory=list)
    total_usd: Decimal = Decimal("0")
    #: The dimensions actually grouped by, echoed for the consumer.
    dimensions: list[str] = field(default_factory=list)
    #: Present only for a tag grouping: how much of the window's spend
    #: carries the requested tag at all.
    tag_coverage: dict[str, Any] | None = None
    #: Providers whose spend in this window carries no tags AT ALL —
    #: structural, not a data-entry gap (their drivers write an empty tag
    #: map). Named so a consumer knows which clouds a tag view cannot see.
    providers_without_tags: set[str] = field(default_factory=set)
    #: Latest day with any data for the visible subscriptions — same
    #: freshness signal the daily / monthly endpoints emit, so a consumer
    #: judges all three the same way.
    data_through: date | None = None
    #: Spend that exists but cannot be split along the requested
    #: dimension (aggregate-fallback subscriptions).
    not_broken_down_usd: Decimal = Decimal("0")
    not_broken_down_subscriptions: list[dict[str, Any]] = field(default_factory=list)
    #: ``resource`` grouping only: spend whose facts carry no resource id.
    without_resource_id_usd: Decimal | None = None
    truncated_to_top: int | None = None
    unconverted_by_currency: dict[str, float] = field(default_factory=dict)
    sources: set[str] = field(default_factory=set)


def _fact_scope_filter(subs: list[SubscriptionRef]) -> list[Any]:
    """WHERE terms selecting exactly these subscriptions' facts.

    Tenant-level subscriptions match on ``scope_id``; resource-group ones
    on ``(integration_id, resource_group)``. Facts with ``scope_id IS
    NULL`` (tenant-level charges) match neither, which is the point —
    they belong to no subscription.
    """
    tenant_ids = [s.id for s in subs if not s.is_resource_group]
    rg_pairs = [
        (s.integration_id, s.external_id) for s in subs if s.is_resource_group
    ]
    terms: list[Any] = []
    if tenant_ids:
        terms.append(CmdbCostFact.scope_id.in_(tenant_ids))
    for integ_id, rg_name in rg_pairs:
        terms.append(and_(
            CmdbCostFact.integration_id == integ_id,
            CmdbCostFact.resource_group == rg_name,
        ))
    if not terms:
        return []
    return [or_(*terms) if len(terms) > 1 else terms[0]]


#: Tag keys are interpolated into a JSON path, so they are held to the
#: same regex the internal cost surface uses. Do NOT loosen it without
#: re-auditing :meth:`Dimension.column` — the ``$."<key>"`` form relies on
#: ``"`` and ``\\`` being rejected (they would close or escape the
#: member-name quote). Other path metacharacters are inert inside a quoted
#: member name but stay rejected for defence in depth.
_TAG_KEY_RE = re.compile(r"^[A-Za-z0-9_\-.:/+@]{1,128}$")

_PLAIN_COLUMNS: dict[str, Any] = {
    GROUP_SERVICE: CmdbCostFact.service,
    GROUP_DEPARTMENT: CmdbCostFact.cost_owner_department,
    GROUP_OWNER: CmdbCostFact.cost_owner_person_id,
    GROUP_RESOURCE: CmdbCostFact.resource_id,
    GROUP_REGION: CmdbCostFact.region,
    GROUP_RESOURCE_TYPE: CmdbCostFact.resource_type,
    GROUP_COST_CENTER: CmdbCostFact.cost_center,
}


@dataclass(frozen=True)
class Dimension:
    """One axis of a breakdown: a plain fact column or a tag key."""

    #: As the caller wrote it — ``service`` / ``tag:usedby``. Echoed back
    #: in ``meta.dimensions`` and used as the per-row key name.
    spec: str
    #: ``None`` for a plain column; the tag key for a ``tag:`` dimension.
    tag_key: str | None = None

    @property
    def is_tag(self) -> bool:
        return self.tag_key is not None

    @property
    def is_subscription_derived(self) -> bool:
        return self.spec in _SUBSCRIPTION_DERIVED

    def column(self) -> Any:
        if self.tag_key is not None:
            # Re-validate at the interpolation site: a future caller might
            # skip :func:`parse_dimensions`, and this is the one place a
            # bad key would reach SQL.
            if not _TAG_KEY_RE.match(self.tag_key):
                raise ValueError(
                    f"refused unvalidated tag key {self.tag_key!r}",
                )
            return func.json_unquote(
                func.json_extract(CmdbCostFact.tags, f'$."{self.tag_key}"'),
            )
        column = _PLAIN_COLUMNS.get(self.spec)
        if column is not None:
            return column
        # subscription / provider / environment all key off the fact's
        # scope and are folded to their label afterwards.
        return CmdbCostFact.scope_id


def parse_dimensions(group_by: str) -> list[Dimension]:
    """Parse ``group_by`` into 1..MAX_DIMENSIONS dimensions.

    Accepts ``service``, ``tag:usedby``, ``tag:team,environment`` … The
    error messages name the vocabulary rather than just refusing, because
    the caller is a machine integration whose author cannot read our
    source.
    """
    specs = [part.strip() for part in (group_by or "").split(",")]
    specs = [part for part in specs if part]
    if not specs:
        raise ValidationError("group_by must name at least one dimension.")
    if len(specs) > MAX_DIMENSIONS:
        raise ValidationError(
            f"group_by accepts at most {MAX_DIMENSIONS} dimensions "
            f"(got {len(specs)}). Cross two at a time — e.g. "
            "`tag:team,environment`.",
        )
    if len(set(specs)) != len(specs):
        raise ValidationError("group_by lists the same dimension twice.")

    dims: list[Dimension] = []
    for spec in specs:
        if spec.startswith(TAG_PREFIX):
            key = spec[len(TAG_PREFIX):].strip()
            if not key:
                raise ValidationError(
                    "a tag dimension needs a key: `tag:<key>`, e.g. "
                    "`tag:usedby`. Discover the available keys with "
                    "GET /billing/tag-keys.",
                )
            if not _TAG_KEY_RE.match(key):
                raise ValidationError(
                    f"invalid tag key {key!r}: letters, digits and "
                    "_-.:/+@ only, max 128 chars.",
                )
            dims.append(Dimension(spec=f"{TAG_PREFIX}{key}", tag_key=key))
            continue
        if spec not in VALID_GROUP_BY:
            raise ValidationError(
                f"unknown dimension {spec!r}; use one of "
                f"{sorted(VALID_GROUP_BY)} or `tag:<key>`.",
            )
        dims.append(Dimension(spec=spec))
    return dims


async def _facts_grouped(
    db: AsyncSession,
    *,
    subs: list[SubscriptionRef],
    query: BillingQuery,
    dims: list[Dimension],
) -> tuple[dict[tuple, Decimal], dict[str, str | None], Decimal]:
    """``key-tuple → amount`` from cost_facts, month-chunked.

    The key tuple has one entry per requested dimension, already
    normalized (``unallocated`` / ``untagged`` / re-keyed RG subscription).
    Also returns resource-name labels and, when a ``resource`` dimension is
    present, the amount carried by facts with no ``resource_id`` at all.
    """
    amount = fact_amount_expr(COST_MEASURE_LIST)
    scope_filter = _fact_scope_filter(subs)
    totals: dict[tuple, Decimal] = defaultdict(Decimal)
    labels: dict[str, str | None] = {}
    without_resource = Decimal("0")
    if not scope_filter:
        return {}, {}, without_resource

    # RG-typed subscriptions carry their facts under ``(integration_id,
    # resource_group)`` with a NULL ``scope_id``. The subscription-derived
    # dimensions key off ``scope_id``, so without this lookup every RG fact
    # would land under the NULL key and be dropped when the rows are
    # folded — the sibling ``/billing/daily`` reports that same spend, so
    # the two surfaces would disagree and ``total_usd`` would quietly
    # under-report. (Caught in review of PR #2750.)
    rg_lookup = {
        (s.integration_id, s.external_id): s.id
        for s in subs if s.is_resource_group
    }
    by_ref = {s.id: s for s in subs}
    wants_rg_rekey = bool(rg_lookup) and any(
        d.is_subscription_derived for d in dims
    )
    wants_resource_name = any(d.spec == GROUP_RESOURCE for d in dims)
    # ``auto_owner_source`` distinguishes "genuinely unattributed" from
    # "attributed to a row whose department happens to be NULL" — both land
    # in the unallocated bucket, but only the first is expected.
    wants_owner_source = any(
        d.spec in (GROUP_DEPARTMENT, GROUP_OWNER) for d in dims
    )

    dim_columns = [d.column().label(f"dim{i}") for i, d in enumerate(dims)]

    for chunk_start, chunk_end in _month_chunks(query.start, query.end):
        cols: list[Any] = [
            *dim_columns,
            func.coalesce(func.sum(amount), 0).label("amount"),
        ]
        group: list[Any] = list(dim_columns)
        if wants_rg_rekey:
            cols.extend([CmdbCostFact.integration_id, CmdbCostFact.resource_group])
            group.extend([CmdbCostFact.integration_id, CmdbCostFact.resource_group])
        if wants_resource_name:
            # Carry a display name alongside the id; grouping by both is
            # safe because the pair is 1:1 in practice and a rename just
            # yields two rows rather than a wrong sum under one label.
            cols.append(CmdbCostFact.resource_name)
            group.append(CmdbCostFact.resource_name)
        if wants_owner_source:
            cols.append(CmdbCostFact.auto_owner_source)
            group.append(CmdbCostFact.auto_owner_source)

        rows = (await db.execute(
            select(*cols).where(
                *scope_filter,
                CmdbCostFact.date >= chunk_start,
                CmdbCostFact.date <= chunk_end,
            ).group_by(*group),
        )).all()

        for row in rows:
            value = Decimal(str(row.amount or 0))
            key: list[Any] = []
            drop = False
            for index, dim in enumerate(dims):
                raw = row[index]
                if dim.is_tag:
                    # Absent key → NULL; explicitly empty value → "".
                    # Neither is a usable slice, and a chart would render
                    # both as a nameless wedge.
                    key.append(UNTAGGED_KEY if raw in (None, "") else str(raw))
                    continue
                if dim.is_subscription_derived:
                    if raw is None and wants_rg_rekey:
                        raw = rg_lookup.get(
                            (str(row.integration_id), row.resource_group),
                        )
                    if raw is None:
                        drop = True
                        break
                    # provider / environment collapse the scope id into
                    # the subscription's attribute HERE, not at row-
                    # folding time: two scopes of the same provider must
                    # sum into one row, which a post-hoc rename cannot do.
                    folded = _subscription_derived_key(
                        dim, by_ref.get(str(raw)), str(raw),
                    )
                    if folded is None:
                        drop = True
                        break
                    key.append(folded)
                    continue
                if dim.spec == GROUP_RESOURCE:
                    if raw is None:
                        without_resource += value
                        drop = True
                        break
                    labels.setdefault(str(raw), row.resource_name)
                    key.append(str(raw))
                    continue
                if dim.spec in (GROUP_DEPARTMENT, GROUP_OWNER):
                    if raw is None or row.auto_owner_source == "none":
                        key.append(UNALLOCATED_KEY)
                    else:
                        key.append(str(raw))
                    continue
                key.append("" if raw is None else str(raw))
            if drop:
                continue
            totals[tuple(key)] += value

        if len(totals) > MAX_GROUP_KEYS:
            raise ValidationError(
                f"Too many distinct combinations of "
                f"{', '.join(d.spec for d in dims)} in this window "
                f"({len(totals)}+, cap {MAX_GROUP_KEYS}). Narrow the date "
                "range, filter to fewer subscriptions, or cross fewer "
                "dimensions.",
            )

    return dict(totals), labels, without_resource


async def _owner_labels(
    db: AsyncSession, person_ids: list[str],
) -> dict[str, dict[str, Any]]:
    if not person_ids:
        return {}
    rows = (await db.execute(
        select(
            IdentityPerson.id,
            IdentityPerson.display_name,
            IdentityPerson.work_email,
        ).where(
            IdentityPerson.id.in_([uuid_mod.UUID(p) for p in person_ids]),
        ),
    )).all()
    return {
        str(pid): {"label": name or email or str(pid), "email": email}
        for pid, name, email in rows
    }


async def breakdown(
    db: AsyncSession,
    *,
    subs: list[SubscriptionRef],
    query: BillingQuery,
    dims: list[Dimension],
    top: int = DEFAULT_TOP,
) -> BreakdownResult:
    """Window total split along one or two dimensions, largest first."""
    result = BreakdownResult()
    result.dimensions = [d.spec for d in dims]
    if not subs:
        return result

    totals, labels, without_resource = await _facts_grouped(
        db, subs=subs, query=query, dims=dims,
    )
    if totals or without_resource:
        result.sources.add("cost_facts")

    # Fallback spend: whole subscription-months the line-item table has
    # no rows for. Fold it in when every requested dimension is derivable
    # from the subscription itself; disclose it otherwise.
    # The fallback decides per scope-month, so it needs the (scope, day)
    # cells the fact table already covers.
    covered = await _covered_scope_days(db, subs=subs, query=query)

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

    by_ref = {s.id: s for s in subs}
    fallback_by_scope: dict[str, Decimal] = defaultdict(Decimal)
    for (sub_id, _day, _service), value in fallback.items():
        fallback_by_scope[sub_id] += value

    # Only when EVERY dimension is subscription-derived can the aggregate
    # row be placed: with a mixed request (subscription × tag) we know the
    # subscription but not the tag, so the amount is disclosed instead of
    # being parked under a guessed key.
    absorbable = bool(dims) and all(d.is_subscription_derived for d in dims)
    if absorbable:
        for sub_id, value in fallback_by_scope.items():
            key = tuple(
                _subscription_derived_key(dim, by_ref.get(sub_id), sub_id)
                for dim in dims
            )
            if any(part is None for part in key):
                continue
            totals[key] = totals.get(key, Decimal("0")) + value
    else:
        for sub_id, value in fallback_by_scope.items():
            ref = by_ref.get(sub_id)
            # A NEGATIVE amount is real and must be disclosed, not
            # dropped: a scope-month can net out below zero on credits /
            # refunds / adjustments (production has seen a single credit
            # day swing an account by six figures). Skipping it would
            # overstate ``total_usd`` and put this endpoint at odds with
            # ``/billing/daily`` and with ``group_by=subscription``, which
            # absorb the same value unconditionally. Only an exact zero is
            # skipped — it carries no information.
            if ref is None or value == 0:
                continue
            result.not_broken_down_usd += value
            result.not_broken_down_subscriptions.append({
                **ref.label(),
                "amount_usd": float(value),
            })

    rows = _to_rows(totals, labels=labels, by_ref=by_ref, dims=dims)

    owner_index = next(
        (i for i, d in enumerate(dims) if d.spec == GROUP_OWNER), None,
    )
    if owner_index is not None:
        person_ids = [
            r["keys"][dims[owner_index].spec] for r in rows
            if r["keys"][dims[owner_index].spec] != UNALLOCATED_KEY
        ]
        owner_meta = await _owner_labels(db, person_ids)
        for row in rows:
            meta = owner_meta.get(row["keys"][dims[owner_index].spec])
            if meta:
                row["labels"][dims[owner_index].spec] = meta["label"]
                row["email"] = meta["email"]
                row["label"] = " · ".join(row["labels"].values())
                if len(dims) == 1:
                    row["label"] = meta["label"]

    rows.sort(key=lambda r: (-r["amount_usd"], r["label"] or ""))
    total = sum((Decimal(str(r["amount_usd"])) for r in rows), Decimal("0"))
    if any(d.spec == GROUP_RESOURCE for d in dims):
        result.without_resource_id_usd = without_resource
    if len(rows) > top:
        # Top-N caps every grouping, not just ``resource``: crossing two
        # dimensions multiplies cardinality, so a tag × service request on
        # a busy tenant can produce thousands of rows where the tail is
        # noise. ``total_usd`` still reports the whole window, and
        # ``truncated_to_top`` says the rows are a head.
        result.truncated_to_top = top
        rows = rows[:top]

    # A share is only meaningful against a positive base. A window whose
    # grouped total nets to zero or below (an all-credits month) would
    # otherwise produce shares that are infinite, or negative for a
    # positive charge — ``null`` says "not meaningful here" instead.
    total_f = float(total)
    for row in rows:
        row["share"] = (
            round(row["amount_usd"] / total_f, 6) if total_f > 0 else None
        )

    result.data_through = await _data_through(db, subs=subs, query=query)
    result.rows = rows
    # Total is the window's full spend for the visible subscriptions —
    # NOT the sum of the (possibly Top-N-truncated) rows, so a consumer
    # can tell how much the shown rows actually cover.
    result.total_usd = (
        total + without_resource + result.not_broken_down_usd
    )
    if any(d.is_tag for d in dims):
        result.tag_coverage = _tag_coverage(rows, dims=dims, total=total)
        result.providers_without_tags = await _providers_without_tags(
            db, subs=subs, query=query,
        )
    return result


def _subscription_derived_key(
    dim: Dimension, ref: SubscriptionRef | None, sub_id: str,
) -> str | None:
    """The group key a subscription-derived dimension takes for one scope."""
    if dim.spec == GROUP_SUBSCRIPTION:
        return sub_id
    if ref is None:
        return None
    if dim.spec == GROUP_PROVIDER:
        return ref.provider or "unknown"
    return ref.environment or "unknown"


def _tag_coverage(
    rows: list[dict[str, Any]], *, dims: list[Dimension], total: Decimal,
) -> dict[str, Any]:
    """How much of the grouped spend actually carries the requested tag.

    Without this a consumer reads a tag chart as the whole picture. On
    production (2026-08-15) roughly a fifth of charges carry no tag at
    all, so "untagged" is not an edge case to hide in a footnote.
    """
    untagged = 0.0
    for row in rows:
        if any(
            row["keys"].get(d.spec) == UNTAGGED_KEY for d in dims if d.is_tag
        ):
            untagged += row["amount_usd"]
    total_f = float(total)
    return {
        "untagged_usd": round(untagged, 6),
        "tagged_usd": round(total_f - untagged, 6),
        "tagged_share": (
            round((total_f - untagged) / total_f, 6) if total_f > 0 else None
        ),
    }


#: Row cap on the tag-key discovery scan. ``tags`` is a JSON column with
#: no index and TiKV has no ``JSON_KEYS`` pushdown, so keys are collected
#: in Python off a bounded, recency-biased sample — the same shape the
#: internal autocomplete uses. The returned spend is therefore a SAMPLE
#: figure for ranking, never a total to quote.
TAG_KEY_SCAN_LIMIT = 200_000


async def tag_keys(
    db: AsyncSession,
    *,
    subs: list[SubscriptionRef],
    query: BillingQuery,
    limit: int = 100,
) -> tuple[list[dict[str, Any]], int]:
    """Which tag keys exist on the visible spend, ranked by money.

    Ranked by covered spend rather than row count: a key on ten large
    resources matters more than one on ten thousand rounding errors, and
    the caller is choosing a dimension to slice cost by.

    Case is reported exactly as stored — ``UsedBy`` and ``usedby`` list
    separately, because they ARE separate keys to the grouping endpoint.
    Merging them here would promise a merge the breakdown does not do.
    """
    scope_filter = _fact_scope_filter(subs)
    if not scope_filter:
        return [], 0

    amount = fact_amount_expr(COST_MEASURE_LIST)
    spend: dict[str, Decimal] = defaultdict(Decimal)
    providers: dict[str, set[str]] = defaultdict(set)
    provider_by_integration = {
        s.integration_id: s.provider for s in subs if s.provider
    }
    sampled = 0

    for chunk_start, chunk_end in _month_chunks(query.start, query.end):
        rows = (await db.execute(
            select(
                CmdbCostFact.tags,
                CmdbCostFact.integration_id,
                amount.label("amount"),
            ).where(
                *scope_filter,
                CmdbCostFact.date >= chunk_start,
                CmdbCostFact.date <= chunk_end,
                func.json_length(CmdbCostFact.tags) > 0,
            ).order_by(CmdbCostFact.date.desc()).limit(TAG_KEY_SCAN_LIMIT),
        )).all()
        for row in rows:
            sampled += 1
            tags = row.tags
            if isinstance(tags, (str, bytes)):
                try:
                    tags = json.loads(tags)
                except (ValueError, TypeError):
                    continue
            if not isinstance(tags, dict):
                continue
            value = Decimal(str(row.amount or 0))
            provider = provider_by_integration.get(str(row.integration_id))
            for key in tags:
                name = str(key)
                spend[name] += value
                if provider:
                    providers[name].add(provider)

    ranked = sorted(spend.items(), key=lambda kv: -kv[1])[:limit]
    return [
        {
            "key": name,
            "sampled_spend_usd": float(value),
            "providers": sorted(providers.get(name, ())),
        }
        for name, value in ranked
    ], sampled


async def _providers_without_tags(
    db: AsyncSession, *, subs: list[SubscriptionRef], query: BillingQuery,
) -> set[str]:
    """Providers with spend in this window but zero tagged spend.

    Measured, not hard-coded: today Azure / UCloud / Tencent write an
    empty tag map from their drivers, but that is a driver fact that can
    change, and a hard-coded list would quietly go stale. Naming them in
    the response is what stops a consumer reading "Azure spent nothing"
    off a tag chart that structurally cannot see Azure.
    """
    scope_filter = _fact_scope_filter(subs)
    if not scope_filter:
        return set()
    by_integration: dict[str, list[Decimal]] = defaultdict(
        lambda: [Decimal("0"), Decimal("0")],
    )
    amount = fact_amount_expr(COST_MEASURE_LIST)
    tagged_amount = case(
        (func.json_length(CmdbCostFact.tags) > 0, amount), else_=0,
    )
    for chunk_start, chunk_end in _month_chunks(query.start, query.end):
        rows = (await db.execute(
            select(
                CmdbCostFact.integration_id,
                func.coalesce(func.sum(amount), 0).label("total"),
                func.coalesce(func.sum(tagged_amount), 0).label("tagged"),
            ).where(
                *scope_filter,
                CmdbCostFact.date >= chunk_start,
                CmdbCostFact.date <= chunk_end,
            ).group_by(CmdbCostFact.integration_id),
        )).all()
        for row in rows:
            slot = by_integration[str(row.integration_id)]
            slot[0] += Decimal(str(row.total or 0))
            slot[1] += Decimal(str(row.tagged or 0))

    provider_by_integration = {
        s.integration_id: s.provider for s in subs if s.provider
    }
    out: set[str] = set()
    tagged_providers: set[str] = set()
    for integration_id, (total, tagged) in by_integration.items():
        provider = provider_by_integration.get(integration_id)
        if provider is None:
            continue
        if tagged > 0:
            tagged_providers.add(provider)
        elif total != 0:
            out.add(provider)
    # A provider with several tenants counts as covered if ANY tenant
    # tags — otherwise one untagged sub-account would libel the provider.
    return out - tagged_providers


async def _covered_scope_days(
    db: AsyncSession, *, subs: list[SubscriptionRef], query: BillingQuery,
) -> set[tuple[str, Any]]:
    """(scope_id, date) pairs the fact table already covers.

    Same input the daily endpoint's fallback decision uses; computed here
    with a narrow projection (no amounts) because the breakdown does not
    need the per-day values.
    """
    scope_filter = _fact_scope_filter(subs)
    if not scope_filter:
        return set()
    rg_lookup = {
        (s.integration_id, s.external_id): s.id
        for s in subs if s.is_resource_group
    }
    covered: set[tuple[str, Any]] = set()
    for chunk_start, chunk_end in _month_chunks(query.start, query.end):
        rows = (await db.execute(
            select(
                CmdbCostFact.scope_id,
                CmdbCostFact.integration_id,
                CmdbCostFact.resource_group,
                CmdbCostFact.date,
            ).where(
                *scope_filter,
                CmdbCostFact.date >= chunk_start,
                CmdbCostFact.date <= chunk_end,
            ).group_by(
                CmdbCostFact.scope_id,
                CmdbCostFact.integration_id,
                CmdbCostFact.resource_group,
                CmdbCostFact.date,
            ),
        )).all()
        for row in rows:
            if row.scope_id:
                covered.add((str(row.scope_id), row.date))
                continue
            sub_id = rg_lookup.get((str(row.integration_id), row.resource_group))
            if sub_id:
                covered.add((sub_id, row.date))
    return covered


def _to_rows(
    totals: dict[tuple, Decimal],
    *,
    labels: dict[str, str | None],
    by_ref: dict[str, SubscriptionRef],
    dims: list[Dimension],
) -> list[dict[str, Any]]:
    """Fold raw key tuples into wire rows.

    Single-dimension rows keep the shape the API already ships (``key`` /
    ``label`` plus the subscription identity fields) — that contract is
    live and documented. ``keys`` / ``labels`` are added alongside for
    every request, so a consumer can read a one- or two-dimension response
    the same way. Additive, so nothing that works today breaks.
    """
    rows: list[dict[str, Any]] = []
    for key_tuple, value in totals.items():
        keys: dict[str, str] = {}
        row_labels: dict[str, str] = {}
        extra: dict[str, Any] = {}
        for dim, raw in zip(dims, key_tuple, strict=True):
            keys[dim.spec] = raw
            row_labels[dim.spec] = _label_for(dim, raw, labels=labels, by_ref=by_ref)
            if dim.spec == GROUP_SUBSCRIPTION:
                ref = by_ref.get(raw)
                if ref is not None:
                    extra.update({
                        "external_id": ref.external_id,
                        "provider": ref.provider,
                        "environment": ref.environment,
                    })
        row = {
            "key": key_tuple[0] if len(dims) == 1 else " · ".join(key_tuple),
            "label": " · ".join(row_labels.values()),
            "keys": keys,
            "labels": row_labels,
            "amount_usd": float(value),
            **extra,
        }
        rows.append(row)
    return rows


def _label_for(
    dim: Dimension,
    raw: str,
    *,
    labels: dict[str, str | None],
    by_ref: dict[str, SubscriptionRef],
) -> str:
    if dim.spec == GROUP_SUBSCRIPTION:
        ref = by_ref.get(raw)
        return (ref.name or ref.external_id) if ref else raw
    if dim.spec == GROUP_RESOURCE:
        return labels.get(raw) or raw
    if raw in (UNALLOCATED_KEY, UNTAGGED_KEY):
        return raw
    # A driver that reported no value would otherwise render as a blank
    # slice with no way to ask what it is. The KEY stays empty (that is
    # the real value in the data); only the human label says so out loud.
    return raw or UNKNOWN_LABEL


async def _data_through(
    db: AsyncSession, *, subs: list[SubscriptionRef], query: BillingQuery,
) -> date | None:
    """Latest day carrying data, across both sources.

    One small max() per table, both index-ranged on ``date`` — cheap
    enough to keep the breakdown's freshness signal identical to the
    daily and monthly endpoints' rather than absent.
    """
    scope_filter = _fact_scope_filter(subs)
    candidates: list[date] = []
    if scope_filter:
        value = (await db.execute(
            select(func.max(CmdbCostFact.date)).where(
                *scope_filter,
                CmdbCostFact.date >= query.start,
                CmdbCostFact.date <= query.end,
            ),
        )).scalar()
        if value:
            candidates.append(value)
    tenant_ids = [s.id for s in subs if not s.is_resource_group]
    if tenant_ids:
        value = (await db.execute(
            select(func.max(CmdbCloudBillingDaily.date)).where(
                CmdbCloudBillingDaily.scope_id.in_(tenant_ids),
                CmdbCloudBillingDaily.date >= query.start,
                CmdbCloudBillingDaily.date <= query.end,
            ),
        )).scalar()
        if value:
            candidates.append(value)
    return max(candidates) if candidates else None


def breakdown_meta(result: BreakdownResult, *, group_by: str) -> dict[str, Any]:
    """Grouping-specific half of the response meta."""
    meta: dict[str, Any] = {
        "group_by": group_by,
        "dimensions": list(result.dimensions),
        "total_usd": float(result.total_usd),
        "data_through": (
            result.data_through.isoformat() if result.data_through else None
        ),
        "sources": sorted(result.sources),
    }
    if result.not_broken_down_subscriptions:
        # The parts do not add up to the whole, and here is exactly why.
        # Gated on the SUBSCRIPTION LIST, not on a positive amount: a
        # window whose un-splittable spend nets to zero (a charge and an
        # equal credit) still has spend the consumer cannot see in the
        # rows, and a negative net is precisely the case worth naming.
        meta["not_broken_down"] = {
            "amount_usd": float(result.not_broken_down_usd),
            "reason": (
                "These subscriptions had no line-item data for part of the "
                "window, so their spend came from the daily aggregate, which "
                f"carries no {group_by} dimension."
            ),
            # Largest ABSOLUTE amount first: a −$70k credit is as
            # noteworthy as a +$70k charge, and sorting by signed value
            # would bury it at the bottom of a truncated list.
            "subscriptions": sorted(
                result.not_broken_down_subscriptions,
                key=lambda s: -abs(s["amount_usd"]),
            )[:50],
        }
    if result.tag_coverage is not None:
        # A tag chart without this reads as the whole picture. It is not:
        # ~21% of production charges carry no tag, and three providers
        # contribute none at all (see ``providers_without_tags``).
        meta["tag_coverage"] = result.tag_coverage
    if result.providers_without_tags:
        meta["providers_without_tags"] = sorted(result.providers_without_tags)
    if result.without_resource_id_usd is not None:
        meta["without_resource_id_usd"] = float(result.without_resource_id_usd)
    if result.truncated_to_top is not None:
        meta["truncated_to_top"] = result.truncated_to_top
    if result.unconverted_by_currency:
        meta["unconverted_by_currency"] = result.unconverted_by_currency
    meta["provisional_from"] = provisional_from().isoformat()
    return meta


__all__ = [
    "DEFAULT_TOP",
    "GROUP_DEPARTMENT",
    "GROUP_ENVIRONMENT",
    "GROUP_OWNER",
    "GROUP_PROVIDER",
    "GROUP_RESOURCE",
    "GROUP_SERVICE",
    "GROUP_SUBSCRIPTION",
    "MAX_TOP",
    "MAX_DIMENSIONS",
    "TAG_PREFIX",
    "UNALLOCATED_KEY",
    "UNKNOWN_LABEL",
    "UNTAGGED_KEY",
    "Dimension",
    "VALID_GROUP_BY",
    "BreakdownResult",
    "TAG_KEY_SCAN_LIMIT",
    "breakdown",
    "breakdown_meta",
    "parse_dimensions",
    "tag_keys",
]