"""SlowAPI's fixed-window counters in PostgreSQL instead of Redis.

SlowAPI's storage interface is synchronous. A small Psycopg pool serves its
atomic counters, while the rest of the application continues using asyncpg.
Only hashed counter keys are persisted, and database failures never bypass limits.
"""

import hashlib
import time
from typing import Any

import psycopg
from limits.storage import Storage
from psycopg_pool import ConnectionPool
from sqlalchemy.engine import make_url


class PostgreSQLRateLimitStorage(Storage):
    """Shared fixed-window rate limits across processes and restarts."""

    STORAGE_SCHEME = ["postgresql+munal"]

    def __init__(self, uri: str, wrap_exceptions: bool = False, **options: float | str | bool) -> None:
        super().__init__(uri, wrap_exceptions=wrap_exceptions, **options)
        url = make_url(uri.replace("postgresql+munal://", "postgresql://", 1))
        query = dict(url.query)
        if "ssl" in query:
            query["sslmode"] = query.pop("ssl")
        self.pool: ConnectionPool[psycopg.Connection[tuple[Any, ...]]] = ConnectionPool(
            url.set(query=query).render_as_string(hide_password=False),
            min_size=0,
            max_size=2,
            timeout=10,
            open=False,
            kwargs={"autocommit": True},
        )

    @property
    def base_exceptions(self) -> type[Exception]:
        return psycopg.Error

    @staticmethod
    def _key(key: str) -> str:
        return "api_rate:" + hashlib.sha256(key.encode("utf-8")).hexdigest()

    def incr(self, key: str, expiry: int, amount: int = 1) -> int:
        self.pool.open()
        with self.pool.connection() as connection:
            row = connection.execute(
                """INSERT INTO expiring_state (key, value, expires_at)
                VALUES (%s, jsonb_build_object('count', %s::integer),
                        statement_timestamp() + make_interval(secs => %s))
                ON CONFLICT (key) DO UPDATE SET
                  value = CASE WHEN expiring_state.expires_at <= statement_timestamp()
                    THEN EXCLUDED.value
                    ELSE jsonb_build_object('count', (expiring_state.value->>'count')::integer + %s)
                    END,
                  expires_at = CASE WHEN expiring_state.expires_at <= statement_timestamp()
                    THEN EXCLUDED.expires_at ELSE expiring_state.expires_at END
                RETURNING (value->>'count')::integer""",
                (self._key(key), amount, max(expiry, 1), amount),
            ).fetchone()
        if row is None:
            raise RuntimeError("Rate limit counter was not persisted.")
        return int(row[0])

    def get(self, key: str) -> int:
        self.pool.open()
        with self.pool.connection() as connection:
            row = connection.execute(
                "SELECT (value->>'count')::integer FROM expiring_state "
                "WHERE key = %s AND expires_at > statement_timestamp()",
                (self._key(key),),
            ).fetchone()
        return int(row[0]) if row else 0

    def get_expiry(self, key: str) -> float:
        self.pool.open()
        with self.pool.connection() as connection:
            row = connection.execute(
                "SELECT extract(epoch FROM expires_at) FROM expiring_state "
                "WHERE key = %s AND expires_at > statement_timestamp()",
                (self._key(key),),
            ).fetchone()
        return float(row[0]) if row else time.time()

    def check(self) -> bool:
        try:
            self.pool.open()
            with self.pool.connection() as connection:
                connection.execute("SELECT 1 FROM expiring_state LIMIT 1")
            return True
        except psycopg.Error:
            return False

    def reset(self) -> int:
        self.pool.open()
        with self.pool.connection() as connection:
            cursor = connection.execute("DELETE FROM expiring_state WHERE key LIKE 'api_rate:%'")
            return cursor.rowcount

    def clear(self, key: str) -> None:
        self.pool.open()
        with self.pool.connection() as connection:
            connection.execute("DELETE FROM expiring_state WHERE key = %s", (self._key(key),))

    def close(self) -> None:
        self.pool.close()