"""Tencent Cloud SDK call throttle + RequestLimitExceeded retry shim.

Tencent Cloud APIs enforce a *per-account-per-endpoint* QPS ceiling that
defaults to **5 RPS** for CAM (the API-key list / sub-account family) —
``ListAttachedUserAllPolicies``, ``ListUsers``, ``ListAccessKeys``,
``AttachUserPolicy``, ``DetachUserPolicy``, ``DeleteUser`` etc. all
share the same per-tenant token bucket on the cloud side. Hitting it
returns ``RequestLimitExceeded`` (Tencent error code) and looks like::

    [TencentCloudSDKException] code:RequestLimitExceeded
    message:您当前每秒请求 6 次，超过了每秒频率上限 5

The CAM grants-sync flow iterates every CAM user and issues one
``ListAttachedUserAllPolicies`` call per user. With more than 5 active
sub-accounts the loop trivially exceeds the ceiling and the whole
``cmdb.grants_sync`` job aborts mid-snapshot — which is the exact bug
the public scheduled-jobs page surfaced on 2026-05-09. Any partial
snapshot is also a *mass-revoke* hazard: ``grants_sync`` interprets a
missing grant on a successfully-synced scope as "the cloud removed it
→ close the Helix row" (see ``lifecycle_service:_apply_discovered_grants``
mark-and-sweep). A throttle failure mid-loop must NOT silently degrade.

This module exposes one primitive — :func:`tencent_call` — that:

  1. Acquires a per-credential token from a process-wide
     :class:`asyncio.Semaphore`-style bucket. The bucket is keyed by
     ``(secret_id, service)`` so two integrations with different keys
     don't share one quota, and the CAM bucket and the billing bucket
     stay independent (Tencent meters them separately).
  2. Calls the Tencent SDK method on a worker thread (the SDK is
     synchronous; matches every existing ``asyncio.to_thread`` call).
  3. On ``RequestLimitExceeded`` retries with exponential backoff +
     jitter, capped at :data:`MAX_RETRIES` attempts. Final attempt
     re-raises so the caller surfaces a real error instead of a 0-row
     snapshot.

The default rate (5 RPS) matches Tencent's documented CAM limit. We
leave headroom (4.5 RPS effective) so admin-triggered manual syncs
running concurrently with the cron don't trip the ceiling. If a
specific service has a different documented rate, callers pass an
override via the ``rate`` keyword.

Concurrency model:

  * The limiter is **process-local**. Helix runs one ``app.worker``
    process plus the FastAPI server; ``cmdb.grants_sync`` runs on the
    worker exclusively (see ``server/app/worker.py``), so the worker's
    token bucket is the authoritative source. The FastAPI process's
    bucket is a thin defence for admin-triggered "Sync now" on the
    /cloud-accounts page: at worst a manual click and the cron fire on
    the same second and we burn a couple of retries — far cheaper than
    cross-process coordination over Redis.
  * Single-flight at the *job* level is still enforced upstream by the
    ``cloud_sync:<integration_id>`` distributed lock (lifecycle_router /
    scheduled_jobs handlers). The throttle here is the per-call
    second-by-second governor inside that lock.

Tested by ``server/tests/test_integrations/test_tencent_throttle.py``.
"""
from __future__ import annotations

import asyncio
import logging
import random
import threading
import time
from collections.abc import Callable
from typing import Any

logger = logging.getLogger(__name__)

# ── Defaults ────────────────────────────────────────────────────────
# 5 requests/second is the documented CAM ceiling reported by the
# error message text ("超过了每秒频率上限 5"). We default to 4.5 to
# leave headroom for an admin-triggered concurrent sync; callers can
# override via ``rate=`` if a specific service has a different cap.
DEFAULT_RPS: float = 4.5

# Each service Tencent meters separately gets its own bucket key.
# Keep this conservative — picking the lowest-known ceiling is safer
# than guessing per-API limits the docs don't publish.
SERVICE_RATES: dict[str, float] = {
    "cam": 4.5,        # CAM (sub-account / policy / access-key APIs)
    "billing": 4.5,    # DescribeBillDetail / DescribeBillSummary*
    "cvm": 18.0,       # DescribeInstances — docs cap at 20 RPS
    "clb": 18.0,
    "cdn": 18.0,
    "dnspod": 18.0,
    "domain": 4.5,     # 域名注册 — undocumented, treat as conservative CAM-tier
}

