"""Resource-creator attribution service — cache + cloud audit-log
resolution for "who created this resource?".

Public surface:

* :func:`get_cached_creator(db, integration_id, resource_id)` — fast
  read of the cached attribution row (no audit-API call). Used by
  the dashboard's resource drawer + scope-billing-detail to render
  the "Created by" field immediately on data load without blocking.

* :func:`resolve_resource_creator(db, integration_id, resource_id)`
  — on-demand resolve: returns the cached row when fresh, otherwise
  calls the driver's :meth:`get_resource_creator`, persists the
  result, and returns it. Used by the dashboard's "Resolve" action
  (user opens drawer → frontend fires this) and the cron sweep.

* :func:`sweep_unresolved(db)` — cron entrypoint. Walks
  ``cmdb_cost_facts`` for newly-discovered resources that don't
  have a ``cmdb_resource_creators`` row yet, plus existing rows in
  ``status='error'`` that are due for retry. Bounded by attempt
  cap and throttled per-integration to stay under the audit APIs'
  rate limits.

Design choices:

* **Persistent cache** — a resolution is never re-attempted unless
  the row is in a transient ``error`` state. Terminal statuses
  (``resolved`` / ``not_found`` / ``window_exceeded`` /
  ``unsupported``) are durable.

* **Person matching** — when the driver returns
  ``ResourceCreatorInfo.email``, we look up
  ``identity_persons.work_email`` (case-insensitive, lowercase
  compare per Helix convention) and stamp ``creator_person_id`` on
  the cache row. A non-match leaves ``creator_person_id=NULL`` but
  ``creator_email`` populated — the operator can manually map
  later.

* **No fanout on failure** — when the driver raises a transient
  exception (network blip, rate-limit), we record ``status='error'``
  + bump ``attempt_count``; cron retries up to
  :data:`RESOURCE_CREATOR_MAX_ATTEMPTS` (5). Beyond that the row
  stays in ``error`` and is excluded from sweep until an operator
  explicitly resets it via the manual-retry endpoint.

* **JSON ``raw`` cap** — drivers already cap their raw payload at
  ~16 KiB; this layer accepts the dict verbatim. The TiDB JSON
  column type can hold up to 64 MiB but per-row growth is the
  expensive dimension.
"""
from __future__ import annotations

import re as _re
from datetime import UTC, datetime, timedelta
from typing import Any

from sqlalchemy import and_, func, or_, select
from sqlalchemy.dialects.mysql import insert as mysql_insert
from sqlalchemy.ext.asyncio import AsyncSession

from app.core.safe_logging import get_safe_logger
from app.domains.cmdb.models import (
    RESOURCE_CREATOR_MAX_ATTEMPTS,
    RESOURCE_CREATOR_SOURCE_ALIYUN_ACTIONTRAIL,
    RESOURCE_CREATOR_SOURCE_AZURE_ACTIVITY,
    RESOURCE_CREATOR_SOURCE_CLOUDTRAIL,
    RESOURCE_CREATOR_SOURCE_GCP_AUDIT,
    RESOURCE_CREATOR_STATUS_ERROR,
    RESOURCE_CREATOR_STATUS_NOT_FOUND,
    RESOURCE_CREATOR_STATUS_RESOLVED,
    RESOURCE_CREATOR_STATUS_UNSUPPORTED,
    RESOURCE_CREATOR_STATUS_WINDOW_EXCEEDED,
    RESOURCE_CREATOR_TERMINAL_STATUSES,
    CmdbCostFact,
    CmdbResourceCreator,
)
from app.identity.person_models import IdentityPerson
from app.integrations.driver_base import (
    CloudDriver,
    ResourceCreatorInfo,
    ResourceCreatorWindowExceeded,
)
from app.integrations.models import Integration
from app.integrations.service import build_driver_by_id

logger = get_safe_logger(__name__)


