"""Persistent, multi-node-safe quota accounting."""

from __future__ import annotations

from dataclasses import asdict, dataclass
from datetime import datetime, timedelta
from decimal import Decimal
from typing import Any

from sqlalchemy import case, func, or_, select

from config.control_plane import get_control_plane_session
from bridge_platform.tenants.models import TenantLimit, TenantUsage


METRIC_LIMITS = {
    "requests": "requests_per_minute",
    "ai_tokens": "ai_tokens_monthly",
    "jobs": "monthly_jobs",
}
RESET_SUFFIX = "_reset"


class QuotaExceeded(PermissionError):
    def __init__(self, decision: "QuotaDecision") -> None:
        self.decision = decision
        super().__init__(
            f"Quota exceeded for {decision.limit_name}: "
            f"{decision.used:g}/{decision.limit:g}"
        )


@dataclass(frozen=True)
class QuotaDecision:
    metric: str
    limit_name: str
    limit: float | None
    used: float
    remaining: float | None
    period_start: datetime
    period_end: datetime
    allowed: bool
    policy: str

    def to_dict(self) -> dict[str, Any]:
        payload = asdict(self)
        payload["period_start"] = self.period_start.isoformat()
        payload["period_end"] = self.period_end.isoformat()
        return payload


def consume(
    *,
    tenant_id: str,
    app_id: str | None,
    metric: str,
    amount: float = 1,
    metadata: dict[str, Any] | None = None,
) -> QuotaDecision:
    """Atomically check and record usage.

    Locking the effective limit row serializes decisions across physical and
    virtual workers when the control plane runs on PostgreSQL/MySQL.
    """
    if amount < 0:
        raise ValueError("Usage amount cannot be negative")
    limit_name = METRIC_LIMITS.get(metric, metric)
    now = datetime.utcnow()

    with get_control_plane_session() as session:
        limit_row = _effective_limit_query(tenant_id, app_id, limit_name).with_for_update()
        limit_record = session.execute(limit_row).scalars().first()
        period_start, period_end = _period(now, limit_record)
        usage_start = _usage_start(session, tenant_id, metric, period_start, period_end)
        used = float(
            session.execute(
                select(func.coalesce(func.sum(TenantUsage.metric_value), 0)).where(
                    TenantUsage.tenant_id == tenant_id,
                    TenantUsage.metric_name == metric,
                    TenantUsage.recorded_at >= usage_start,
                    TenantUsage.recorded_at < period_end,
                )
            ).scalar_one()
            or 0
        )
        limit = float(limit_record.limit_value) if limit_record and limit_record.limit_value is not None else None
        policy = (limit_record.overage_policy or "block") if limit_record else "unlimited"
        allowed = limit is None or used + amount <= limit or policy != "block"
        projected = used + amount
        decision = QuotaDecision(
            metric=metric,
            limit_name=limit_name,
            limit=limit,
            used=projected if allowed else used,
            remaining=None if limit is None else max(0.0, limit - (projected if allowed else used)),
            period_start=period_start,
            period_end=period_end,
            allowed=allowed,
            policy=policy,
        )
        if not allowed:
            session.rollback()
            raise QuotaExceeded(decision)

        session.add(
            TenantUsage(
                tenant_id=tenant_id,
                app_id=app_id,
                metric_name=metric,
                metric_value=Decimal(str(amount)),
                usage_period_start=period_start,
                usage_period_end=period_end,
                metadata_json=metadata or {},
            )
        )
        session.commit()
        return decision


def ensure_capacity(*, tenant_id: str, app_id: str | None, metric: str, amount: float) -> QuotaDecision:
    """Check capacity atomically without recording usage or calling a provider."""
    if amount < 0:
        raise ValueError("Usage amount cannot be negative")
    limit_name = METRIC_LIMITS.get(metric, metric)
    now = datetime.utcnow()
    with get_control_plane_session() as session:
        limit_record = session.execute(
            _effective_limit_query(tenant_id, app_id, limit_name).with_for_update()
        ).scalars().first()
        period_start, period_end = _period(now, limit_record)
        usage_start = _usage_start(session, tenant_id, metric, period_start, period_end)
        used = float(session.execute(select(func.coalesce(func.sum(TenantUsage.metric_value), 0)).where(
            TenantUsage.tenant_id == tenant_id, TenantUsage.metric_name == metric,
            TenantUsage.recorded_at >= usage_start, TenantUsage.recorded_at < period_end,
        )).scalar_one() or 0)
        limit = float(limit_record.limit_value) if limit_record and limit_record.limit_value is not None else None
        policy = (limit_record.overage_policy or "block") if limit_record else "unlimited"
        allowed = limit is None or used + amount <= limit or policy != "block"
        decision = QuotaDecision(
            metric=metric, limit_name=limit_name, limit=limit, used=used,
            remaining=None if limit is None else max(0.0, limit - used),
            period_start=period_start, period_end=period_end, allowed=allowed, policy=policy,
        )
        if not allowed:
            raise QuotaExceeded(decision)
        return decision


