sklik rozsireni na typove metody

app/idempotency.py

@@ -0,0 +1,124 @@
+"""Idempotency for write operations.
+
+Why this exists: if a create request times out on the network, the caller cannot
+know whether the campaign was created. Retrying without protection creates a
+second campaign that also spends money. An idempotency key makes the retry
+return the ORIGINAL result instead of doing the work twice.
+
+Scope and honest limitations - this is an **in-memory, per-container** store:
+
+  * it does not survive a restart or a redeploy;
+  * it is not shared between replicas, so with more than one container a retry
+    can land on a replica that has never seen the key.
+
+That is a deliberate trade-off matching the stateless design (the same one the
+Google token cache in ``clients/google.py`` makes). It removes the common
+failure - an immediate client retry after a timeout - and it is a large
+improvement over nothing. If Sklik writes ever run at a volume where a duplicated
+campaign is unacceptable, this needs to move to shared storage (Redis) and that
+should be a conscious decision, not a surprise. Keys are hashed together with the
+caller's token so two tenants can never read each other's stored responses.
+"""
+from __future__ import annotations
+
+import hashlib
+import threading
+import time
+from dataclasses import dataclass
+from typing import Any
+
+from . import config
+from .errors import UpstreamError
+from .logging_config import get_logger
+
+logger = get_logger(__name__)
+
+
+@dataclass
+class _Entry:
+    #: None while the original request is still running.
+    response: Any
+    created_at: float
+    in_flight: bool
+
+
+_store: dict[str, _Entry] = {}
+_lock = threading.Lock()
+
+
+def _cache_key(idempotency_key: str, token: str, operation: str) -> str:
+    # Hashed so neither the raw token nor the caller's key sits in a dict key we
+    # might later log. The token is part of the key so tenants stay isolated.
+    raw = f"{idempotency_key}|{token}|{operation}"
+    return hashlib.sha256(raw.encode("utf-8")).hexdigest()
+
+
+def _purge_expired(now: float) -> None:
+    """Drop entries past their TTL. Called under the lock."""
+    ttl = config.SKLIK_IDEMPOTENCY_TTL_SECONDS
+    expired = [k for k, e in _store.items() if now - e.created_at > ttl]
+    for key in expired:
+        del _store[key]
+
+
+def begin(idempotency_key: str, token: str, operation: str) -> tuple[str, Any | None]:
+    """Claim an idempotency key before performing a write.
+
+    Returns ``(cache_key, previous_response)``. A non-None previous response
+    means this exact request already succeeded and must NOT be performed again -
+    return the stored response instead.
+
+    Raises ``UpstreamError`` (409) if an identical request is still in flight,
+    so two concurrent retries cannot both reach Sklik.
+    """
+    key = _cache_key(idempotency_key, token, operation)
+    now = time.time()
+    with _lock:
+        _purge_expired(now)
+        entry = _store.get(key)
+        if entry is None:
+            _store[key] = _Entry(response=None, created_at=now, in_flight=True)
+            return key, None
+        if entry.in_flight:
+            raise UpstreamError(
+                "A request with this X-Idempotency-Key is still in progress. "
+                "Wait for it to finish rather than retrying.",
+                status=409,
+            )
+        logger.info(
+            "Idempotent replay for %s (key hash %s...) - not calling Sklik again.",
+            operation,
+            key[:12],
+        )
+        return key, entry.response
+
+
+def complete(cache_key: str, response: Any) -> None:
+    """Store the successful result so a retry replays it."""
+    with _lock:
+        entry = _store.get(cache_key)
+        if entry is None:
+            # Purged mid-flight (very long request vs. a short TTL). Recreate it
+            # rather than losing the record silently.
+            _store[cache_key] = _Entry(
+                response=response, created_at=time.time(), in_flight=False
+            )
+            logger.warning(
+                "Idempotency entry %s... vanished before completion; recreated.",
+                cache_key[:12],
+            )
+            return
+        entry.response = response
+        entry.in_flight = False
+
+
+def abandon(cache_key: str) -> None:
+    """Release a claimed key after a FAILED write, so the caller can retry.
+
+    A failed create must stay retryable: keeping the key would lock the caller
+    out of ever retrying with it.
+    """
+    with _lock:
+        entry = _store.get(cache_key)
+        if entry is not None and entry.in_flight:
+            del _store[cache_key]