# Patterns we redact from driver-exception strings before persisting
# to ``cmdb_resource_creators.last_error``. The full ``last_error``
# blob is returned in the GET / POST / MCP creator responses to any
# user passing ``can_view_finance OR can_read_cloud`` — finance,
# management, auditor, cloud_admin, global_admin. Driver exceptions
# from boto3 / googleapiclient / Aliyun / Azure SDK can embed:
#
#   * AWS pre-signed URL params (``Signature=`` / ``X-Amz-Signature``)
#   * OAuth bearer tokens (``Bearer eyJ…``)
#   * AWS access-key IDs in canonical-request error messages
#   * Service-account email / refresh-token tails in HttpError body
#
# We scrub the obvious ones inline; the rest stays as type-name +
# truncated message which is enough for "what kind of failure" but
# not enough to grant the reader an unintended capability if the
# row leaks elsewhere (export, screenshot, copy-paste).
_LAST_ERROR_REDACT_PATTERNS = [
    # AWS pre-signed URL query params (any order). Match key=value
    # up to the next ``&`` / end-of-string / quote.
    _re.compile(
        r"(?i)\b(Signature|X-Amz-Signature|X-Amz-Credential|"
        r"X-Amz-Security-Token|AWSAccessKeyId|access_token|"
        r"refresh_token|id_token|client_secret)=[^&\s\"']+",
    ),
    # OAuth / JWT Bearer tokens (eyJ… common Auth0/Google/Azure shape).
    _re.compile(r"(?i)\bBearer\s+[A-Za-z0-9_\-\.=]{12,}"),
    # Lone JWT in any context (no Bearer prefix).
    _re.compile(r"\beyJ[A-Za-z0-9_\-\.=]{20,}"),
    # Google OAuth access tokens (ya29.…).
    _re.compile(r"\bya29\.[A-Za-z0-9_\-\.]+"),
    # Generic "<thing>: <looks-like-secret>" — long base64-ish blobs
    # right after ``token``/``secret``/``key``/``password`` keywords.
    # Allow ``.`` and ``-`` inside the value run so Google + JWT shapes
    # don't terminate the match early.
    _re.compile(
        r"(?i)\b(token|secret|password|api[_-]?key)\b[^A-Za-z0-9]+"
        r"[A-Za-z0-9_\-+/=\.]{16,}",
    ),
    # AWS access-key ID (long-lived or session). Pattern is
    # ``[ASRK]KIA`` + 16 [A-Z0-9]. Standalone occurrences in
    # ``InvalidAccessKeyId`` / ``SignatureDoesNotMatch`` error
    # bodies must scrub.
    _re.compile(r"\b(?:AKIA|ASIA|AROA|AIDA|ANPA)[0-9A-Z]{16}\b"),
    # Aliyun access-key ID (long-term + STS shapes). Real shape is
    # ``LTAI`` + ~16-20 base62 chars.
    _re.compile(r"\bLTAI[0-9A-Za-z]{12,20}\b"),
    # AWS ARN with explicit 12-digit account id — IAM principal +
    # account leakage from boto3 ``AccessDenied`` messages. We
    # redact the account-id segment only; the resource path stays
    # so operators can still triage which service errored.
    _re.compile(
        r"(arn:aws[a-z0-9\-]*:[a-z0-9\-]*:[a-z0-9\-]*:)\d{12}(:)",
    ),
]


def _redact_one(m: _re.Match[str]) -> str:
    """Per-match replacement used by ``_sanitize_last_error``.

    Three shapes:
      * ``key=value``    → ``key=[REDACTED]``  (preserve `=` separator)
      * ``key value``    → ``key [REDACTED]``  (preserve space separator)
      * lone token       → ``[REDACTED]``      (no separator to keep)

    Special-cases:
      * AWS ARN with account id → keep ARN scaffolding, redact
        the 12-digit account number only. ARN-shape patterns
        capture two groups; the substitution reuses them.
    """
    s = m.group(0)
    # ARN account-id redaction (group-substitution shape).
    if m.lastindex and m.re.pattern.startswith("(arn:aws"):
        # Pattern has 2 groups: ``arn:aws…:`` prefix and ``:``
        # suffix; account-id segment between them is what we hide.
        return f"{m.group(1)}[REDACTED-ACCOUNT-ID]{m.group(2)}"
    if "=" in s:
        key = s.split("=", 1)[0]
        return f"{key}=[REDACTED]"
    if " " in s:
        # ``Bearer eyJ…`` / ``Authorization: Bearer eyJ…`` shape —
        # keep the keyword + space + REDACTED so the audit row
        # still says "this WAS a Bearer token" not the misleading
        # ``Bearer=[REDACTED]``.
        key = s.split(" ", 1)[0]
        return f"{key} [REDACTED]"
    return "[REDACTED]"


