Add Hygraph circuit breaker client and caching (#108)

* Add Hygraph circuit breaker client and caching

* Handle half-open failures with fallback

Author: shayancoin

backend/api/routes_sync.py

@@ -32,6 +32,7 @@
 from opentelemetry import trace
 from opentelemetry.trace import Status, StatusCode
 from services.hygraph_service import HygraphService
+from services.circuit_breaker import CircuitOpenError
 
 logger = logging.getLogger(__name__)
 router = APIRouter(prefix="/api/sync", tags=["sync"])
@@ -239,6 +240,16 @@ async def hygraph_pull(
             raise HTTPException(status_code=400, detail=_error_envelope("BAD_REQUEST", "unsupported type"))
     except HTTPException:
         raise
+    except CircuitOpenError as exc:
+        hygraph_cb_trips_total.inc()
+        fallback_result = exc.fallback_result
+        if fallback_result is not None:
+            hygraph_cb_fallback_total.inc()
+            return {"ok": True, "data": fallback_result, "cached": True}
+        raise HTTPException(
+            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
+            detail=_error_envelope("SERVICE_UNAVAILABLE", "Hygraph unavailable", {"type": sync_type}),
+        )
     except Exception as e:  # noqa: BLE001
         record_sync_failure(sync_type or "all")
         record_sync_duration(sync_type or "all", (time.perf_counter() - start) * 1000)

backend/services/cache.py

@@ -0,0 +1,73 @@
+"""Simple Redis-backed cache helper with in-memory fallback."""
+
+from __future__ import annotations
+
+import json
+import time
+from dataclasses import dataclass, field
+from typing import Any, Dict, Optional
+
+try:
+    from redis import asyncio as redis_asyncio
+except Exception:  # pragma: no cover - optional dependency guard
+    redis_asyncio = None  # type: ignore[assignment]
+
+
+@dataclass
+class RedisCache:
+    """Minimal JSON cache built on top of Redis with TTL support."""
+
+    url: str
+    namespace: str = "hygraph"
+    use_memory_fallback: bool = True
+    _client: Optional["redis_asyncio.Redis"] = field(default=None, init=False, repr=False)
+    _memory_store: Dict[str, tuple[Any, Optional[float]]] = field(default_factory=dict, init=False, repr=False)
+
+    def __post_init__(self) -> None:
+        if redis_asyncio is not None:
+            try:
+                self._client = redis_asyncio.from_url(self.url, decode_responses=True)
+            except Exception:
+                self._client = None
+
+    def _prefixed(self, key: str) -> str:
+        return f"{self.namespace}:{key}" if self.namespace else key
+
+    async def get_json(self, key: str) -> Any | None:
+        full_key = self._prefixed(key)
+        if self._client is not None:
+            try:
+                raw = await self._client.get(full_key)
+            except Exception:
+                raw = None
+            if raw is not None:
+                try:
+                    return json.loads(raw)
+                except json.JSONDecodeError:
+                    return None
+        if not self.use_memory_fallback:
+            return None
+        payload = self._memory_store.get(full_key)
+        if payload is None:
+            return None
+        value, expires_at = payload
+        if expires_at is not None and expires_at < time.monotonic():
+            self._memory_store.pop(full_key, None)
+            return None
+        return value
+
+    async def set_json(self, key: str, value: Any, *, ttl: Optional[int] = None) -> None:
+        full_key = self._prefixed(key)
+        dumped = json.dumps(value)
+        if self._client is not None:
+            try:
+                await self._client.set(full_key, dumped, ex=ttl)
+            except Exception:
+                # Fall back to memory cache if Redis write fails.
+                pass
+        if not self.use_memory_fallback:
+            return
+        expires_at = None
+        if ttl is not None:
+            expires_at = time.monotonic() + ttl
+        self._memory_store[full_key] = (value, expires_at)

backend/services/circuit_breaker.py

@@ -0,0 +1,108 @@
+"""Simple circuit breaker implementation for Hygraph integrations."""
+
+from __future__ import annotations
+
+import time
+from dataclasses import dataclass, field
+from typing import Any, Callable, Optional, TypeVar
+
+
+T = TypeVar("T")
+
+
+class CircuitOpenError(RuntimeError):
+    """Raised when the circuit breaker is OPEN and calls are blocked."""
+
+    def __init__(self, message: str, fallback_result: Any | None = None) -> None:
+        super().__init__(message)
+        self.fallback_result = fallback_result
+
+
+@dataclass
+class CircuitBreaker:
+    """Dataclass based circuit breaker with simple HALF_OPEN support."""
+
+    failure_threshold: int = 5
+    recovery_timeout: float = 30.0
+    half_open_success_threshold: int = 1
+    time_provider: Callable[[], float] = time.monotonic
+    state: str = field(default="CLOSED", init=False)
+    failure_count: int = field(default=0, init=False)
+    success_count: int = field(default=0, init=False)
+    opened_at: Optional[float] = field(default=None, init=False)
+
+    CLOSED: str = "CLOSED"
+    OPEN: str = "OPEN"
+    HALF_OPEN: str = "HALF_OPEN"
+
+    def _transition_to_open(self) -> None:
+        self.state = self.OPEN
+        self.opened_at = self.time_provider()
+        self.failure_count = 0
+        self.success_count = 0
+
+    def _transition_to_half_open(self) -> None:
+        self.state = self.HALF_OPEN
+        self.opened_at = None
+        self.failure_count = 0
+        self.success_count = 0
+
+    def _transition_to_closed(self) -> None:
+        self.state = self.CLOSED
+        self.opened_at = None
+        self.failure_count = 0
+        self.success_count = 0
+
+    def _ready_for_half_open(self) -> bool:
+        if self.opened_at is None:
+            return False
+        return (self.time_provider() - self.opened_at) >= self.recovery_timeout
+
+    def _record_failure(self) -> None:
+        if self.state == self.HALF_OPEN:
+            self._transition_to_open()
+            return
+        self.failure_count += 1
+        if self.failure_count >= self.failure_threshold:
+            self._transition_to_open()
+
+    def _record_success(self) -> None:
+        if self.state == self.HALF_OPEN:
+            self.success_count += 1
+            if self.success_count >= self.half_open_success_threshold:
+                self._transition_to_closed()
+                return
+        elif self.state == self.OPEN:
+            # Should not happen, but close proactively.
+            self._transition_to_closed()
+        else:
+            self.failure_count = 0
+
+    def call(
+        self,
+        func: Callable[..., T],
+        *args: Any,
+        fallback: Optional[Callable[[], T]] = None,
+        **kwargs: Any,
+    ) -> T:
+        """Execute ``func`` guarding access through the circuit breaker."""
+
+        if self.state == self.OPEN:
+            if self._ready_for_half_open():
+                self._transition_to_half_open()
+            else:
+                fallback_result: T | None = fallback() if fallback else None
+                raise CircuitOpenError("Circuit breaker is open", fallback_result)
+
+        try:
+            result = func(*args, **kwargs)
+        except Exception as exc:
+            was_half_open = self.state == self.HALF_OPEN
+            self._record_failure()
+            if was_half_open:
+                fallback_result: T | None = fallback() if fallback else None
+                raise CircuitOpenError("Circuit breaker is open", fallback_result) from exc
+            raise
+        else:
+            self._record_success()
+            return result

backend/services/hygraph_client.py

@@ -0,0 +1,77 @@
+"""Hygraph GraphQL client with circuit breaker protection."""
+
+from __future__ import annotations
+
+import httpx
+from typing import Any, Callable, Dict, Optional
+
+from services.circuit_breaker import CircuitBreaker, CircuitOpenError
+
+
+class HygraphGraphQLError(RuntimeError):
+    """Raised when Hygraph returns GraphQL errors."""
+
+    def __init__(self, errors: Any) -> None:
+        super().__init__("Hygraph GraphQL error")
+        self.errors = errors
+
+
+class HygraphClient:
+    """HTTPX based client for Hygraph GraphQL requests."""
+
+    def __init__(
+        self,
+        endpoint: str,
+        token: str,
+        *,
+        timeout: float = 6.0,
+        breaker: Optional[CircuitBreaker] = None,
+        client: Optional[httpx.Client] = None,
+    ) -> None:
+        if not endpoint:
+            raise ValueError("Hygraph endpoint must be configured")
+        self._endpoint = endpoint
+        self._token = token
+        self._client = client or httpx.Client(timeout=timeout)
+        self._breaker = breaker or CircuitBreaker()
+        self._client.headers.setdefault("Content-Type", "application/json")
+        self._client.headers.setdefault("Accept", "application/json")
+        if token:
+            self._client.headers.setdefault("Authorization", f"Bearer {token}")
+
+    @property
+    def breaker(self) -> CircuitBreaker:
+        return self._breaker
+
+    def execute(
+        self,
+        query: str,
+        variables: Optional[Dict[str, Any]] = None,
+        *,
+        fallback: Optional[Callable[[], Any]] = None,
+    ) -> Dict[str, Any]:
+        """Run the GraphQL ``query`` through the circuit breaker."""
+
+        def _request() -> Dict[str, Any]:
+            response = self._client.post(
+                self._endpoint,
+                json={"query": query, "variables": variables or {}},
+            )
+            response.raise_for_status()
+            payload = response.json()
+            errors = payload.get("errors")
+            if errors:
+                raise HygraphGraphQLError(errors)
+            data = payload.get("data")
+            if not isinstance(data, dict):
+                return {}
+            return data
+
+        try:
+            return self._breaker.call(_request, fallback=fallback)
+        except CircuitOpenError:
+            # Re-raise so callers can handle fallback semantics.
+            raise
+
+    def close(self) -> None:
+        self._client.close()