"""Provider-neutral materialisation for the cloud resource directory. This module deliberately reads the established CMDB tables instead of making the new directory the source of truth. It is safe to run repeatedly, records only semantic changes, and never calls or mutates a cloud provider. """ from __future__ import annotations import asyncio import hashlib import json import logging import re import uuid from collections.abc import Callable from dataclasses import dataclass, replace from datetime import UTC, date, datetime, timedelta from decimal import Decimal from typing import Any, cast from sqlalchemy import delete, select from sqlalchemy.dialects.mysql import insert as mysql_insert from sqlalchemy.ext.asyncio import AsyncSession from app.domains.cmdb.models import ( CmdbCdnDomain, CmdbCertificate, CmdbCloudResource, CmdbCloudResourceChange, CmdbCloudScope, CmdbDnsRecord, CmdbDomain, CmdbIamUser, CmdbLoadBalancer, CmdbObjectStorage, CmdbResourceType, CmdbServer, ) from app.domains.cmdb.provider_factory import build_cloud_provider from app.domains.cmdb.providers.base import CloudResourceInventoryItem from app.domains.cmdb.providers.exceptions import CapabilityNotSupported from app.integrations.models import Integration logger = logging.getLogger(__name__) _DEFAULT_REFRESH_BATCH_SIZE = 100 _MAX_REFRESH_BATCH_SIZE = 500 def _canonical_connector_scope( provider: str, scopes: list[CmdbCloudScope], ) -> str | None: """Return the one unambiguous provider account behind a connector. Some provider APIs (notably Tencent DNS) return account-global inventory without repeating the account id on every object. A connector-level fallback is safe only for providers whose native account identifier has a strict shape and only when exactly one such active scope exists. Azure subscriptions and GCP projects are deliberately excluded because one connector commonly spans many scopes. """ candidates: set[str] = set() for scope in scopes: external_id = str(scope.external_id or "").strip() scope_type = str(scope.scope_type or "").strip().casefold() if not external_id or scope_type in { "azure_resource_group", "gcp_billing_account", "legacy_billing_unit", }: continue if provider == "aws" and external_id.isdigit() and len(external_id) == 12: candidates.add(external_id) elif provider in {"aliyun", "alicloud_intl", "tencent"} and external_id.isdigit(): candidates.add(external_id) elif provider == "ucloud" and external_id.startswith("org-"): candidates.add(external_id) return next(iter(candidates)) if len(candidates) == 1 else None @dataclass(frozen=True) class TypeDefinition: resource_type: str service: str category: str display_name: str icon_key: str sensitivity: str = "internal" discovered_dynamically: bool = False @dataclass(frozen=True) class Projection: definition: TypeDefinition provider_resource_id: str name: str source_table: str source_id: str scope_external_id: str | None region: str | None zone: str | None status: str | None environment: str | None owner_person_id: str | None tags: dict | None normalized: dict[str, Any] raw: dict | None source_sync_job_id: str | None first_seen_at: datetime | None last_seen_at: datetime | None deleted_at: datetime | None _TYPES = { "server": TypeDefinition("server", "compute", "compute", "Cloud server", "server"), "domain": TypeDefinition("domain", "dns", "network", "Domain", "global"), "dns_record": TypeDefinition("dns_record", "dns", "network", "DNS record", "dns"), "certificate": TypeDefinition("certificate", "certificate", "security", "Certificate", "certificate"), "load_balancer": TypeDefinition("load_balancer", "load_balancing", "network", "Load balancer", "gateway"), "cdn_domain": TypeDefinition("cdn_domain", "cdn", "network", "CDN domain", "cdn"), "object_storage": TypeDefinition("object_storage", "object_storage", "storage", "Object storage", "database"), "iam_user": TypeDefinition("iam_user", "iam", "identity", "IAM identity", "user", "restricted"), "cloud_scope": TypeDefinition("cloud_scope", "resource_manager", "account", "Cloud account / scope", "cloud"), } _SENSITIVE_KEY_PARTS = ( "access_key", "client_secret", "connection_string", "credential", "password", "private_key", "sas_token", "secret", "token", ) _SENSITIVE_VALUE_RE = re.compile( r"(?i)(-----BEGIN [A-Z ]*PRIVATE KEY-----|" r"\b(?:password|passwd|client_secret|access_token|refresh_token|secret_access_key|" r"connection_string)\s*[:=]\s*\S+|" r"\bbearer\s+[A-Za-z0-9._~+/=-]{12,}|\bAKIA[0-9A-Z]{16}\b)", ) _MAX_METADATA_DEPTH = 8 _MAX_METADATA_ITEMS = 256 _MAX_METADATA_NODES = 1_024 _MAX_METADATA_CHARS = 65_536 def _json_value(value: Any) -> Any: if isinstance(value, dict): return {str(k): _json_value(v) for k, v in sorted(value.items(), key=lambda i: str(i[0]))} if isinstance(value, (list, tuple, set)): return [_json_value(v) for v in value] if isinstance(value, (datetime, date)): return value.isoformat() if isinstance(value, (uuid.UUID, Decimal)): return str(value) if isinstance(value, (str, int, float, bool)) or value is None: return value return str(value) def sanitize_metadata(value: Any) -> Any: """Return deterministic, bounded metadata with likely credentials removed. Cloud inventory payloads are not trusted to be small or secret-free. The directory is a metadata index, so it intentionally drops excess structure and never persists values whose key looks credential-bearing. """ budget = {"nodes": _MAX_METADATA_NODES, "chars": _MAX_METADATA_CHARS} def visit(item: Any, *, depth: int, key: str | None = None) -> Any: if depth > _MAX_METADATA_DEPTH or budget["nodes"] <= 0 or budget["chars"] <= 0: return "[TRUNCATED]" budget["nodes"] -= 1 if key and any(part in key.casefold() for part in _SENSITIVE_KEY_PARTS): return "[REDACTED]" if isinstance(item, dict): result: dict[str, Any] = {} entries = sorted(item.items(), key=lambda pair: str(pair[0])) limited_entries = entries[:_MAX_METADATA_ITEMS] for index, (child_key, child) in enumerate(limited_entries): if budget["nodes"] <= 0 or budget["chars"] <= 0: result["__truncated_items__"] = len(entries) - index break string_key = str(child_key)[:512] result[string_key] = visit(child, depth=depth + 1, key=string_key) if len(entries) > _MAX_METADATA_ITEMS and "__truncated_items__" not in result: result["__truncated_items__"] = len(entries) - _MAX_METADATA_ITEMS return result if isinstance(item, (list, tuple, set)): items = sorted(item, key=repr) if isinstance(item, set) else list(item) list_result: list[Any] = [] limited_items = items[:_MAX_METADATA_ITEMS] for index, child in enumerate(limited_items): if budget["nodes"] <= 0 or budget["chars"] <= 0: list_result.append({"__truncated_items__": len(items) - index}) break list_result.append(visit(child, depth=depth + 1)) else: if len(items) > _MAX_METADATA_ITEMS: list_result.append({"__truncated_items__": len(items) - _MAX_METADATA_ITEMS}) return list_result scalar = _json_value(item) if isinstance(scalar, str): if _SENSITIVE_VALUE_RE.search(scalar): return "[REDACTED]" allowed = max(0, min(len(scalar), 4_096, budget["chars"])) budget["chars"] -= allowed scalar = scalar[:allowed] + ("[TRUNCATED]" if allowed < len(scalar) else "") return scalar return visit(value, depth=0) def canonical_snapshot(projection: Projection) -> dict[str, Any]: """Stable, secret-agnostic snapshot used for change detection.""" return cast(dict[str, Any], _json_value({ "provider_resource_id": projection.provider_resource_id, "name": projection.name, "scope_external_id": projection.scope_external_id, "region": projection.region, "zone": projection.zone, "status": projection.status, "environment": projection.environment, "owner_person_id": projection.owner_person_id, "tags": sanitize_metadata(projection.tags or {}), "normalized": sanitize_metadata(projection.normalized), "raw": sanitize_metadata(projection.raw or {}), "deleted": projection.deleted_at is not None, })) def snapshot_hash(snapshot: dict[str, Any]) -> str: encoded = json.dumps(snapshot, ensure_ascii=False, sort_keys=True, separators=(",", ":")) return hashlib.sha256(encoded.encode("utf-8")).hexdigest() def resource_identity_hash(resource_type: str, provider_resource_id: str) -> str: return hashlib.sha256(f"{resource_type}\0{provider_resource_id}".encode()).hexdigest() def changed_fields(before: dict[str, Any] | None, after: dict[str, Any]) -> list[str]: if before is None: return sorted(after) return sorted(key for key in set(before) | set(after) if before.get(key) != after.get(key)) def _raw_scope(raw: dict | None) -> str | None: raw = raw or {} for key in ("_aws_account_id", "__sub_account_id", "_subscription_id", "_project_id", "project_id", "subscription_id"): value = raw.get(key) if value: return str(value) return None def _str(value: Any) -> str | None: return str(value) if value is not None and str(value) else None def _iam_scope(row: CmdbIamUser) -> str | None: """Return the provider account carried by an IAM identity. AWS IAM rows historically left ``sub_account_id`` empty even though the account id is an immutable component of every user ARN. Reading it here repairs existing rows on the next directory refresh without rewriting the authoritative IAM mirror. """ if row.sub_account_id: return str(row.sub_account_id) raw_scope = _raw_scope(row.raw_data) if raw_scope: return raw_scope arn = str(row.arn or "") if arn.startswith("arn:aws:iam::"): account_id = arn.split(":", 5)[4] if account_id.isdigit() and len(account_id) == 12: return account_id return None def _projection_specs() -> list[tuple[Any, Callable[[Any], Projection]]]: return [ (CmdbServer, lambda r: Projection( _TYPES["server"], r.instance_id, r.name or r.instance_id, r.__tablename__, str(r.id), _raw_scope(r.raw_data) or _str(r.project_id), r.region, r.zone, r.status, (r.raw_data or {}).get("environment"), _str(r.owner_person_id), r.tags, { "instance_type": r.instance_type, "vcpu": r.vcpu, "memory_mb": r.memory_mb, "os_type": r.os_type, "public_ips": r.public_ips, "private_ips": r.private_ips, "vpc_id": r.vpc_id, "subnet_id": r.subnet_id, "app": r.app, }, r.raw_data, r.last_seen_sync_job_id, r.first_seen_at, r.last_synced_at, r.deleted_at, )), (CmdbDomain, lambda r: Projection( _TYPES["domain"], r.domain_name, r.domain_name, r.__tablename__, str(r.id), _raw_scope(r.raw_data), None, None, r.status, r.environment, _str(r.owner_person_id), None, {"registrar": r.registrar, "expiry_date": r.expiry_date, "auto_renew": r.auto_renew, "grade": r.grade}, r.raw_data, r.last_seen_sync_job_id, r.first_seen_at, r.last_synced_at, r.deleted_at, )), (CmdbDnsRecord, lambda r: Projection( _TYPES["dns_record"], f"{r.zone_name}:{r.record_id}", f"{r.name}.{r.zone_name}".strip("."), r.__tablename__, str(r.id), _raw_scope(r.raw_data), None, None, r.status, None, _str(r.owner_person_id), None, {"zone_name": r.zone_name, "record_type": r.record_type, "value": r.value, "ttl": r.ttl, "line": r.line}, r.raw_data, r.last_seen_sync_job_id, r.first_seen_at, r.last_synced_at, r.deleted_at, )), (CmdbCertificate, lambda r: Projection( _TYPES["certificate"], r.cloud_cert_id or r.dedup_key, r.common_name or r.probe_host or r.dedup_key, r.__tablename__, str(r.id), _raw_scope(r.raw_data), None, None, r.status, r.environment, _str(r.owner_person_id) if r.owner_is_manual else None, None, {"source": r.source, "issuer": r.issuer, "not_before": r.not_before, "not_after": r.not_after, "sans": r.sans, "key_algorithm": r.key_algorithm, "key_bits": r.key_bits}, r.raw_data, r.last_seen_sync_job_id, r.first_seen_at, r.last_synced_at or r.last_probed_at, r.deleted_at, )), (CmdbLoadBalancer, lambda r: Projection( _TYPES["load_balancer"], r.lb_id, r.name or r.lb_id, r.__tablename__, str(r.id), _raw_scope(r.raw_data) or _str(r.project_id), r.region, None, r.status, None, _str(r.owner_person_id), None, {"lb_type": r.lb_type, "network_type": r.network_type, "ip_address": r.ip_address, "vpc_id": r.vpc_id, "subnet_id": r.subnet_id, "listener_count": r.listener_count, "app": r.app}, r.raw_data, r.last_seen_sync_job_id, r.first_seen_at, r.last_synced_at, r.deleted_at, )), (CmdbCdnDomain, lambda r: Projection( _TYPES["cdn_domain"], r.domain_name, r.domain_name, r.__tablename__, str(r.id), _raw_scope(r.raw_data), None, None, r.status, None, _str(r.owner_person_id), None, {"cname": r.cname, "source_type": r.source_type, "source_address": r.source_address, "is_https": r.is_https}, r.raw_data, r.last_seen_sync_job_id, r.first_seen_at, r.last_synced_at, r.deleted_at, )), (CmdbObjectStorage, lambda r: Projection( _TYPES["object_storage"], r.bucket_name, r.bucket_name, r.__tablename__, str(r.id), _raw_scope(r.raw_data), r.region, None, r.status, None, _str(r.owner_person_id), None, {"storage_class": r.storage_class, "access_type": r.access_type, "size_bytes": r.size_bytes, "object_count": r.object_count}, r.raw_data, r.last_seen_sync_job_id, r.first_seen_at, r.last_synced_at, r.deleted_at, )), (CmdbIamUser, lambda r: Projection( _TYPES["iam_user"], r.arn or f"{r.sub_account_id}:{r.user_name}", r.user_name, r.__tablename__, str(r.id), _iam_scope(r), None, None, "revoked" if r.revoked_at else "active", None, _str(r.owner_person_id), r.tags, {"arn": r.arn, "account_kind": r.account_kind, "mfa_state": r.mfa_state, "groups": r.groups, "attached_policies": r.attached_policies, "last_login_at": r.last_login_at}, r.raw_data, r.last_seen_sync_job_id, r.first_seen_at, r.last_synced_at, r.deleted_at, )), (CmdbCloudScope, lambda r: Projection( _TYPES["cloud_scope"], f"{r.scope_type}:{r.external_id}", r.display_name or r.external_id, r.__tablename__, str(r.id), r.external_id, None, None, r.status, r.environment, _str(r.owner_person_id), None, {"scope_type": r.scope_type, "parent_scope_id": r.parent_scope_id, "decommissioned_at": r.decommissioned_at}, r.raw_data, r.last_seen_sync_job_id, r.first_seen_at, r.last_synced_at, r.decommissioned_at, )), ] async def _resource_type(db: AsyncSession, provider: str, definition: TypeDefinition) -> CmdbResourceType: stmt = mysql_insert(CmdbResourceType).values( id=str(uuid.uuid4()), provider=provider, resource_type=definition.resource_type, service=definition.service, category=definition.category, display_name=definition.display_name, icon_key=definition.icon_key, sensitivity=definition.sensitivity, last_seen_at=datetime.now(UTC), discovered_dynamically=definition.discovered_dynamically, ).on_duplicate_key_update( service=definition.service, category=definition.category, display_name=definition.display_name, icon_key=definition.icon_key, sensitivity=definition.sensitivity, active=True, last_seen_at=datetime.now(UTC), discovered_dynamically=definition.discovered_dynamically, ) await db.execute(stmt) row = await db.scalar(select(CmdbResourceType).where( CmdbResourceType.provider == provider, CmdbResourceType.resource_type == definition.resource_type, )) assert row is not None return row _SPECIALIZED_INVENTORY_TYPES: dict[str, frozenset[str]] = { "aws": frozenset({ "AWS::EC2::Instance", "AWS::S3::Bucket", "AWS::IAM::User", "AWS::ElasticLoadBalancing::LoadBalancer", "AWS::ElasticLoadBalancingV2::LoadBalancer", "AWS::CloudFront::Distribution", "AWS::ACM::Certificate", }), "azure": frozenset({ "microsoft.compute/virtualmachines", "microsoft.storage/storageaccounts", "microsoft.network/loadbalancers", "microsoft.network/applicationgateways", "microsoft.network/dnszones", "microsoft.network/privatednszones", }), "gcp": frozenset({ "compute.googleapis.com/Instance", "storage.googleapis.com/Bucket", "dns.googleapis.com/ManagedZone", }), } def _inventory_source_id(integration_id: str, item: CloudResourceInventoryItem) -> str: return str(uuid.uuid5( uuid.NAMESPACE_URL, f"helix:{integration_id}:{item.resource_type}:{item.provider_resource_id}", )) def _dynamic_definition(item: CloudResourceInventoryItem) -> TypeDefinition: display = item.resource_type.rsplit("/", 1)[-1].split("::")[-1].replace("_", " ") return TypeDefinition( resource_type=item.resource_type, service=item.service or "other", category=item.category or "other", display_name=display or item.resource_type, icon_key=item.category or "cloud", sensitivity=item.sensitivity, discovered_dynamically=True, ) async def refresh_provider_inventory( db: AsyncSession, integration: Integration, items: list[CloudResourceInventoryItem], *, authoritative: bool, ) -> dict[str, int]: """Upsert one broad inventory snapshot with a completeness sweep fence.""" existing_rows = (await db.scalars(select(CmdbCloudResource).where( CmdbCloudResource.integration_id == str(integration.id), CmdbCloudResource.source_table == "provider_inventory", ))).all() existing = {row.source_id: row for row in existing_rows} seen_source_ids: set[str] = set() type_cache: dict[str, CmdbResourceType] = {} summary = {"seen": 0, "created": 0, "changed": 0, "disappeared": 0, "restored": 0} skipped_types = { value.casefold() for value in _SPECIALIZED_INVENTORY_TYPES.get(integration.provider, frozenset()) } unique_items: dict[str, CloudResourceInventoryItem] = {} for item in items: if ( not item.provider_resource_id or not item.resource_type or len(item.resource_type) > 192 ): continue if item.resource_type.casefold() in skipped_types: continue source_id = _inventory_source_id(str(integration.id), item) unique_items[source_id] = item now = datetime.now(UTC) for source_id, item in unique_items.items(): seen_source_ids.add(source_id) definition = _dynamic_definition(item) type_row = type_cache.get(definition.resource_type) if type_row is None: type_row = await _resource_type(db, integration.provider, definition) type_cache[definition.resource_type] = type_row projection = Projection( definition=definition, provider_resource_id=item.provider_resource_id, name=item.name or item.provider_resource_id, source_table="provider_inventory", source_id=source_id, scope_external_id=item.scope_external_id, region=item.region, zone=item.zone, status=item.status, environment=None, owner_person_id=None, tags=item.tags, normalized={"inventory_source": "provider_catalog"}, raw=item.metadata, source_sync_job_id=None, first_seen_at=now, last_seen_at=now, deleted_at=None, ) snapshot = canonical_snapshot(projection) digest = snapshot_hash(snapshot) row = existing.get(source_id) if row is None: row = CmdbCloudResource( id=uuid.uuid4(), integration_id=str(integration.id), resource_type_id=str(type_row.id), provider=integration.provider, service=definition.service, category=definition.category, resource_type=definition.resource_type, provider_resource_id=item.provider_resource_id, resource_identity_hash=resource_identity_hash(definition.resource_type, item.provider_resource_id), name=projection.name[:512], scope_external_id=item.scope_external_id, region=item.region, zone=item.zone, lifecycle_status=item.status, sensitivity=item.sensitivity, tags=snapshot["tags"], normalized_metadata=snapshot["normalized"], raw_metadata=snapshot["raw"], content_hash=digest, source_table="provider_inventory", source_id=source_id, first_seen_at=now, last_seen_at=now, last_changed_at=now, ) db.add(row) await db.flush() db.add(CmdbCloudResourceChange( resource_id=str(row.id), change_type="discovered", changed_fields=changed_fields(None, snapshot), current_hash=digest, snapshot=snapshot, observed_at=now, )) summary["created"] += 1 else: before = { "provider_resource_id": row.provider_resource_id, "name": row.name, "scope_external_id": row.scope_external_id, "region": row.region, "zone": row.zone, "status": row.lifecycle_status, "environment": row.environment, "owner_person_id": row.owner_person_id, "tags": row.tags or {}, "normalized": row.normalized_metadata or {}, "raw": row.raw_metadata or {}, "deleted": row.deleted_at is not None, } change_type = "restored" if row.deleted_at is not None else ("changed" if row.content_hash != digest else None) if change_type: db.add(CmdbCloudResourceChange( resource_id=str(row.id), change_type=change_type, changed_fields=changed_fields(before, snapshot), previous_hash=row.content_hash, current_hash=digest, snapshot=snapshot, observed_at=now, )) summary[change_type] += 1 row.last_changed_at = now row.resource_type_id = str(type_row.id) row.service = definition.service row.category = definition.category row.name = projection.name[:512] row.scope_external_id = item.scope_external_id row.region = item.region row.zone = item.zone row.lifecycle_status = item.status row.sensitivity = item.sensitivity row.tags = snapshot["tags"] row.normalized_metadata = snapshot["normalized"] row.raw_metadata = snapshot["raw"] row.content_hash = digest row.last_seen_at = now row.missing_since = None row.deleted_at = None summary["seen"] += 1 if authoritative: for source_id, row in existing.items(): if source_id in seen_source_ids or row.deleted_at is not None: continue row.missing_since = now row.deleted_at = now row.last_changed_at = now before_hash = row.content_hash snapshot = { "provider_resource_id": row.provider_resource_id, "name": row.name, "scope_external_id": row.scope_external_id, "region": row.region, "zone": row.zone, "status": row.lifecycle_status, "environment": row.environment, "owner_person_id": row.owner_person_id, "tags": row.tags or {}, "normalized": row.normalized_metadata or {}, "raw": row.raw_metadata or {}, "deleted": True, } digest = snapshot_hash(snapshot) row.content_hash = digest db.add(CmdbCloudResourceChange( resource_id=str(row.id), change_type="disappeared", changed_fields=["deleted"], previous_hash=before_hash, current_hash=digest, snapshot=snapshot, observed_at=now, )) summary["disappeared"] += 1 await db.commit() return summary async def refresh_integration( db: AsyncSession, integration: Integration, *, batch_size: int = _DEFAULT_REFRESH_BATCH_SIZE, ) -> dict[str, int]: """Project one integration using bounded, resumable keyset pages. Source and directory rows both carry sizeable JSON metadata. The first production backfill kept a whole integration in one transaction; TiDB eventually cancelled a change-ledger ``INSERT`` with error 8176 as those JSON writes accumulated. The old unbounded reads also inflated client memory. The narrow ID page, bounded wide-row fetch, and per-page commit below cap both query and transaction memory, while content hashes preserve safe idempotent recovery after an interrupted run. """ batch_size = max(1, min(int(batch_size), _MAX_REFRESH_BATCH_SIZE)) integration_id = str(integration.id) provider = integration.provider now = datetime.now(UTC) type_cache: dict[str, str] = {} summary = {"seen": 0, "created": 0, "changed": 0, "disappeared": 0, "restored": 0} active_scopes = list((await db.scalars(select(CmdbCloudScope).where( CmdbCloudScope.integration_id == integration_id, CmdbCloudScope.status == "active", CmdbCloudScope.deleted_at.is_(None), ))).all()) connector_scope = _canonical_connector_scope(provider, active_scopes) for model, adapter in _projection_specs(): cursor: Any = None while True: id_stmt = select(model.id).where(model.integration_id == integration_id) if cursor is not None: id_stmt = id_stmt.where(model.id > cursor) source_ids = list((await db.scalars( id_stmt.order_by(model.id).limit(batch_size), )).all()) if not source_ids: break sources: list[Any] = list((await db.scalars( select(model).where(model.id.in_(source_ids)).order_by(model.id), )).all()) projections = [adapter(source) for source in sources] if connector_scope: projections = [ projection if projection.scope_external_id else replace(projection, scope_external_id=connector_scope) for projection in projections ] projection_ids = [projection.source_id for projection in projections] directory_rows = list((await db.scalars(select(CmdbCloudResource).where( CmdbCloudResource.source_table == model.__tablename__, CmdbCloudResource.source_id.in_(projection_ids), ))).all()) existing = {row.source_id: row for row in directory_rows} for projection in projections: definition = projection.definition type_id = type_cache.get(definition.resource_type) if type_id is None: type_row = await _resource_type(db, provider, definition) type_id = str(type_row.id) type_cache[definition.resource_type] = type_id snapshot = canonical_snapshot(projection) digest = snapshot_hash(snapshot) row = existing.get(projection.source_id) observed_at = projection.last_seen_at or now is_missing = projection.deleted_at is not None if row is None: row = CmdbCloudResource( id=uuid.uuid4(), integration_id=integration_id, resource_type_id=type_id, provider=provider, service=definition.service, category=definition.category, resource_type=definition.resource_type, provider_resource_id=projection.provider_resource_id, resource_identity_hash=resource_identity_hash( definition.resource_type, projection.provider_resource_id, ), name=projection.name[:512], scope_external_id=projection.scope_external_id, region=projection.region, zone=projection.zone, lifecycle_status=projection.status, environment=projection.environment, owner_person_id=projection.owner_person_id, sensitivity=definition.sensitivity, tags=snapshot["tags"], normalized_metadata=snapshot["normalized"], raw_metadata=snapshot["raw"], content_hash=digest, source_table=projection.source_table, source_id=projection.source_id, source_sync_job_id=projection.source_sync_job_id, first_seen_at=projection.first_seen_at or observed_at, last_seen_at=observed_at, last_changed_at=now, missing_since=projection.deleted_at, deleted_at=projection.deleted_at, ) db.add(row) await db.flush() db.add(CmdbCloudResourceChange( resource_id=str(row.id), change_type="discovered", changed_fields=changed_fields(None, snapshot), current_hash=digest, snapshot=snapshot, source_sync_job_id=projection.source_sync_job_id, observed_at=now, )) summary["created"] += 1 else: previous_snapshot = { "provider_resource_id": row.provider_resource_id, "name": row.name, "scope_external_id": row.scope_external_id, "region": row.region, "zone": row.zone, "status": row.lifecycle_status, "environment": row.environment, "owner_person_id": row.owner_person_id, "tags": row.tags or {}, "normalized": row.normalized_metadata or {}, "raw": row.raw_metadata or {}, "deleted": row.deleted_at is not None, } change_type: str | None = None if row.deleted_at is None and is_missing: change_type = "disappeared" summary["disappeared"] += 1 elif row.deleted_at is not None and not is_missing: change_type = "restored" summary["restored"] += 1 elif row.content_hash != digest: change_type = "changed" summary["changed"] += 1 if change_type: db.add(CmdbCloudResourceChange( resource_id=str(row.id), change_type=change_type, changed_fields=changed_fields(previous_snapshot, snapshot), previous_hash=row.content_hash, current_hash=digest, snapshot=snapshot, source_sync_job_id=projection.source_sync_job_id, observed_at=now, )) row.last_changed_at = now row.resource_type_id = type_id row.provider = provider row.service = definition.service row.category = definition.category row.resource_type = definition.resource_type row.provider_resource_id = projection.provider_resource_id row.resource_identity_hash = resource_identity_hash( definition.resource_type, projection.provider_resource_id, ) row.name = projection.name[:512] row.scope_external_id = projection.scope_external_id row.region = projection.region row.zone = projection.zone row.lifecycle_status = projection.status row.environment = projection.environment row.owner_person_id = projection.owner_person_id row.sensitivity = definition.sensitivity row.tags = snapshot["tags"] row.normalized_metadata = snapshot["normalized"] row.raw_metadata = snapshot["raw"] row.content_hash = digest row.source_sync_job_id = projection.source_sync_job_id row.last_seen_at = observed_at row.missing_since = projection.deleted_at row.deleted_at = projection.deleted_at summary["seen"] += 1 await db.commit() cursor = source_ids[-1] return summary async def refresh_inventories(db: AsyncSession) -> dict[str, int]: totals = { "integrations": 0, "seen": 0, "created": 0, "changed": 0, "disappeared": 0, "restored": 0, "inventory_skipped": 0, "inventory_partial": 0, "inventory_errors": 0, } integrations = (await db.scalars(select(Integration).where( Integration.kind == "cloud", Integration.enabled.is_(True), ))).all() for integration in integrations: totals["integrations"] += 1 provider = None try: provider, _ = await build_cloud_provider(db, str(integration.id)) inventory = await asyncio.wait_for( provider.list_resource_inventory(), timeout=15 * 60, ) partial = provider.consume_partial_failures() skipped = provider.consume_skipped_legs() if partial or skipped: totals["inventory_partial"] += 1 inventory_result = await refresh_provider_inventory( db, integration, inventory, authoritative=not partial and not skipped, ) for key, value in inventory_result.items(): totals[key] += value except CapabilityNotSupported: totals["inventory_skipped"] += 1 except Exception as exc: # noqa: BLE001 - isolate integrations totals["inventory_errors"] += 1 logger.warning( "Broad resource inventory failed integration=%s provider=%s error=%s", integration.id, integration.provider, type(exc).__name__, ) await db.rollback() finally: close = getattr(provider, "close", None) if close is not None: try: await close() except Exception: # noqa: BLE001 - cleanup cannot block projection logger.warning( "Cloud inventory provider cleanup failed integration=%s", integration.id, ) return totals async def refresh_all( db: AsyncSession, *, history_retention_days: int = 730, batch_size: int = _DEFAULT_REFRESH_BATCH_SIZE, ) -> dict[str, int]: totals = { "integrations": 0, "seen": 0, "created": 0, "changed": 0, "disappeared": 0, "restored": 0, "history_pruned": 0, } integrations = (await db.scalars(select(Integration).where(Integration.kind == "cloud"))).all() for integration in integrations: refresh_result = await refresh_integration(db, integration, batch_size=batch_size) totals["integrations"] += 1 for key, value in refresh_result.items(): totals[key] += value cutoff = datetime.now(UTC) - timedelta(days=max(30, min(history_retention_days, 3650))) bounded_batch_size = max(1, min(int(batch_size), _MAX_REFRESH_BATCH_SIZE)) while True: expired_ids = list((await db.scalars( select(CmdbCloudResourceChange.id) .where(CmdbCloudResourceChange.observed_at < cutoff) .order_by(CmdbCloudResourceChange.observed_at, CmdbCloudResourceChange.id) .limit(bounded_batch_size), )).all()) if not expired_ids: break prune_result = await db.execute( delete(CmdbCloudResourceChange).where(CmdbCloudResourceChange.id.in_(expired_ids)), ) totals["history_pruned"] += int(getattr(prune_result, "rowcount", 0) or 0) await db.commit() return totals