def _sanitize_last_error(exc: BaseException) -> str:
    """Return a redacted ``type(exc).__name__: str(exc)`` suitable for
    persisting to ``cmdb_resource_creators.last_error``.

    Always returns a non-empty string capped at 1000 chars (matches
    the column's effective consumer cap — the column itself is
    ``Text`` so the cap is for response-payload size + audit-row size
    discipline, not column constraint). Sensitive substrings are
    replaced with the literal ``[REDACTED]`` (or
    ``[REDACTED-ACCOUNT-ID]`` for AWS account-id segments).

    Covers:

      * Pre-signed URL params (Signature / X-Amz-* / access tokens)
      * Bearer tokens (with or without ``Authorization:`` prefix)
      * Lone JWT (eyJ…) blobs
      * Google OAuth (ya29.…) tokens
      * AWS access-key IDs (AKIA / ASIA / AROA / AIDA / ANPA)
      * Aliyun access-key IDs (LTAI…)
      * AWS account IDs inside ARN strings (``:123456789012:``)
      * Generic ``token=`` / ``secret:`` / ``api_key:`` + long blob
    """
    raw = f"{type(exc).__name__}: {exc}"
    for pat in _LAST_ERROR_REDACT_PATTERNS:
        raw = pat.sub(_redact_one, raw)
    return raw[:1000]


# Map driver ``provider`` slug → ``source`` enum value. Used to
# stamp the cache row consistently even when a driver evolves its
# audit-API choice (e.g. AWS adds a long-term S3 trail option later
# — source is still ``cloudtrail`` from the consumer's perspective).
_PROVIDER_TO_SOURCE = {
    "aws_cloud": RESOURCE_CREATOR_SOURCE_CLOUDTRAIL,
    "gcp_cloud": RESOURCE_CREATOR_SOURCE_GCP_AUDIT,
    "aliyun_cloud": RESOURCE_CREATOR_SOURCE_ALIYUN_ACTIONTRAIL,
    "aliyun_intl_cloud": RESOURCE_CREATOR_SOURCE_ALIYUN_ACTIONTRAIL,
    "azure_cloud": RESOURCE_CREATOR_SOURCE_AZURE_ACTIVITY,
}


def _source_for_provider(provider: str) -> str:
    """Map a driver ``provider`` slug to the audit-log source label.

    Unknown providers default to the driver's own slug so the cache
    row is still queryable; the cron sweep will mark such rows as
    ``unsupported`` since the driver-base default raises
    NotImplementedError.
    """
    return _PROVIDER_TO_SOURCE.get(provider, provider)


async def _match_person_by_email(
    db: AsyncSession, email: str | None,
) -> str | None:
    """Look up an identity_persons row by lowercase work_email.

    Returns the person id (uuid string) on match or ``None``. The
    audit log's principal email is the canonical key — matching by
    display name is unreliable across renames + duplicates.
    """
    if not email:
        return None
    normalized = email.strip().lower()
    if not normalized or "@" not in normalized:
        return None
    res = await db.execute(
        select(IdentityPerson.id)
        .where(func.lower(IdentityPerson.work_email) == normalized)
        .limit(1),
    )
    person_id = res.scalar_one_or_none()
    return str(person_id) if person_id else None


async def get_cached_creator(
    db: AsyncSession,
    integration_id: str,
    resource_id: str,
) -> CmdbResourceCreator | None:
    """Read the cached attribution row, or None if no row exists."""
    res = await db.execute(
        select(CmdbResourceCreator).where(
            and_(
                CmdbResourceCreator.integration_id == integration_id,
                CmdbResourceCreator.resource_id == resource_id,
            ),
        ),
    )
    return res.scalar_one_or_none()


