"""Release gate and bounded operational defaults for Intelligence Center."""
from __future__ import annotations

import logging
from typing import Any
from uuid import UUID

from sqlalchemy.ext.asyncio import AsyncSession

from app.core.exceptions import AppException
from app.core.redis_client import get_redis
from app.integrations.models import PlatformSetting

_log = logging.getLogger(__name__)

SETTING_ENABLED = "intelligence.enabled"
SETTING_DAILY_LIMIT = "intelligence.generate_per_user_per_day"
SETTING_CLOUD_CACHE_DAYS = "intelligence.cloud_scope.cache_days"
SETTING_CLOUD_COOLDOWN_MINUTES = "intelligence.cloud_scope.cooldown_minutes"
SETTING_READS_PER_MINUTE = "intelligence.reads_per_minute"

DEFAULT_DAILY_LIMIT = 5
DEFAULT_CLOUD_CACHE_DAYS = 7
DEFAULT_CLOUD_COOLDOWN_MINUTES = 30
DEFAULT_READS_PER_MINUTE = 120
_READ_WINDOW_SECONDS = 60
_READ_LIMIT_SCRIPT = """
local current = redis.call('INCR', KEYS[1])
local ttl = redis.call('TTL', KEYS[1])
if current == 1 or ttl < 0 then
  redis.call('EXPIRE', KEYS[1], ARGV[1])
end
return current
"""


async def setting_value(db: AsyncSession, key: str) -> Any:
    row = await db.get(PlatformSetting, key)
    return row.value if row else None


async def intelligence_enabled(db: AsyncSession) -> bool:
    # Exact bool only: strings such as "true" must not accidentally release a
    # hidden product because an operator edited JSON incorrectly.
    return await setting_value(db, SETTING_ENABLED) is True


async def require_intelligence_enabled(db: AsyncSession) -> None:
    if not await intelligence_enabled(db):
        raise AppException(
            code="INTELLIGENCE_DISABLED",
            message="Intelligence Center is not enabled.",
            status_code=404,
        )


async def bounded_int_setting(
    db: AsyncSession, key: str, *, default: int, minimum: int, maximum: int,
) -> int:
    raw = await setting_value(db, key)
    try:
        value = int(raw) if raw is not None else default
    except (TypeError, ValueError):
        value = default
    return max(minimum, min(maximum, value))


async def apply_intelligence_read_limit(
    db: AsyncSession,
    *,
    user_id: UUID | str,
) -> None:
    """Apply one shared per-user budget to Intelligence HTTP and MCP reads.

    The product has its own inexpensive read budget instead of consuming the
    much smaller sensitive-data allowance. Redis failure is fail-open: these
    endpoints expose only the already-authorized aggregate projection and an
    infrastructure outage must not take the management dashboard down.
    """
    limit = await bounded_int_setting(
        db,
        SETTING_READS_PER_MINUTE,
        default=DEFAULT_READS_PER_MINUTE,
        minimum=30,
        maximum=600,
    )
    key = f"intelligence_read_limit:{user_id}"
    try:
        redis = get_redis()
        # One Lua operation guarantees the first increment and TTL are atomic.
        # It also works before Redis 7 (unlike EXPIRE NX), so a crash or mixed
        # deployment can never strand a permanent counter that locks a user out.
        count = int(await redis.eval(
            _READ_LIMIT_SCRIPT,
            1,
            key,
            _READ_WINDOW_SECONDS,
        ))
        if count > limit:
            ttl = await redis.ttl(key)
            if ttl < 0:
                ttl = _READ_WINDOW_SECONDS
            raise AppException(
                code="INTELLIGENCE_READ_RATE_LIMITED",
                message="Too many Intelligence Center reads; slow down.",
                status_code=429,
                headers={"Retry-After": str(ttl)},
            )
    except AppException:
        raise
    except Exception as exc:  # pragma: no cover - Redis outage path
        _log.warning("Intelligence read limiter unavailable: %s", exc)