"""Shared IAM-user upsert used by autogrant flows.

The CMDB IAM-sync job (``server/app/domains/cmdb/sync.py::_sync_iam_users``)
writes a row in ``cmdb_iam_users`` for every IAM/RAM user it discovers
in a cloud tenant. Without this helper, an autogrant that mints a fresh
RAM user has to wait for the next scheduled IAM sync (default cadence:
hourly) before the user shows up on the ``/cloud/iam-users`` page —
admins reported the row "missing" when checking immediately after
approval.

This helper performs the same upsert the sync job does, scoped to a
single freshly-created IAM/RAM user. Same dedupe key
``(integration_id, user_name, sub_account_id)``, same MySQL
``ON DUPLICATE KEY UPDATE`` semantics, so a subsequent sync pass
collapses cleanly onto this row instead of creating a duplicate.

Provenance note: callers tag rows with ``__source=helix_autogrant.<x>``
in the ``tags`` dict. That tag is meaningful in the window between
autogrant landing and the next IAM-sync pass — sync overwrites the
``tags`` and ``raw_data`` columns from the cloud-side response, so
``__source`` is a transient signal, not a durable provenance column.
For permanent provenance the autogrant flow already writes the
relationship into ``api_key_assignments.iam_integration_id`` +
``iam_user_id`` (joined by sync-time backfill) and into
``detail.bailian.granted_at`` / ``detail.bedrock.granted_at`` JSON.

Caller commits — this matches the autogrant convention where the
caller already commits ``key_value``/status/detail in one transaction
right after the IAM upsert.
"""
from __future__ import annotations

import uuid
from datetime import UTC, datetime
from typing import Any

from sqlalchemy import select
from sqlalchemy.dialects.mysql import insert as mysql_insert
from sqlalchemy.ext.asyncio import AsyncSession

from app.core.exceptions import AppException
from app.domains.cmdb.models import CmdbIamUser

# Sentinel emitted by ``AlicloudIntlProvider._stamp_account_id`` for users
# that live on the umbrella (master) account rather than an RD-member
# sub-account. The provider keeps this label in ``raw_data.__sub_account_id``
# for observability ("which leg of the fan-out produced this row?") but
# the indexed ``cmdb_iam_users.sub_account_id`` column normalizes it to
# ``''`` — the master account has no Aliyun-side numeric account id we
# could store, and conflating it with the empty-string convention used
# by single-account providers (AWS, Tencent, plain Aliyun) keeps the
# unique key consistent across providers.
_MASTER_ACCOUNT_SENTINEL = "__master__"


def normalize_iam_sub_account_id(raw: object) -> str:
    """Map a provider-stamped ``__sub_account_id`` tag value to the
    canonical ``cmdb_iam_users.sub_account_id`` column value.

    Rules:
      * ``None`` / empty string / the ``__master__`` sentinel → ``''``
        (the column convention for "no sub-account scope").
      * Any other value is coerced to ``str`` and truncated to the
        column's 64-char width.

    Centralized here so the upsert path in ``sync.py``, the autogrant
    fast-path in this module, and any future helper share one mapping.
    Migration 120 originally backfilled the column without applying
    this normalization, which caused a 79-cycle IAM-sync failure on
    AlicloudIntl prod (see migration 157 for the data fix).
    """
    if raw is None or raw == "" or raw == _MASTER_ACCOUNT_SENTINEL:
        return ""
    return str(raw)[:64]