async def _upsert_creator_row(
    db: AsyncSession,
    *,
    integration_id: str,
    resource_id: str,
    source: str,
    status: str,
    info: ResourceCreatorInfo | None = None,
    person_id: str | None = None,
    last_error: str | None = None,
    increment_attempt: bool = True,
) -> CmdbResourceCreator:
    """Idempotent insert-or-update of a ``cmdb_resource_creators``
    row.

    Uses MySQL's ``INSERT ... ON DUPLICATE KEY UPDATE`` keyed by the
    unique ``(integration_id, resource_id)`` constraint so two
    concurrent resolves don't race-create duplicate rows.
    """
    now = datetime.now(UTC)
    base_values: dict[str, Any] = {
        "integration_id": integration_id,
        "resource_id": resource_id,
        "source": source,
        "status": status,
        "last_attempt_at": now,
        "last_error": last_error,
        "attempt_count": 1 if increment_attempt else 0,
    }
    if info is not None:
        base_values.update({
            "creator_principal_id": info.principal_id,
            "creator_email": info.email,
            "creator_display_name": info.display_name,
            "created_at_cloud": info.created_at_cloud,
            "resolved_at": now if status == RESOURCE_CREATOR_STATUS_RESOLVED else None,
            "raw": info.raw or None,
        })
    if person_id is not None:
        base_values["creator_person_id"] = person_id

    stmt = mysql_insert(CmdbResourceCreator).values(**base_values)
    # ON DUPLICATE UPDATE: increment attempt_count, refresh status +
    # last_attempt_at, overwrite resolved fields when present.
    update_values: dict[str, Any] = {
        "status": stmt.inserted.status,
        "source": stmt.inserted.source,
        "last_attempt_at": stmt.inserted.last_attempt_at,
        "last_error": stmt.inserted.last_error,
    }
    if increment_attempt:
        update_values["attempt_count"] = (
            CmdbResourceCreator.attempt_count + 1
        )
    if info is not None:
        update_values.update({
            "creator_principal_id": stmt.inserted.creator_principal_id,
            "creator_email": stmt.inserted.creator_email,
            "creator_display_name": stmt.inserted.creator_display_name,
            "created_at_cloud": stmt.inserted.created_at_cloud,
            "resolved_at": stmt.inserted.resolved_at,
            "raw": stmt.inserted.raw,
        })
    if person_id is not None:
        update_values["creator_person_id"] = stmt.inserted.creator_person_id
    stmt = stmt.on_duplicate_key_update(**update_values)
    await db.execute(stmt)
    await db.flush()
    # ON DUPLICATE KEY UPDATE bypasses the identity-map invariant —
    # ORM cache may still hold the pre-update row. Expire the
    # cached entry by deleting from the identity map so the
    # subsequent SELECT re-fetches from the DB and we return a row
    # with the post-update ``attempt_count`` + ``status``.
    existing = await get_cached_creator(db, integration_id, resource_id)
    if existing is not None:
        await db.refresh(existing)
    return existing if existing is not None else (
        # Fall-through shouldn't be reachable — the upsert just
        # inserted a row keyed on (integration_id, resource_id).
        await _upsert_creator_row(  # pragma: no cover
            db,
            integration_id=integration_id,
            resource_id=resource_id,
            source=source, status=status,
            info=info, person_id=person_id,
            last_error=last_error,
            increment_attempt=False,
        )
    )


# Minimum gap between two ``force=True`` resolves for the same
# (integration_id, resource_id). Prevents a hot-mashed "Re-resolve"
# button from draining the audit-API rate budget. The reviewer
# flagged this as a follow-up; we land it inline. 60s is enough
# for a human user to retry after seeing a result; cron's regular
# (non-force) calls are unaffected because they fall through the
# cache-hit path on terminal statuses.
FORCE_REFRESH_MIN_GAP_SECONDS = 60


