ksef_app/app/services/redis_service.py

from __future__ import annotations

import json
import time
from threading import Lock
from typing import Any
from urllib.parse import urlparse

from flask import current_app
from redis import Redis
from redis.exceptions import RedisError


class RedisService:
    _memory_store: dict[str, tuple[float | None, str]] = {}
    _lock = Lock()
    _failure_logged_at = 0.0
    _availability_cache: tuple[bool, float] = (False, 0.0)

    @classmethod
    def _config(cls, app=None):
        if app is not None:
            return app.config
        return current_app.config

    @classmethod
    def enabled(cls, app=None) -> bool:
        cfg = cls._config(app)
        return str(cfg.get('REDIS_URL', 'memory://')).strip().lower().startswith('redis://')

    @classmethod
    def url(cls, app=None) -> str:
        cfg = cls._config(app)
        return str(cfg.get('REDIS_URL', 'memory://')).strip() or 'memory://'

    @classmethod
    def _logger(cls, app=None):
        if app is not None:
            return app.logger
        return current_app.logger

    @classmethod
    def _log_failure_once(cls, action: str, exc: Exception, app=None) -> None:
        now = time.time()
        if now - cls._failure_logged_at < 60:
            return
        cls._failure_logged_at = now
        cls._logger(app).warning(
            'Redis %s niedostępny, przełączam na cache pamięciowy: %s', action, exc
        )

    @classmethod
    def client(cls, app=None) -> Redis | None:
        if not cls.enabled(app):
            return None
        try:
            return Redis.from_url(
                cls.url(app),
                decode_responses=True,
                socket_connect_timeout=1,
                socket_timeout=1,
            )
        except Exception as exc:
            cls._log_failure_once('client', exc, app)
            return None

    @classmethod
    def available(cls, app=None) -> bool:
        if not cls.enabled(app):
            return False
        cached_ok, checked_at = cls._availability_cache
        if time.time() - checked_at < 15:
            return cached_ok
        client = cls.client(app)
        if client is None:
            cls._availability_cache = (False, time.time())
            return False
        try:
            client.ping()
            cls._availability_cache = (True, time.time())
            return True
        except RedisError as exc:
            cls._log_failure_once('ping', exc, app)
            cls._availability_cache = (False, time.time())
            return False
        except Exception as exc:
            cls._log_failure_once('ping', exc, app)
            cls._availability_cache = (False, time.time())
            return False

    @classmethod
    def ping(cls, app=None) -> tuple[str, str]:
        url = cls.url(app)
        if not cls.enabled(app):
            return 'disabled', 'Cache pamięciowy aktywny (Redis wyłączony).'
        if cls.available(app):
            return 'ok', f'{url} · połączenie aktywne'
        parsed = urlparse(url)
        hint = ''
        if parsed.hostname in {'localhost', '127.0.0.1'}:
            hint = ' · w Dockerze użyj nazwy usługi redis zamiast localhost'
        return 'fallback', f'{url} · brak połączenia, aktywny fallback pamięciowy{hint}'

    @classmethod
    def _memory_get(cls, key: str) -> Any | None:
        now = time.time()
        with cls._lock:
            payload = cls._memory_store.get(key)
            if not payload:
                return None
            expires_at, raw = payload
            if expires_at is not None and expires_at <= now:
                cls._memory_store.pop(key, None)
                return None
        try:
            return json.loads(raw)
        except Exception:
            return None

    @classmethod
    def _memory_set(cls, key: str, value: Any, ttl: int = 60) -> bool:
        expires_at = None if ttl <= 0 else time.time() + ttl
        with cls._lock:
            cls._memory_store[key] = (expires_at, json.dumps(value, default=str))
        return True

    @classmethod
    def get_json(cls, key: str, app=None) -> Any | None:
        if not cls.enabled(app) or not cls.available(app):
            return cls._memory_get(key)
        client = cls.client(app)
        if client is None:
            return cls._memory_get(key)
        try:
            raw = client.get(key)
            if raw:
                return json.loads(raw)
            return cls._memory_get(key)
        except Exception as exc:
            cls._log_failure_once(f'get_json({key})', exc, app)
            return cls._memory_get(key)

    @classmethod
    def set_json(cls, key: str, value: Any, ttl: int = 60, app=None) -> bool:
        cls._memory_set(key, value, ttl=ttl)
        if not cls.enabled(app) or not cls.available(app):
            return False
        client = cls.client(app)
        if client is None:
            return False
        try:
            client.setex(key, ttl, json.dumps(value, default=str))
            return True
        except Exception as exc:
            cls._log_failure_once(f'set_json({key})', exc, app)
            return False

    @classmethod
    def delete(cls, key: str, app=None) -> None:
        with cls._lock:
            cls._memory_store.pop(key, None)
        if not cls.enabled(app) or not cls.available(app):
            return
        client = cls.client(app)
        if client is None:
            return
        try:
            client.delete(key)
        except Exception as exc:
            cls._log_failure_once(f'delete({key})', exc, app)