"""Incremental-pull signal for the cloud billing External API. Answers "which (subscription, day) cells were re-ingested since T", so a consumer can re-pull those instead of the whole window every night. **What the signal actually is.** Not a per-row change feed — an *ingest watermark*. A day whose amount was re-fetched and came back identical is still reported: the alternative (diffing amounts) would mean storing a previous value per cell, and a consumer that overwrites by primary key does not care. Over-reporting is safe here; under-reporting is not. **Which column, measured rather than assumed.** The two source tables are mirror images of each other, and picking wrong yields a feed that silently returns nothing: ====================== ========================== ============== table advances on re-ingest frozen ====================== ========================== ============== cmdb_cloud_billing_daily ``fetched_at`` ``updated_at`` cmdb_cost_facts ``updated_at`` ``fetched_at`` ====================== ========================== ============== (Probed on the real ``INSERT … ON DUPLICATE KEY UPDATE`` paths on 2026-08-15; each table's SET list contains exactly one of the two, and neither DDL carries ``ON UPDATE CURRENT_TIMESTAMP``.) **The horizon is the honest part.** Both queries are bounded to ``date >= horizon`` so neither degenerates into a full-table scan: the daily table's is an index range on ``fetched_at`` (migration 471) and the facts one rides ``ix_facts_scope_date``. The cost of that bound is real and is stated in the response: a backfill that rewrites a day older than the horizon is **not** detected, and the consumer is told to re-pull such a window explicitly. Silently pretending to be a complete change feed would be the worse trade. """ from __future__ import annotations from datetime import UTC, date, datetime, timedelta from typing import Any from sqlalchemy import and_, func, or_, select from sqlalchemy.ext.asyncio import AsyncSession from app.domains.cmdb.models import CmdbCloudBillingDaily, CmdbCostFact from app.domains.cmdb.open_billing_service import SubscriptionRef #: Default look-back for change detection. Comfortably wider than the #: crons' 3-day rolling re-ingest window, so a consumer that polls daily #: (or misses a few days) still sees everything that moved. DEFAULT_HORIZON_DAYS = 32 #: Hard cap. Beyond this the bounded scans stop being cheap, and a #: consumer that far behind should just re-pull the window. MAX_HORIZON_DAYS = 92 SIGNAL_DAILY_AGGREGATE = "billing_daily.fetched_at" SIGNAL_FACTS = "cost_facts.updated_at" def horizon_start(horizon_days: int, *, today: date | None = None) -> date: return (today or datetime.now(UTC).date()) - timedelta(days=horizon_days) async def changed_cells( db: AsyncSession, *, subs: list[SubscriptionRef], since: datetime, horizon: date, ) -> tuple[dict[tuple[str, date], datetime], set[str]]: """``(subscription_id, date) → last ingest time``, plus the signals that actually contributed. Both source tables are consulted because they are re-ingested by independent crons: a manual line-item backfill moves only the facts table, and a daily-aggregate re-fetch moves only the aggregate one. Reading one would miss half the changes. """ out: dict[tuple[str, date], datetime] = {} signals: set[str] = set() if not subs: return out, signals tenant_ids = [s.id for s in subs if not s.is_resource_group] by_rg = { (s.integration_id, s.external_id): s.id for s in subs if s.is_resource_group } if tenant_ids: rows = (await db.execute( select( CmdbCloudBillingDaily.scope_id, CmdbCloudBillingDaily.date, func.max(CmdbCloudBillingDaily.fetched_at).label("at"), ).where( CmdbCloudBillingDaily.fetched_at > since, CmdbCloudBillingDaily.date >= horizon, CmdbCloudBillingDaily.scope_id.in_(tenant_ids), ).group_by( CmdbCloudBillingDaily.scope_id, CmdbCloudBillingDaily.date, ), )).all() for daily_row in rows: if daily_row.at is None: continue signals.add(SIGNAL_DAILY_AGGREGATE) key = (str(daily_row.scope_id), daily_row.date) if key not in out or out[key] < daily_row.at: out[key] = daily_row.at # Facts side. Scoped + date-bounded so it rides ix_facts_scope_date # rather than scanning the whole (multi-million-row) table on a # timestamp with no index of its own. if tenant_ids: fact_rows = (await db.execute( select( CmdbCostFact.scope_id, CmdbCostFact.date, func.max(CmdbCostFact.updated_at).label("at"), ).where( CmdbCostFact.scope_id.in_(tenant_ids), CmdbCostFact.date >= horizon, CmdbCostFact.updated_at > since, ).group_by(CmdbCostFact.scope_id, CmdbCostFact.date), )).all() for fact_row in fact_rows: if fact_row.at is None: continue signals.add(SIGNAL_FACTS) key = (str(fact_row.scope_id), fact_row.date) if key not in out or out[key] < fact_row.at: out[key] = fact_row.at if by_rg: # One query for every resource-group subscription, not one each: # a client whitelisted to a few dozen Azure RGs would otherwise # pay a round trip per RG on every incremental poll. rg_rows = (await db.execute( select( CmdbCostFact.integration_id, CmdbCostFact.resource_group, CmdbCostFact.date, func.max(CmdbCostFact.updated_at).label("at"), ).where( or_(*[ and_( CmdbCostFact.integration_id == integ_id, CmdbCostFact.resource_group == rg_name, ) for integ_id, rg_name in by_rg ]), CmdbCostFact.date >= horizon, CmdbCostFact.updated_at > since, ).group_by( CmdbCostFact.integration_id, CmdbCostFact.resource_group, CmdbCostFact.date, ), )).all() for rg_row in rg_rows: if rg_row.at is None: continue sub_id = by_rg.get( (str(rg_row.integration_id), rg_row.resource_group), ) if sub_id is None: continue signals.add(SIGNAL_FACTS) key = (sub_id, rg_row.date) if key not in out or out[key] < rg_row.at: out[key] = rg_row.at return out, signals def changes_rows( cells: dict[tuple[str, date], datetime], *, subs: list[SubscriptionRef], ) -> list[dict[str, Any]]: by_ref = {s.id: s for s in subs} rows = [] for (sub_id, day), at in cells.items(): ref = by_ref.get(sub_id) if ref is None: continue rows.append({ "date": day.isoformat(), **ref.label(), "last_ingested_at": at.isoformat() if at else None, }) rows.sort(key=lambda r: (r["date"], r["external_id"])) return rows def changes_meta( *, since: datetime, horizon: date, signals: set[str], cell_count: int, ) -> dict[str, Any]: """The block that keeps this from being mistaken for a full change feed.""" return { "since": since.isoformat(), "horizon": horizon.isoformat(), "changed_cells": cell_count, "signals": sorted(signals), "detects": ( "Any (subscription, day) whose cloud cost data was re-ingested " "on or after `since`, for days on or after `horizon`. A day " "re-fetched with an unchanged amount is included — this is an " "ingest watermark, not an amount diff." ), "does_not_detect": ( "Changes to days older than `horizon`. If an admin backfills a " "historical window, re-pull that window explicitly." ), } __all__ = [ "DEFAULT_HORIZON_DAYS", "MAX_HORIZON_DAYS", "SIGNAL_DAILY_AGGREGATE", "SIGNAL_FACTS", "changed_cells", "changes_meta", "changes_rows", "horizon_start", ]