async def resolve_resource_creator(
    db: AsyncSession,
    integration_id: str,
    resource_id: str,
    *,
    force: bool = False,
) -> CmdbResourceCreator:
    """Return the (possibly cached) creator attribution for one
    resource.

    Flow:
      1. Read cache. If terminal status and not ``force``, return.
         If ``force`` and the last attempt was within the
         :data:`FORCE_REFRESH_MIN_GAP_SECONDS` window, downgrade
         to a cache read (mash-click protection).
      2. Build the integration's driver.
      3. Call :meth:`CloudDriver.get_resource_creator`.
      4. Persist the result; map email → identity_persons.

    Exceptions during the driver call are caught and recorded as
    ``status='error'`` so the cron sweep can retry up to
    :data:`RESOURCE_CREATOR_MAX_ATTEMPTS`. Re-raises only when the
    integration itself can't be loaded (404 / 503) — those are
    operator-actionable errors, not transient.
    """
    cached = await get_cached_creator(db, integration_id, resource_id)
    if (
        cached is not None
        and not force
        and cached.status in RESOURCE_CREATOR_TERMINAL_STATUSES
    ):
        return cached
    # ``force=True`` re-attempt cooldown — protects the audit-API
    # rate budget against mashed "Re-resolve" buttons. Only
    # downgrades when there IS a cached row (first-ever resolve
    # always proceeds).
    if (
        force
        and cached is not None
        and cached.last_attempt_at is not None
    ):
        # Per `mysql_datetime_naive_on_fresh_load` memory: MySQL
        # DATETIME(tz=True) round-trips as a tz-NAIVE datetime on
        # fresh load. Coerce to UTC-aware before the subtraction
        # to avoid a TypeError when this branch hits on a freshly-
        # loaded row.
        last_at = cached.last_attempt_at
        if last_at.tzinfo is None:
            last_at = last_at.replace(tzinfo=UTC)
        # ``elapsed`` can be slightly negative when MySQL's
        # DATETIME storage truncates microseconds + applies its
        # session timezone, returning a ``last_attempt_at`` that's
        # a few hundred ms ahead of Python's ``now()``. Clamp to
        # zero so the cooldown predicate doesn't suppress a
        # genuine retry on a fresh row that was created milliseconds
        # ago AND whose timestamp appears "in the future".
        elapsed = max(
            0.0,
            (datetime.now(UTC) - last_at).total_seconds(),
        )
        if elapsed < FORCE_REFRESH_MIN_GAP_SECONDS:
            return cached

    # Load the integration row to discover the provider slug, then
    # build the driver.
    integ_res = await db.execute(
        select(Integration).where(Integration.id == integration_id),
    )
    integration = integ_res.scalar_one_or_none()
    if integration is None:
        # No integration → caller error; persist nothing.
        raise LookupError(
            f"Integration {integration_id!r} not found",
        )
    source = _source_for_provider(integration.provider or "")

    driver: CloudDriver
    try:
        driver = await build_driver_by_id(  # type: ignore[assignment]
            integration_id, session=db,
        )
    except Exception as exc:
        # Integration disabled / driver class missing — surface
        # via cache row so the cron sweep stops retrying.
        return await _upsert_creator_row(
            db,
            integration_id=integration_id,
            resource_id=resource_id,
            source=source,
            status=RESOURCE_CREATOR_STATUS_ERROR,
            last_error=f"build_driver: {_sanitize_last_error(exc)}",
        )

    if not isinstance(driver, CloudDriver):
        # Non-cloud integration passed through — this is a caller
        # error but we record it for visibility.
        return await _upsert_creator_row(
            db,
            integration_id=integration_id,
            resource_id=resource_id,
            source=source,
            status=RESOURCE_CREATOR_STATUS_UNSUPPORTED,
            last_error="integration is not a CloudDriver",
        )

    try:
        info = await driver.get_resource_creator(resource_id)
    except NotImplementedError:
        return await _upsert_creator_row(
            db,
            integration_id=integration_id,
            resource_id=resource_id,
            source=source,
            status=RESOURCE_CREATOR_STATUS_UNSUPPORTED,
            last_error=(
                f"driver {integration.provider!r} does not implement "
                "get_resource_creator"
            ),
        )
    except ResourceCreatorWindowExceeded as exc:
        return await _upsert_creator_row(
            db,
            integration_id=integration_id,
            resource_id=resource_id,
            source=source,
            status=RESOURCE_CREATOR_STATUS_WINDOW_EXCEEDED,
            last_error=_sanitize_last_error(exc),
        )
    except Exception as exc:  # noqa: BLE001
        logger.exception(
            "resource-creator resolve failed integration=%s resource=%s",
            integration_id, resource_id,
        )
        return await _upsert_creator_row(
            db,
            integration_id=integration_id,
            resource_id=resource_id,
            source=source,
            status=RESOURCE_CREATOR_STATUS_ERROR,
            # Sanitize before persist — boto3 / googleapiclient /
            # Aliyun / Azure exception strings can carry pre-signed
            # URL params / bearer tokens / access-key IDs that we
            # MUST NOT serve back to RBAC-passing readers (finance /
            # cloud-admin / management / auditor). See
            # ``_sanitize_last_error`` docstring. The unredacted
            # exception is in the application log via ``logger.
            # exception`` above (admin-only via log access).
            last_error=_sanitize_last_error(exc),
        )
    finally:
        # Drivers may have buffered SDK clients; release them.
        try:
            await driver.close()
        except Exception:  # noqa: BLE001
            pass

    if info is None:
        # Audit log returned no matching create event in the window —
        # terminal "not_found" so the cron sweep stops retrying.
        return await _upsert_creator_row(
            db,
            integration_id=integration_id,
            resource_id=resource_id,
            source=source,
            status=RESOURCE_CREATOR_STATUS_NOT_FOUND,
        )

    # Success — match email → Person and persist.
    person_id = await _match_person_by_email(db, info.email)
    return await _upsert_creator_row(
        db,
        integration_id=integration_id,
        resource_id=resource_id,
        source=source,
        status=RESOURCE_CREATOR_STATUS_RESOLVED,
        info=info,
        person_id=person_id,
    )