async def upsert_local_iam_user(
    db: AsyncSession,
    *,
    integration_id: str,
    user_name: str,
    arn: str | None,
    user_id_external: str | None,
    create_date: datetime | None = None,
    attached_policies: list[Any] | None = None,
    groups: list[Any] | None = None,
    tags: dict[str, Any] | None = None,
    raw_data: dict[str, Any] | None = None,
    owner_person_id: str | None = None,
    sub_account_id: str = "",
) -> CmdbIamUser:
    """Insert (or update) a row in ``cmdb_iam_users`` and return it.

    Mirrors the upsert semantics of ``_sync_iam_users`` so an autogrant
    that just minted a RAM user makes the row visible immediately on
    ``/cloud/iam-users``. The next scheduled IAM sync is idempotent
    against this row (same UQ key, same column writes).

    ``mfa_enabled`` is hard-coded to ``False`` because RAM users created
    via OpenAPI start without MFA; sync will flip it once the user
    enrols a virtual device.

    Caller is responsible for committing the surrounding transaction.
    """
    now = datetime.now(UTC)

    # Asymmetric-tombstone guard (Round-44, paired with PR #196's
    # scope-side fix in ``feedback_on_duplicate_key_preserve_consistency``).
    # If a previously-revoked-then-soft-deleted IAM/RAM row exists for
    # this ``(integration_id, user_name, sub_account_id)`` triple, an
    # autogrant must NOT silently flip ``deleted_at`` back to NULL while
    # leaving ``revoked_at`` set — the row half-revives ("active"
    # everywhere except in the revoked tab), and the requester walks
    # away with cloud credentials Helix still labels as revoked. Loud
    # refuse on the autogrant path; the sync path retains its
    # 5-minute revival window (sync.py::_sync_iam_users).
    existing_revoked = (
        await db.execute(
            select(CmdbIamUser).where(
                CmdbIamUser.integration_id == integration_id,
                CmdbIamUser.user_name == user_name,
                CmdbIamUser.sub_account_id == sub_account_id,
                CmdbIamUser.revoked_at.isnot(None),
            )
        )
    ).scalar_one_or_none()
    if existing_revoked is not None:
        raise AppException(
            code="IAM_USER_PREVIOUSLY_REVOKED",
            message=(
                f"RAM/IAM user {user_name!r} on integration "
                f"{integration_id} was previously revoked at "
                f"{existing_revoked.revoked_at}. Autogrant refuses "
                "to re-mint credentials onto a tombstoned row — IT "
                "admin must clear the revoke (or pick a different "
                "user_name) before retry."
            ),
            status_code=409,
        )

    # Round-22 audit: when the autogrant call provides only the bare
    # RAM user metadata (no policies / groups / tags resolved yet),
    # ``attached_policies`` / ``groups`` / ``tags`` / ``raw_data``
    # come in as ``None`` / ``[]`` / ``{}``. Previously the
    # on_duplicate_key_update wrote these empty values back over a
    # potentially freshly-synced row that the IAM sync just populated
    # with real policy data — clobbering it back to empty until the
    # next IAM sync cycle. We now ONLY include the descriptive
    # columns in the ON DUPLICATE KEY UPDATE clause when the caller
    # actually provided a value; the "owner_person_id" / "last_synced
    # _at" / "deleted_at" columns are always updated since those
    # ARE the autogrant's authoritative writes.
    update_clauses: dict[str, Any] = {
        "last_synced_at": now,
        "deleted_at": None,
    }
    # Round-22 review-2: Bedrock / Bailian autogrant call sites pass
    # ``owner_person_id=None`` when the underlying assignment row
    # carries no person (e.g. a delegated service account or a not-
    # yet-resolved owner). The earlier round-22 fix unconditionally
    # wrote owner_person_id into the UPDATE clause; that would clobber
    # an existing row's owner with NULL on every autogrant retry. Only
    # update when the caller provided a real owner; otherwise let the
    # IAM-sync path remain the authoritative writer for that column.
    if owner_person_id is not None:
        update_clauses["owner_person_id"] = owner_person_id
    if arn is not None:
        update_clauses["arn"] = arn
    if user_id_external is not None:
        update_clauses["user_id_external"] = user_id_external
    if create_date is not None:
        update_clauses["create_date"] = create_date
    if attached_policies:  # non-empty list
        update_clauses["attached_policies"] = attached_policies
    if groups:
        update_clauses["groups"] = groups
    if tags:
        update_clauses["tags"] = tags
    if raw_data:
        update_clauses["raw_data"] = raw_data

    stmt = mysql_insert(CmdbIamUser).values(
        id=str(uuid.uuid4()),
        integration_id=integration_id,
        user_name=user_name,
        sub_account_id=sub_account_id,
        arn=arn,
        user_id_external=user_id_external,
        create_date=create_date,
        attached_policies=attached_policies or [],
        groups=groups or [],
        mfa_enabled=False,
        tags=tags or {},
        owner_person_id=owner_person_id,
        raw_data=raw_data or {},
        last_synced_at=now,
    ).on_duplicate_key_update(**update_clauses)
    await db.execute(stmt)
    # Re-fetch by the unique key so we return a managed instance the
    # caller can use to populate FKs (e.g. ``assignment.iam_user_id``).
    row = (
        await db.execute(
            select(CmdbIamUser).where(
                CmdbIamUser.integration_id == integration_id,
                CmdbIamUser.user_name == user_name,
                CmdbIamUser.sub_account_id == sub_account_id,
            )
        )
    ).scalar_one()
    return row


__all__ = ["normalize_iam_sub_account_id", "upsert_local_iam_user"]