# Hard cap on retries to bound cron runtime. With the throttle in place
# we should rarely retry at all; retries here are a backstop for the
# rare race with a parallel admin click, not the primary defence.
MAX_RETRIES: int = 5

# Backoff base (seconds). With exponential growth (2^attempt) plus
# jitter, the first retry waits ~1.0s ± 0.3s — long enough for the
# Tencent token bucket to refill 5 tokens at the documented rate.
BACKOFF_BASE: float = 1.0
BACKOFF_CAP: float = 8.0


# ── Per-bucket state ────────────────────────────────────────────────

class _TokenBucket:
    """Asyncio-friendly token bucket.

    A simple "earliest next allowed time" scheduler. Each acquire
    advances ``_next`` by ``1/rate``; if that's in the future the
    coroutine sleeps until then. Concurrent acquires serialize through
    ``_lock`` so two coroutines can't both advance ``_next`` past each
    other.
    """

    __slots__ = ("_rate", "_min_interval", "_next", "_lock")

    def __init__(self, rate: float) -> None:
        # Defensive: ``rate <= 0`` would divide-by-zero in the next
        # line and silently make the bucket a no-op (worst-of-both
        # outcomes — quietly drops the throttle without any signal).
        # Treat sub-1 RPS as a config error and surface it.
        if rate <= 0:
            raise ValueError(f"_TokenBucket rate must be > 0, got {rate}")
        self._rate = rate
        self._min_interval = 1.0 / rate
        self._next = 0.0
        self._lock = asyncio.Lock()

    async def acquire(self) -> None:
        async with self._lock:
            now = time.monotonic()
            wait = self._next - now
            if wait > 0:
                # Lock is HELD during ``asyncio.sleep`` so concurrent
                # coroutines queue strictly fifo and each picks up the
                # updated ``_next`` after the prior one schedules itself.
                # With N waiters at rate r, total wait is ~N/r — same as
                # a release-then-recompute design but with deterministic
                # ordering. Acceptable at 4.5 RPS (each waiter waits at
                # most ~0.22s while holding the lock).
                await asyncio.sleep(wait)
                now = time.monotonic()
            # Schedule the next allowed instant. Using max() here so a
            # bucket that's been idle doesn't accumulate "credit" — at
            # most one extra burst slot, never an unbounded reservoir.
            self._next = max(now, self._next) + self._min_interval


_buckets: dict[tuple[str, str], _TokenBucket] = {}
# Guards ``_buckets`` lookup/insert. asyncio is single-threaded, so a
# plain dict.setdefault would suffice — but the lock keeps the bucket
# instantiation atomic across `await` points if we ever add async work
# inside the factory.
_buckets_lock = asyncio.Lock()


def _bucket_key(secret_id: str, service: str) -> tuple[str, str]:
    # Empty / falsy secret_id collapses to ``"<unset>"`` so a
    # mis-configured driver still gets a bucket and can't accidentally
    # share with a real tenant's quota.
    return (secret_id or "<unset>", service or "default")


async def _get_bucket(secret_id: str, service: str, rate: float) -> _TokenBucket:
    key = _bucket_key(secret_id, service)
    async with _buckets_lock:
        bucket = _buckets.get(key)
        if bucket is None:
            bucket = _TokenBucket(rate=rate)
            _buckets[key] = bucket
        return bucket


# Test-only hook — clears the process-wide bucket map. Production code
# never needs this; the buckets are immortal and that's intentional
# (the rate carries across calls within the worker process lifetime).
def _reset_buckets_for_tests() -> None:
    _buckets.clear()


# ── Retry classification ────────────────────────────────────────────

# Tencent-side codes for "the server is sad, retry should work" — added
# Round-30 (2026-05-24) parity with Azure / GCP / UCloud 5xx retry. Pre-
# fix a single ``InternalError`` mid-pagination on ``DescribeInstances``
# stopped pagination, recorded a partial_failure, and flipped the
# servers cron red — even though the very next sync would have
# recovered. Documented Tencent codes that indicate a transient
# server fault (NOT a permanent client error, NOT auth):
#   * ``InternalError`` — generic server fault bucket
#   * ``InternalError.*`` — service-specific variants
#     (``InternalError.GetInstanceListError``, etc.)
#   * ``ServiceUnavailable`` — backend overloaded
#   * ``ServiceUnavailable.*``
# These join ``RequestLimitExceeded`` (rate-limit) in the retry pool so
# the existing throttle + backoff plumbing covers both classes.
_TENCENT_TRANSIENT_CODE_PREFIXES = (
    "RequestLimitExceeded",
    "InternalError",
    "ServiceUnavailable",
)