def serialize_resource_creator_row(row: CmdbResourceCreator) -> dict[str, Any]:
    """Single source of truth for the ``cmdb_resource_creators`` row →
    API JSON shape.

    Used by:
      * ``GET /api/v1/cmdb/cost/resources/{int}/{rid}/creator``
      * ``POST /api/v1/cmdb/cost/resources/{int}/{rid}/creator/retry``
      * MCP tool ``cost_resolve_resource_creator``
      * MCP tool ``cost_retry_resource_creator``

    Lifted out of ``cost_router.py`` (2026-05-26) so the MCP tools
    stop drifting from the HTTP twin — prior to this both MCP tools
    omitted ``last_attempt_at`` while the HTTP path included it.
    The schema is now mechanically identical across all four
    surfaces; adding a future field lands everywhere.
    """
    return {
        "integration_id": str(row.integration_id),
        "resource_id": row.resource_id,
        "status": row.status,
        "source": row.source,
        "creator": {
            "principal_id": row.creator_principal_id,
            "email": row.creator_email,
            "display_name": row.creator_display_name,
            "person_id": (
                str(row.creator_person_id)
                if row.creator_person_id else None
            ),
        } if row.status == "resolved" else None,
        "created_at_cloud": (
            row.created_at_cloud.isoformat()
            if row.created_at_cloud else None
        ),
        "resolved_at": (
            row.resolved_at.isoformat()
            if row.resolved_at else None
        ),
        "attempt_count": row.attempt_count,
        "last_attempt_at": (
            row.last_attempt_at.isoformat()
            if row.last_attempt_at else None
        ),
        "last_error": row.last_error,
    }