def record_observed_usage(
    *, tenant_id: str, app_id: str | None, metric: str, amount: float,
    metadata: dict[str, Any] | None = None,
) -> None:
    """Record provider usage that has already occurred; never reject it retroactively."""
    # Signed adjustments reconcile a prior reservation with observed provider usage.
    now = datetime.utcnow()
    with get_control_plane_session() as session:
        limit_record = session.execute(_effective_limit_query(
            tenant_id, app_id, METRIC_LIMITS.get(metric, metric)
        )).scalars().first()
        period_start, period_end = _period(now, limit_record)
        session.add(TenantUsage(
            tenant_id=tenant_id, app_id=app_id, metric_name=metric,
            metric_value=Decimal(str(amount)), usage_period_start=period_start,
            usage_period_end=period_end, metadata_json=metadata or {},
        ))
        session.commit()


def usage_snapshot(tenant_id: str) -> dict[str, dict[str, Any]]:
    """Return current usage and balances for all configured tenant limits."""
    now = datetime.utcnow()
    result: dict[str, dict[str, Any]] = {}
    with get_control_plane_session() as session:
        limits = session.execute(
            select(TenantLimit)
            .where(TenantLimit.tenant_id == tenant_id)
            .order_by(TenantLimit.app_id, TenantLimit.limit_name)
        ).scalars().all()
        for record in limits:
            # Storage is a point-in-time filesystem measurement, not an
            # additive ledger metric. It will be exposed by the storage layer.
            if record.limit_name == "storage_mb":
                continue
            metric = next((key for key, value in METRIC_LIMITS.items() if value == record.limit_name), record.limit_name)
            start, end = _period(now, record)
            usage_start = _usage_start(session, tenant_id, metric, start, end)
            filters = [
                TenantUsage.tenant_id == tenant_id,
                TenantUsage.metric_name == metric,
                TenantUsage.recorded_at >= usage_start,
                TenantUsage.recorded_at < end,
            ]
            if record.app_id:
                filters.append(TenantUsage.app_id == record.app_id)
            used = float(session.execute(
                select(func.coalesce(func.sum(TenantUsage.metric_value), 0)).where(*filters)
            ).scalar_one() or 0)
            limit = float(record.limit_value) if record.limit_value is not None else None
            key = f"{record.app_id}:{record.limit_name}" if record.app_id else record.limit_name
            result[key] = {
                "metric": metric,
                "app_id": record.app_id,
                "used": used,
                "limit": limit,
                "remaining": None if limit is None else max(0.0, limit - used),
                "percentage": 0 if not limit else min(100, round((used / limit) * 100, 2)),
                "period_start": start.isoformat(),
                "period_end": end.isoformat(),
                "status": "blocked" if limit is not None and used >= limit and (record.overage_policy or "block") == "block" else "available",
            }
    return result


def reset_usage_counter(*, tenant_id: str, metric: str, reason: str) -> dict[str, Any]:
    """Reset a tenant counter without deleting its immutable usage history."""
    if not tenant_id or metric not in METRIC_LIMITS or not str(reason).strip():
        raise ValueError("invalid_usage_reset")
    now = datetime.utcnow()
    with get_control_plane_session() as session:
        period_start, period_end = _period(now, None)
        previous = float(session.execute(
            select(func.coalesce(func.sum(TenantUsage.metric_value), 0)).where(
                TenantUsage.tenant_id == tenant_id,
                TenantUsage.metric_name == metric,
                TenantUsage.recorded_at >= period_start,
                TenantUsage.recorded_at < period_end,
            )
        ).scalar_one() or 0)
        session.add(TenantUsage(
            tenant_id=tenant_id, app_id=None,
            metric_name=f"{metric}{RESET_SUFFIX}", metric_value=Decimal("0"),
            usage_period_start=period_start, usage_period_end=period_end,
            metadata_json={"reason": str(reason).strip()[:160], "previous_usage": previous},
        ))
        session.commit()
    return {"tenant_id": tenant_id, "metric": metric, "previous_usage": previous, "used": 0.0}


def _usage_start(session, tenant_id: str, metric: str, period_start: datetime, period_end: datetime) -> datetime:
    reset_at = session.execute(select(func.max(TenantUsage.recorded_at)).where(
        TenantUsage.tenant_id == tenant_id,
        TenantUsage.metric_name == f"{metric}{RESET_SUFFIX}",
        TenantUsage.recorded_at >= period_start,
        TenantUsage.recorded_at < period_end,
    )).scalar_one_or_none()
    return max(period_start, reset_at) if reset_at else period_start


def _effective_limit_query(tenant_id: str, app_id: str | None, limit_name: str):
    scope_filter = (
        or_(TenantLimit.app_id == app_id, TenantLimit.app_id.is_(None))
        if app_id else TenantLimit.app_id.is_(None)
    )
    return (
        select(TenantLimit)
        .where(
            TenantLimit.tenant_id == tenant_id,
            TenantLimit.limit_name == limit_name,
            scope_filter,
        )
        .order_by(case((TenantLimit.app_id == app_id, 0), else_=1))
        .limit(1)
    )


def _period(now: datetime, limit: TenantLimit | None) -> tuple[datetime, datetime]:
    if limit and limit.window_seconds:
        seconds = int(limit.window_seconds)
        epoch = int(now.timestamp())
        start = datetime.utcfromtimestamp(epoch - epoch % seconds)
        return start, start + timedelta(seconds=seconds)
    start = datetime(now.year, now.month, 1)
    end = datetime(now.year + (now.month == 12), 1 if now.month == 12 else now.month + 1, 1)
    return start, end