def _is_request_limit_error(exc: BaseException) -> bool:
    """Match TencentCloudSDKException whose ``code`` indicates a
    retryable transient (rate-limit OR server fault).

    Despite the name (kept for backward compat with imports), this
    now covers BOTH rate-limit AND transient server-fault classes —
    they share the same backoff treatment per the bucket-and-retry
    contract documented above ``_TENCENT_TRANSIENT_CODE_PREFIXES``.

    Tencent's documented codes:
      * ``RequestLimitExceeded`` (and ``.GlobalRegionUinLimitExceeded`` /
        ``.IPLimitExceeded`` / ``.UinLimitExceeded`` suffixes) —
        "you're going too fast"
      * ``InternalError`` (and service-suffix variants) — server fault
      * ``ServiceUnavailable`` (and suffix variants) — backend overload

    The SDK exposes the structured code on the ``.code`` attribute, but
    we also fall back to substring matching on ``str(exc)`` because
    older SDK versions surface the code only inside the message.
    """
    code = ""
    try:
        code = str(getattr(exc, "code", "") or "")
    except Exception:  # noqa: BLE001 — defensive against odd SDK shapes
        code = ""
    for prefix in _TENCENT_TRANSIENT_CODE_PREFIXES:
        if code.startswith(prefix):
            return True
    # Fallback for older SDK versions that surface the code only in the
    # message. Anchor on ``code:<prefix>`` (the SDK's canonical
    # ``[TencentCloudSDKException] code:X message:Y`` format) so a
    # wrapper exception that merely *quotes* the literal string inside
    # its message — e.g. an upstream ``RuntimeError("seen
    # RequestLimitExceeded earlier")`` — does NOT trigger a retry.
    exc_str = str(exc)
    for prefix in _TENCENT_TRANSIENT_CODE_PREFIXES:
        if f"code:{prefix}" in exc_str:
            return True
    return False


# ── Public entrypoint ──────────────────────────────────────────────

async def tencent_call[T](
    fn: Callable[..., T],
    *args: Any,
    secret_id: str,
    service: str = "cam",
    rate: float | None = None,
    max_retries: int = MAX_RETRIES,
    **kwargs: Any,
) -> T:
    """Run a Tencent SDK call under throttle + transient-error retry.

    The Tencent SDK is synchronous, so this awaits ``asyncio.to_thread``
    around ``fn(*args, **kwargs)`` after the per-credential bucket
    grants a token.

    Retries cover both rate-limit (``RequestLimitExceeded.*``) AND
    server-side transient (``InternalError.*``, ``ServiceUnavailable.*``)
    classes — see :data:`_TENCENT_TRANSIENT_CODE_PREFIXES`. Auth
    errors / not-found / structural client errors propagate immediately
    so the caller's classification branches still fire.

    Args:
      fn: The bound SDK method, e.g. ``client.ListUsers``. Anything
        that takes the request object and returns the response.
      args / kwargs: Forwarded verbatim to ``fn``.
      secret_id: Tenant credential identifier — distinct credentials
        have distinct buckets so customers don't share quotas.
      service: Tencent service short name (``cam``, ``billing``, …).
        Buckets are per ``(secret_id, service)``.
      rate: Optional override for the per-second cap. Defaults to
        :data:`SERVICE_RATES` lookup, falling back to
        :data:`DEFAULT_RPS`.
      max_retries: Cap on transient-error retries (covers both
        rate-limit and server-fault classes).

    Raises:
      The original SDK exception on the final attempt (retries
      exhausted), or any non-transient exception immediately so the
      caller's auth / not-found branches still fire.
    """
    effective_rate = rate if rate is not None else SERVICE_RATES.get(service, DEFAULT_RPS)
    bucket = await _get_bucket(secret_id, service, effective_rate)

    last_exc: Exception | None = None
    for attempt in range(max_retries + 1):
        await bucket.acquire()
        try:
            return await asyncio.to_thread(fn, *args, **kwargs)
        except Exception as exc:  # noqa: BLE001 — re-raise after classify
            # Note: ``asyncio.CancelledError`` inherits from BaseException
            # (Python 3.8+), so catching ``Exception`` here intentionally
            # lets cancellation propagate without being mis-classified.
            last_exc = exc
            if not _is_request_limit_error(exc):
                # Non-transient error — propagate immediately so the
                # caller's ``UserNotExist`` / ``AuthFailure`` etc.
                # handling fires unchanged.
                raise
            if attempt >= max_retries:
                logger.warning(
                    "Tencent %s call exhausted %d retries on transient: %s",
                    service, max_retries, exc,
                )
                raise
            # Exponential backoff + jitter. Jitter avoids the
            # thundering-herd shape where every retry from concurrent
            # coroutines wakes on the same second and re-trips the cap.
            sleep_for = min(BACKOFF_BASE * (2 ** attempt), BACKOFF_CAP)
            sleep_for += random.uniform(0, sleep_for * 0.3)
            logger.info(
                "Tencent %s transient (attempt %d/%d) — backoff %.2fs: %s",
                service, attempt + 1, max_retries, sleep_for,
                getattr(exc, "code", type(exc).__name__),
            )
            await asyncio.sleep(sleep_for)
    # Defensive: loop should always either return or raise above.
    if last_exc is not None:
        raise last_exc
    raise RuntimeError("tencent_call exited loop without return or raise — unreachable")