async def reset_and_resolve_resource_creator(
    db: AsyncSession,
    integration_id: str,
    resource_id: str,
) -> CmdbResourceCreator:
    """Operator escape hatch — zero ``attempt_count`` + clear the
    force-refresh cooldown, then call
    :func:`resolve_resource_creator` with ``force=True``.

    Why this is its own function rather than a flag on the
    ``GET /creator?refresh=true`` path: a row that has hit
    :data:`RESOURCE_CREATOR_MAX_ATTEMPTS` is excluded from the cron
    sweep's candidate pool (see :func:`list_unresolved_resources`),
    so even after a successful manual ``force=True`` resolve the row
    stays at ``attempt_count >= MAX`` and the cron will never pick it
    up again on its own. The operator needs an explicit "reset the
    counter so the cron can keep trying" action; bundling that into
    the refresh path would silently re-enable retries for every
    refresh click, eroding the cap's purpose (protect audit-API rate
    budget against perma-broken provider configs).

    Idempotent — safe to call on terminal-status rows too (resolved /
    not_found / window_exceeded / unsupported); the subsequent
    ``force=True`` resolve will refresh against the live audit log.

    Returns the post-resolve row.
    """
    # SQL update is keyed by (integration_id, resource_id); a missing
    # row is a no-op here and ``resolve_resource_creator`` will
    # create one on first attempt. Clearing ``last_attempt_at`` also
    # neutralises the :data:`FORCE_REFRESH_MIN_GAP_SECONDS` cooldown
    # so the resolve actually fires.
    #
    # We also zero ``creator_*`` columns so a prior resolved row's
    # PII (email / display_name / person_id) doesn't survive the
    # reset → driver-failed-retry trajectory. Without this clear,
    # state transitions {resolved → reset → fail → error} leave
    # the DB row at ``status='error', creator_email=<old@>'``: the
    # serializer correctly masks (creator: null on non-resolved),
    # but the at-rest PII residue outlives a GDPR-erasure on the
    # creator's Person row (``identity_persons.work_email`` purge
    # has no FK cascade to this cache). Treating reset as "back to
    # unknown" closes that boundary; if the driver succeeds on the
    # subsequent ``resolve_resource_creator(force=True)`` call,
    # creator_* gets written fresh from the audit-log response.
    from sqlalchemy import update as _update
    await db.execute(
        _update(CmdbResourceCreator)
        .where(
            and_(
                CmdbResourceCreator.integration_id == integration_id,
                CmdbResourceCreator.resource_id == resource_id,
            ),
        )
        .values(
            attempt_count=0,
            last_attempt_at=None,
            last_error=None,
            creator_principal_id=None,
            creator_email=None,
            creator_display_name=None,
            creator_person_id=None,
            created_at_cloud=None,
            resolved_at=None,
            raw=None,
        ),
    )
    await db.flush()
    return await resolve_resource_creator(
        db,
        integration_id=integration_id,
        resource_id=resource_id,
        force=True,
    )


# ── Cron sweep ──────────────────────────────────────────────────


# Candidate freshness window. Resources whose most-recent
# ``cmdb_cost_facts.date`` is older than this are excluded from the
# sweep:
#
# * AWS CloudTrail / Aliyun ActionTrail / Azure Activity Log all
#   default to a 90-day audit-event retention. A resource that hasn't
#   billed in 90+ days is past every provider's lookback window, so
#   the audit call WILL return :class:`ResourceCreatorWindowExceeded`
#   / empty. Including those in the candidate pool just burns
#   provider rate budget for a guaranteed terminal "won't attribute"
#   result.
# * GCP Cloud Audit Logs retain 400 days, but the cap exists to keep
#   the candidate-pool size predictable across all four clouds. The
#   small attribution loss on 90-400-day-old GCP resources is the
#   trade — sub-1% of the pool in practice (cost_facts churn is
#   dominated by hot resources).
#
# Without this filter, prod's ``cmdb_cost_facts`` (~277k distinct
# rids accumulated since 2025-10) would feed the sweep ~277k
# candidates and the daily-200-tick budget would take ~690 days to
# drain. The 90-day filter caps the candidate pool at ~38k initially
# and ~150/day at steady state, keeping the sweep useful instead of
# decorative.
RESOURCE_CREATOR_CANDIDATE_WINDOW_DAYS = 90


