Using Redis to Cache Spatial Webhook Signatures

import hashlib
import hmac
import json
import os
from typing import Any

import redis
from redis.exceptions import RedisError

# ── Configuration ────────────────────────────────────────────────────────────
REDIS_URL      = os.getenv("REDIS_URL", "redis://localhost:6379/0")
WEBHOOK_SECRET = os.getenv("WEBHOOK_SECRET", "change-me")
# Set to 20–30 % above the provider's max redelivery window.
# E.g. provider retries for 24 h → use 108 000 s (30 h).
SIGNATURE_TTL  = int(os.getenv("SIGNATURE_TTL", "108000"))

# Thread-safe connection pool; socket_timeout evicts blocked calls quickly.
_redis = redis.Redis.from_url(
    REDIS_URL,
    decode_responses=True,
    socket_timeout=2,
    socket_connect_timeout=2,
)

# ── Spatial canonicalization ─────────────────────────────────────────────────
_TRANSIENT_KEYS = frozenset({"_meta", "received_at", "request_id", "delivery_attempt"})

def _round_coords(value: Any, precision: int = 6) -> Any:
    """Recursively round floats inside GeoJSON coordinate arrays (RFC 7946
    recommends 6 decimal places ≈ 0.11 m accuracy in EPSG:4326)."""
    if isinstance(value, float):
        return round(value, precision)
    if isinstance(value, list):
        return [_round_coords(v, precision) for v in value]
    if isinstance(value, dict):
        return {k: _round_coords(v, precision) for k, v in value.items()}
    return value

def normalize_geojson(payload: dict[str, Any]) -> str:
    """Return a canonical UTF-8 string suitable for deterministic hashing.

    Strips transient provider fields, rounds coordinate floats, and
    enforces sort_keys so key insertion order cannot affect the digest.
    """
    clean = {k: v for k, v in payload.items() if k not in _TRANSIENT_KEYS}
    clean = _round_coords(clean)
    # separators=(",", ":") removes all optional whitespace
    return json.dumps(clean, sort_keys=True, separators=(",", ":"))

# ── Signature computation ────────────────────────────────────────────────────
def compute_signature(canonical: str) -> str:
    """HMAC-SHA256 over the canonical payload string."""
    return hmac.new(
        WEBHOOK_SECRET.encode(),
        canonical.encode(),
        hashlib.sha256,
    ).hexdigest()

# ── Idempotency check ────────────────────────────────────────────────────────
def is_new_spatial_event(payload: dict[str, Any]) -> bool:
    """Return True if this event is new and should be processed.

    Uses a single atomic SET NX EX round-trip — no separate GET needed.
    Fails open on RedisError so a cache outage never stalls the pipeline;
    downstream spatial DB constraints remain the last-resort safety net.
    """
    canonical = normalize_geojson(payload)
    sig       = compute_signature(canonical)
    cache_key = f"swh:sig:{sig}"   # namespace prefix avoids key collisions

    try:
        # SET returns True (key was absent, now set) or None (key existed).
        result = _redis.set(cache_key, "1", nx=True, ex=SIGNATURE_TTL)
        return result is True
    except RedisError as exc:
        # Log for alerting but do not block ingestion.
        print(f"[idempotency] Redis unavailable, failing open: {exc}")
        return True

# ── Example webhook handler (FastAPI) ────────────────────────────────────────
# from fastapi import FastAPI, Request, Response
# app = FastAPI()
#
# @app.post("/webhook/spatial")
# async def receive_spatial_event(request: Request):
#     payload = await request.json()
#     if not is_new_spatial_event(payload):
#         return Response(status_code=200, content="duplicate ignored")
#     # Hand off to your spatial processing pipeline here.
#     return {"status": "accepted"}

import pytest
from unittest.mock import patch, MagicMock

# Isolate the module under test from a live Redis connection.
from your_module import is_new_spatial_event

SAMPLE_PAYLOAD = {
    "type": "Feature",
    "geometry": {"type": "Point", "coordinates": [-122.4194155, 37.7749295]},
    "properties": {"sensor_id": "gps-42"},
}

def _make_redis_mock(set_returns):
    mock = MagicMock()
    mock.set.side_effect = set_returns
    return mock

def test_new_event_returns_true():
    with patch("your_module._redis", _make_redis_mock([True])):
        assert is_new_spatial_event(SAMPLE_PAYLOAD) is True

def test_duplicate_event_returns_false():
    # First call claims the key; second call returns None (key already exists).
    with patch("your_module._redis", _make_redis_mock([True, None])):
        assert is_new_spatial_event(SAMPLE_PAYLOAD) is True
        assert is_new_spatial_event(SAMPLE_PAYLOAD) is False

def test_transient_fields_stripped_before_hashing():
    """Events identical except for 'received_at' must map to the same key."""
    payload_a = {**SAMPLE_PAYLOAD, "received_at": "2026-01-01T00:00:00Z"}
    payload_b = {**SAMPLE_PAYLOAD, "received_at": "2026-01-01T00:05:00Z"}
    with patch("your_module._redis", _make_redis_mock([True, None])):
        assert is_new_spatial_event(payload_a) is True
        assert is_new_spatial_event(payload_b) is False  # same spatial content

def test_redis_unavailable_fails_open():
    from redis.exceptions import RedisError
    mock = MagicMock()
    mock.set.side_effect = RedisError("connection refused")
    with patch("your_module._redis", mock):
        # Fail-open: event must be processed even when Redis is down.
        assert is_new_spatial_event(SAMPLE_PAYLOAD) is True