# ── Sync helper for code paths already inside ``asyncio.to_thread`` ──

def tencent_call_sync[T](
    fn: Callable[..., T],
    *args: Any,
    secret_id: str,
    service: str = "cam",
    rate: float | None = None,
    max_retries: int = MAX_RETRIES,
    **kwargs: Any,
) -> T:
    """Sync flavour of :func:`tencent_call`.

    Used from drivers that already wrap the whole flow in a single
    ``asyncio.to_thread`` call (e.g. ``_revoke_account_sync`` does
    several SDK calls in one thread). Re-entering ``asyncio.to_thread``
    from a worker thread is harmless but wasteful — the sync helper
    enforces the same throttle+retry contract using a per-credential
    threading lock and ``time.sleep``.
    """
    effective_rate = rate if rate is not None else SERVICE_RATES.get(service, DEFAULT_RPS)
    if effective_rate <= 0:
        raise ValueError(f"effective_rate must be > 0, got {effective_rate}")
    min_interval = 1.0 / effective_rate
    key = _bucket_key(secret_id, service)
    state = _get_sync_state(key)
    last_exc: Exception | None = None
    for attempt in range(max_retries + 1):
        with state.lock:
            now = time.monotonic()
            wait = state.next_at - now
            if wait > 0:
                # Hold the lock during the sleep — the next caller
                # picks up the post-wait schedule. This serializes the
                # throttle but the wait is bounded by 1/rate (~0.22s
                # at 4.5 RPS), so contention is negligible.
                time.sleep(wait)
                now = time.monotonic()
            state.next_at = max(now, state.next_at) + min_interval
        try:
            return fn(*args, **kwargs)
        except Exception as exc:  # noqa: BLE001 — re-raise after classify
            last_exc = exc
            if not _is_request_limit_error(exc):
                raise
            if attempt >= max_retries:
                logger.warning(
                    "Tencent %s sync call exhausted %d retries on transient: %s",
                    service, max_retries, exc,
                )
                raise
            sleep_for = min(BACKOFF_BASE * (2 ** attempt), BACKOFF_CAP)
            sleep_for += random.uniform(0, sleep_for * 0.3)
            logger.info(
                "Tencent %s sync transient (attempt %d/%d) — backoff %.2fs: %s",
                service, attempt + 1, max_retries, sleep_for,
                getattr(exc, "code", type(exc).__name__),
            )
            time.sleep(sleep_for)
    if last_exc is not None:
        raise last_exc
    raise RuntimeError("tencent_call_sync exited loop without return or raise — unreachable")


class _SyncBucketState:
    __slots__ = ("lock", "next_at")

    def __init__(self) -> None:
        self.lock = threading.Lock()
        self.next_at = 0.0


_sync_state: dict[tuple[str, str], _SyncBucketState] = {}
_sync_state_lock = threading.Lock()


def _get_sync_state(key: tuple[str, str]) -> _SyncBucketState:
    """Thread-safe ``_sync_state`` lookup-or-create.

    The async path uses an ``asyncio.Lock``; the sync path runs from
    arbitrary worker threads (every ``asyncio.to_thread`` lands on a
    fresh executor thread), so we need a real threading.Lock here. A
    plain ``dict.setdefault`` would also race the constructor (two
    threads each create a fresh state) — the lock prevents that.
    """
    state = _sync_state.get(key)
    if state is not None:
        return state
    with _sync_state_lock:
        state = _sync_state.get(key)
        if state is None:
            state = _SyncBucketState()
            _sync_state[key] = state
    return state