async def list_unresolved_resources(
    db: AsyncSession,
    *,
    integration_id: str | None = None,
    limit: int = 200,
    candidate_window_days: int = RESOURCE_CREATOR_CANDIDATE_WINDOW_DAYS,
) -> list[tuple[str, str]]:
    """Return (integration_id, resource_id) pairs to resolve next.

    Candidates are resources seen in ``cmdb_cost_facts`` within the
    last ``candidate_window_days`` (default
    :data:`RESOURCE_CREATOR_CANDIDATE_WINDOW_DAYS`) that either (a)
    have no ``cmdb_resource_creators`` row yet or (b) have a row in
    ``status='error'`` with ``attempt_count <``
    :data:`RESOURCE_CREATOR_MAX_ATTEMPTS`.

    The freshness filter matches the shortest provider audit-log
    retention window (90d on AWS / Aliyun / Azure). Resources that
    haven't billed inside it are past every provider's lookback and
    would just burn rate budget for a guaranteed window-exceeded
    response — see the constant's docstring above.

    Result is bounded by ``limit`` to keep one cron tick under the
    audit-API rate budget. Default cadence (2/day × 200 = 400/day)
    is sized for the steady-state new-resource discovery rate
    (~150 distinct rids/day across all four cloud integrations).
    """
    # New resources: distinct resource_id from cost_facts that
    # don't have a creator row. The MAX(date) HAVING clause is the
    # freshness filter — see the module-level constant docstring for
    # why we cap at 90 days.
    cutoff_date = (datetime.now(UTC) - timedelta(
        days=int(candidate_window_days),
    )).date()
    new_resources_q = (
        select(
            CmdbCostFact.integration_id,
            CmdbCostFact.resource_id,
        )
        .where(
            and_(
                CmdbCostFact.resource_id.is_not(None),
                # Skip synthetic ingest-side ids — drivers will
                # short-circuit them but we save the round-trip.
                ~CmdbCostFact.resource_id.like("aws-account:%"),
                ~CmdbCostFact.resource_id.like("aws-principal:%"),
                ~CmdbCostFact.resource_id.like("aliyun-account:%"),
                ~CmdbCostFact.resource_id.like("aliyun-bailian:%"),
                ~CmdbCostFact.resource_id.like("azure-aggregated:%"),
                ~CmdbCostFact.resource_id.like("azure-subscription:%"),
                ~CmdbCostFact.resource_id.like("gcp-aggregated:%"),
                # Freshness: only resources with a cost_fact row
                # in the audit-log lookback window. Older rids are
                # past every provider's retention and would just
                # burn rate budget for a guaranteed terminal.
                CmdbCostFact.date >= cutoff_date,
            ),
        )
        .group_by(
            CmdbCostFact.integration_id,
            CmdbCostFact.resource_id,
        )
        .limit(limit)
    )
    if integration_id is not None:
        new_resources_q = new_resources_q.where(
            CmdbCostFact.integration_id == integration_id,
        )

    res = await db.execute(new_resources_q)
    candidates: list[tuple[str, str]] = [
        (str(r.integration_id), r.resource_id)
        for r in res.all()
    ]

    # Filter out those that already have a terminal creator row.
    # Bulk lookup keyed on (integration_id, resource_id) — TiDB
    # handles this via the unique index.
    if not candidates:
        return []

    existing_q = await db.execute(
        select(
            CmdbResourceCreator.integration_id,
            CmdbResourceCreator.resource_id,
            CmdbResourceCreator.status,
            CmdbResourceCreator.attempt_count,
        )
        .where(
            or_(*[
                and_(
                    CmdbResourceCreator.integration_id == i,
                    CmdbResourceCreator.resource_id == r,
                )
                for i, r in candidates
            ]),
        ),
    )
    existing_map: dict[tuple[str, str], tuple[str, int]] = {
        (str(row.integration_id), row.resource_id): (
            row.status, int(row.attempt_count or 0),
        )
        for row in existing_q
    }

    out: list[tuple[str, str]] = []
    for integ, rid in candidates:
        existing = existing_map.get((integ, rid))
        if existing is None:
            out.append((integ, rid))
            continue
        status, attempts = existing
        if status in RESOURCE_CREATOR_TERMINAL_STATUSES:
            continue
        if (
            status == RESOURCE_CREATOR_STATUS_ERROR
            and attempts < RESOURCE_CREATOR_MAX_ATTEMPTS
        ):
            out.append((integ, rid))
    return out


__all__ = [
    "get_cached_creator",
    "list_unresolved_resources",
    "reset_and_resolve_resource_creator",
    "resolve_resource_creator",
    "serialize_resource_creator_row",
]