"""Refresh + read the persisted seat usage signals (PR4). Write side: :func:`refresh_usage_signals` walks every active software connector whose provider has a collector, reads the vendor roster once per integration, matches each observation to a Helix person by work email, and upserts one row per ``(person, provider, product_key)``. Read side: :func:`signals_for_people` hands the reclaim job and the admin page an already-freshness-resolved view, so no caller can accidentally read a stale row's raw ``state``. Two rules the write side follows, both learned the expensive way in this repo: * **A vendor that could not be read leaves its rows alone.** Overwriting them with a fresh ``unknown`` would destroy real evidence and make the page look answered; the freshness gate already retires them on its own schedule. The failure is reported instead. * **Per-provider isolation.** One vendor being down must not stop the others; failures are collected and re-raised at the end so the scheduled-job framework still marks the run failed. """ from __future__ import annotations import logging import uuid from datetime import UTC, datetime from typing import Any from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.domains.asset.models import SeatUsageSignal from app.identity.person_models import IdentityPerson from app.integrations.usage_signals import ( SIGNAL_UNSUPPORTED, resolve_signal, usage_collectors, ) logger = logging.getLogger(__name__) async def _active_software_integrations(db: AsyncSession) -> list[Any]: """Integrations that back at least one ACTIVE software connector. Scoped to connectors rather than to all integrations: an integration nobody has bound to a catalog entry grants no seats, so collecting its roster would be a vendor call with no consumer. """ from app.domains.provisioning.models import SoftwareConnector from app.integrations.models import Integration rows = (await db.execute( select(Integration) .join(SoftwareConnector, SoftwareConnector.integration_id == Integration.id) .where( Integration.enabled.is_(True), SoftwareConnector.is_active.is_(True), ) .distinct() )).scalars().all() return list(rows) async def _email_to_person_id(db: AsyncSession) -> dict[str, uuid.UUID]: """Work email (lowercased) → person id, for match-up. Leavers are included: a departed person still holding a licensed seat is precisely what the idle report should show, and dropping them here would silently narrow the signal to current staff. """ rows = (await db.execute( select(IdentityPerson.id, IdentityPerson.work_email) .where(IdentityPerson.deleted_at.is_(None)) )).all() return { (email or "").strip().lower(): pid for pid, email in rows if (email or "").strip() } async def refresh_usage_signals( db: AsyncSession, *, now: datetime | None = None, ) -> dict[str, Any]: """Collect from every collectable provider and upsert the store.""" now = now or datetime.now(UTC) collectors = usage_collectors() by_email = await _email_to_person_id(db) summary: dict[str, Any] = { "providers": 0, "observations": 0, "rows_written": 0, "unmatched_emails": 0, "per_provider": [], "errors": [], } for integration in await _active_software_integrations(db): collector = collectors.get(integration.provider) if collector is None: # Declared unsupported (or unknown to this build) — not an # error. The coverage test is what makes sure a NEW provider # cannot land here silently. continue try: driver = await _load_driver(db, integration) except Exception as exc: # noqa: BLE001 summary["errors"].append( f"{integration.provider}: driver load failed: {exc}", ) continue result = await collector(db, driver) if result.error: # Rows deliberately untouched — see the module docstring. summary["errors"].append(f"{integration.provider}: {result.error}") summary["per_provider"].append({ "provider": integration.provider, "status": "failed", "error": result.error, }) continue written = 0 unmatched = 0 for obs in result.observations: person_id = by_email.get(obs.person_email) if person_id is None: unmatched += 1 continue await _upsert( db, person_id=person_id, provider=integration.provider, product_key=obs.product_key, state=obs.state, last_active_at=obs.last_active_at, evidence=obs.evidence, collected_at=now, ) written += 1 await db.flush() summary["providers"] += 1 summary["observations"] += len(result.observations) summary["rows_written"] += written summary["unmatched_emails"] += unmatched summary["per_provider"].append({ "provider": integration.provider, "status": "ok", "observations": len(result.observations), "rows": written, "unmatched": unmatched, }) return summary async def _load_driver(db: AsyncSession, integration: Any) -> Any: from app.integrations.registry import get_driver_class from app.integrations.service import _resolve cls = get_driver_class(integration.kind, integration.provider) if cls is None: raise RuntimeError( f"no driver registered for {integration.kind}/{integration.provider}", ) resolved = await _resolve(db, integration) return cls(integration_id=str(integration.id), config=resolved.config) async def _upsert( db: AsyncSession, *, person_id: uuid.UUID, provider: str, product_key: str, state: str, last_active_at: datetime | None, evidence: dict[str, Any] | None, collected_at: datetime, ) -> None: row = (await db.execute( select(SeatUsageSignal).where( SeatUsageSignal.person_id == person_id, SeatUsageSignal.provider == provider, SeatUsageSignal.product_key == product_key, ) )).scalars().first() if row is None: db.add(SeatUsageSignal( person_id=person_id, provider=provider, product_key=product_key, state=state, last_active_at=last_active_at, evidence=evidence or {}, collected_at=collected_at, )) return row.state = state row.last_active_at = last_active_at row.evidence = evidence or {} row.collected_at = collected_at async def signals_for_people( db: AsyncSession, person_ids: list[uuid.UUID], *, provider: str | None = None, now: datetime | None = None, ) -> dict[tuple[uuid.UUID, str, str], dict[str, Any]]: """``(person, provider, product) → resolved signal``. The returned ``state`` has already passed through the freshness gate, so callers cannot read a stale row's raw claim by accident. The raw stored value is kept alongside as ``stored_state`` for the UI, which legitimately wants to say "it said active, but that was 4 days ago". """ now = now or datetime.now(UTC) if not person_ids: return {} query = select(SeatUsageSignal).where( SeatUsageSignal.person_id.in_(person_ids), ) if provider: query = query.where(SeatUsageSignal.provider == provider) out: dict[tuple[uuid.UUID, str, str], dict[str, Any]] = {} for row in (await db.execute(query)).scalars().all(): out[(row.person_id, row.provider, row.product_key)] = { "state": resolve_signal(row.state, row.collected_at, now=now), "stored_state": row.state, "last_active_at": row.last_active_at, "collected_at": row.collected_at, "evidence": row.evidence or {}, } return out def unsupported_signal(provider: str) -> dict[str, Any]: """The read-side shape for a provider that cannot answer at all.""" from app.integrations.usage_signals import NO_USAGE_SIGNAL return { "state": SIGNAL_UNSUPPORTED, "stored_state": SIGNAL_UNSUPPORTED, "last_active_at": None, "collected_at": None, "evidence": {"reason": NO_USAGE_SIGNAL.get(provider, "")}, } __all__ = [ "refresh_usage_signals", "signals_for_people", "unsupported_signal", ]