from __future__ import annotations

import json
import re
import threading
from typing import Any, Dict, Optional

from fastapi import HTTPException, Request

from grantflow.api.errors import agent_error_detail
from grantflow.api.idempotency_store_facade import _get_job, _update_job
from grantflow.api.redis_store import redis_idem_get, redis_idem_set
from grantflow.api.review_runtime_helpers import _utcnow_iso

REQUEST_ID_RE = re.compile(r"^[A-Za-z0-9._:-]{1,120}$")
MAX_IDEMPOTENCY_RECORDS = 300
MAX_GLOBAL_IDEMPOTENCY_RECORDS = 1000
GLOBAL_IDEMPOTENCY_RECORDS: Dict[str, Dict[str, Any]] = {}
GLOBAL_IDEMPOTENCY_LOCK = threading.Lock()


def _normalize_request_id(value: Any) -> Optional[str]:
    token = str(value or "").strip()
    if not token:
        return None
    if not REQUEST_ID_RE.fullmatch(token):
        raise HTTPException(
            status_code=400,
            detail=agent_error_detail(
                code="invalid_request_id",
                message="Invalid request_id. Allowed characters: letters, numbers, ., _, :, -, length 1..120.",
                retryable=False,
                next_action="Generate a stable idempotency key using only letters, numbers, ., _, :, or -.",
                request_id=token,
            ),
        )
    return token


def _resolve_request_id(request: Request, explicit_request_id: Optional[str] = None) -> Optional[str]:
    return _normalize_request_id(explicit_request_id) or _normalize_request_id(request.headers.get("x-request-id"))


def _idempotency_fingerprint(payload: Dict[str, Any]) -> str:
    return json.dumps(payload, sort_keys=True, separators=(",", ":"), ensure_ascii=True)


def _idempotency_record_key(scope: str, request_id: str) -> str:
    return f"{scope}:{request_id}"


def _idempotency_records(job: Dict[str, Any]) -> Dict[str, Dict[str, Any]]:
    raw = job.get("idempotency_records")
    if not isinstance(raw, dict):
        return {}
    out: Dict[str, Dict[str, Any]] = {}
    for key, value in raw.items():
        if not isinstance(key, str) or not isinstance(value, dict):
            continue
        out[key] = dict(value)
    return out


def _idempotency_replay_response(
    job: Dict[str, Any],
    *,
    scope: str,
    request_id: Optional[str],
    fingerprint: str,
) -> Optional[Dict[str, Any]]:
    token = _normalize_request_id(request_id)
    if not token:
        return None
    key = _idempotency_record_key(scope, token)
    record = _idempotency_records(job).get(key)
    if not isinstance(record, dict):
        return None
    record_fingerprint = str(record.get("fingerprint") or "")
    if record_fingerprint and record_fingerprint != fingerprint:
        raise HTTPException(
            status_code=409,
            detail=agent_error_detail(
                code="request_id_reused_with_different_payload",
                message="request_id is already used for a different action payload.",
                retryable=False,
                next_action="Reuse the original payload for this request_id or generate a new request_id.",
                reason="request_id_reused_with_different_payload",
                request_id=token,
                scope=scope,
            ),
        )
    response_payload = record.get("response")
    if isinstance(response_payload, dict):
        replay = dict(response_payload)
    else:
        replay = {}
    replay["request_id"] = token
    replay["idempotent_replay"] = True
    replay["persisted"] = bool(record.get("persisted", True))
    return replay


def _store_idempotency_response(
    job_id: str,
    *,
    scope: str,
    request_id: Optional[str],
    fingerprint: str,
    response: Dict[str, Any],
    persisted: bool,
) -> None:
    token = _normalize_request_id(request_id)
    if not token:
        return
    job = _get_job(job_id)
    if not job:
        return
    records = _idempotency_records(job)
    key = _idempotency_record_key(scope, token)
    stored_response = dict(response)
    stored_response.pop("idempotent_replay", None)
    stored_response["request_id"] = token
    records[key] = {
        "scope": scope,
        "request_id": token,
        "fingerprint": fingerprint,
        "persisted": bool(persisted),
        "ts": _utcnow_iso(),
        "response": stored_response,
    }
    while len(records) > MAX_IDEMPOTENCY_RECORDS:
        oldest_key = next(iter(records))
        records.pop(oldest_key, None)
    _update_job(job_id, idempotency_records=records)


def _global_idempotency_record_key(scope: str, request_id: str) -> str:
    return f"{scope}:{request_id}"


def _global_idempotency_replay_response(
    *,
    scope: str,
    request_id: Optional[str],
    fingerprint: str,
) -> Optional[Dict[str, Any]]:
    token = _normalize_request_id(request_id)
    if not token:
        return None
    key = _global_idempotency_record_key(scope, token)
    # Try Redis first (survives restarts); fall back to in-process dict.
    redis_record = redis_idem_get(scope, token)
    with GLOBAL_IDEMPOTENCY_LOCK:
        in_proc_record = dict(GLOBAL_IDEMPOTENCY_RECORDS.get(key) or {})
    record = redis_record or in_proc_record
    if not record:
        return None
    record_fingerprint = str(record.get("fingerprint") or "")
    if record_fingerprint and record_fingerprint != fingerprint:
        raise HTTPException(
            status_code=409,
            detail=agent_error_detail(
                code="request_id_reused_with_different_payload",
                message="request_id is already used for a different action payload.",
                retryable=False,
                next_action="Reuse the original payload for this request_id or generate a new request_id.",
                reason="request_id_reused_with_different_payload",
                request_id=token,
                scope=scope,
            ),
        )
    response_payload = record.get("response")
    if isinstance(response_payload, dict):
        replay = dict(response_payload)
    else:
        replay = {}
    replay["request_id"] = token
    replay["idempotent_replay"] = True
    replay["persisted"] = bool(record.get("persisted", True))
    return replay


def _store_global_idempotency_response(
    *,
    scope: str,
    request_id: Optional[str],
    fingerprint: str,
    response: Dict[str, Any],
    persisted: bool,
) -> None:
    token = _normalize_request_id(request_id)
    if not token:
        return
    key = _global_idempotency_record_key(scope, token)
    stored_response = dict(response)
    stored_response.pop("idempotent_replay", None)
    stored_response["request_id"] = token
    record: Dict[str, Any] = {
        "scope": scope,
        "request_id": token,
        "fingerprint": fingerprint,
        "persisted": bool(persisted),
        "ts": _utcnow_iso(),
        "response": stored_response,
    }
    # Write to Redis (durable) and in-process dict (fast local reads).
    redis_idem_set(scope, token, record)
    with GLOBAL_IDEMPOTENCY_LOCK:
        GLOBAL_IDEMPOTENCY_RECORDS[key] = record
        while len(GLOBAL_IDEMPOTENCY_RECORDS) > MAX_GLOBAL_IDEMPOTENCY_RECORDS:
            oldest_key = next(iter(GLOBAL_IDEMPOTENCY_RECORDS))
            GLOBAL_IDEMPOTENCY_RECORDS.pop(oldest_key, None)
