"""Database-backed idempotency for cookie-authenticated mutations."""

from __future__ import annotations

import hashlib
import json
import re
from dataclasses import dataclass
from datetime import datetime, timedelta, timezone
from functools import wraps
from typing import Any

from flask import current_app, g, jsonify, make_response, request
from sqlalchemy import delete, select
from sqlalchemy.orm import Session

from ..models.operations import IdempotencyRecord
from ..utils.api_errors import ApiError
from ..utils.locale import t

_KEY_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$")


class IdempotencyConflict(ValueError):
    """The key was reused for a different request body."""


@dataclass(frozen=True)
class Reservation:
    record: IdempotencyRecord
    is_new: bool


class IdempotencyService:
    DEFAULT_TTL_SECONDS = 24 * 60 * 60

    def __init__(self, session: Session, *, ttl_seconds: int = DEFAULT_TTL_SECONDS):
        if ttl_seconds < 60 or ttl_seconds > 7 * 24 * 60 * 60:
            raise ValueError("invalid_idempotency_ttl")
        self.session = session
        self.ttl_seconds = ttl_seconds

    @staticmethod
    def _request_hash(request_body: Any) -> str:
        try:
            encoded = json.dumps(
                request_body,
                ensure_ascii=False,
                sort_keys=True,
                separators=(",", ":"),
            ).encode("utf-8")
        except (TypeError, ValueError) as exc:
            raise ValueError("invalid_idempotency_body") from exc
        return hashlib.sha256(encoded).hexdigest()

    @staticmethod
    def validate_key(key: str) -> str:
        if not isinstance(key, str) or not _KEY_RE.fullmatch(key):
            raise ValueError("invalid_idempotency_key")
        return key

    def reserve(
        self,
        *,
        organization_id: str,
        user_id: str,
        key: str,
        request_body: Any,
    ) -> Reservation:
        if not organization_id or not user_id:
            raise ValueError("invalid_idempotency_scope")
        normalized_key = self.validate_key(key)
        request_hash = self._request_hash(request_body)
        now = datetime.now(timezone.utc)
        existing = self.session.scalar(
            select(IdempotencyRecord)
            .where(
                IdempotencyRecord.organization_id == organization_id,
                IdempotencyRecord.user_id == user_id,
                IdempotencyRecord.key == normalized_key,
            )
            .with_for_update()
        )
        if existing is not None:
            expires_at = existing.expires_at
            if expires_at.tzinfo is None:
                expires_at = expires_at.replace(tzinfo=timezone.utc)
        else:
            expires_at = None
        if existing is not None and expires_at is not None and expires_at <= now:
            self.session.execute(delete(IdempotencyRecord).where(IdempotencyRecord.id == existing.id))
            self.session.flush()
            existing = None
        if existing is not None:
            if existing.request_hash != request_hash:
                raise IdempotencyConflict("idempotency_key_reused")
            return Reservation(record=existing, is_new=False)

        record = IdempotencyRecord(
            organization_id=organization_id,
            user_id=user_id,
            key=normalized_key,
            request_hash=request_hash,
            status="reserved",
            expires_at=now + timedelta(seconds=self.ttl_seconds),
        )
        self.session.add(record)
        self.session.flush()
        return Reservation(record=record, is_new=True)

    def complete(self, record: IdempotencyRecord, *, status_code: int, body: dict | list) -> None:
        if status_code < 100 or status_code > 599:
            raise ValueError("invalid_response_status")
        record.status = "completed"
        record.response_status = status_code
        record.response_body = body
        record.completed_at = datetime.now(timezone.utc)
        self.session.flush()

    def fail(self, record: IdempotencyRecord, *, status_code: int, body: dict | list) -> None:
        if status_code < 400 or status_code > 599:
            raise ValueError("invalid_failure_status")
        record.status = "failed"
        record.response_status = status_code
        record.response_body = body
        record.completed_at = datetime.now(timezone.utc)
        self.session.flush()


def _stream_sha256(stream) -> str:
    try:
        start_position = stream.tell()
    except (AttributeError, OSError, ValueError) as exc:
        raise ValueError("invalid_idempotency_body") from exc

    digest = hashlib.sha256()
    try:
        while True:
            chunk = stream.read(1024 * 1024)
            if not chunk:
                break
            digest.update(chunk)
    except (AttributeError, OSError, TypeError, ValueError) as exc:
        raise ValueError("invalid_idempotency_body") from exc
    finally:
        try:
            stream.seek(start_position)
        except (AttributeError, OSError, ValueError) as exc:
            raise ValueError("invalid_idempotency_body") from exc
    return digest.hexdigest()


def _request_fingerprint_payload() -> Any:
    if request.is_json:
        body = request.get_json(silent=True)
    elif request.form or request.files:
        files = [
            {
                "field": field,
                "filename": file.filename or "",
                "content_type": file.content_type or "",
                "size": request.content_length or 0,
                "content_sha256": _stream_sha256(file.stream),
            }
            for field, files in request.files.lists()
            for file in files
        ]
        body = {"form": request.form.to_dict(flat=False), "files": files}
    else:
        body = {}
    return {
        "method": request.method,
        "path": request.path,
        "query": request.args.to_dict(flat=False),
        "body": body,
    }


def idempotent(view):
    """Require and persist an ``Idempotency-Key`` for a JSON mutation."""

    @wraps(view)
    def wrapped(*args, **kwargs):
        key = request.headers.get("Idempotency-Key", "")
        if not key:
            raise ApiError("idempotency_required", 400, "api.idempotencyRequired")
        session = getattr(g, "db_session", None)
        managed_session = False
        if session is None:
            factory = current_app.extensions.get("crowdsight_session_factory")
            if not callable(factory):
                raise ApiError("auth_unavailable", 503, "api.internalError")
            session = factory()
            managed_session = True
        context = getattr(g, "auth_context", None)
        if context is None:
            if managed_session:
                session.close()
            raise ApiError("auth_unavailable", 503, "api.internalError")
        try:
            try:
                reservation = IdempotencyService(session).reserve(
                    organization_id=context.organization.id,
                    user_id=context.user.id,
                    key=key,
                    request_body=_request_fingerprint_payload(),
                )
            except IdempotencyConflict as exc:
                raise ApiError("idempotency_key_reused", 409, "api.idempotencyConflict") from exc
            except ValueError as exc:
                code = str(exc)
                if code == "invalid_idempotency_key":
                    raise ApiError(code, 400, "api.idempotencyInvalid") from exc
                raise ApiError("invalid_request", 400, "api.requestError") from exc

            if not reservation.is_new:
                record = reservation.record
                if record.status == "completed" and record.response_body is not None:
                    return jsonify(record.response_body), record.response_status or 200
                if record.status == "failed" and record.response_body is not None:
                    return jsonify(record.response_body), record.response_status or 500
                raise ApiError("idempotency_in_progress", 409, "api.idempotencyInProgress")

            response = make_response(view(*args, **kwargs))
            body = response.get_json(silent=True)
            if not isinstance(body, (dict, list)):
                raise ApiError("idempotency_response_invalid", 500, "api.internalError")
            IdempotencyService(session).complete(
                reservation.record,
                status_code=response.status_code,
                body=body,
            )
            if managed_session:
                session.commit()
            return response
        except Exception:
            if managed_session:
                session.rollback()
            raise
        finally:
            if managed_session:
                session.close()

    return wrapped