Implementing At-Least-Once Delivery for GIS Webhooks

"""
at_least_once_gis.py — Production at-least-once GIS webhook receiver.

Dependencies: fastapi, uvicorn, redis[asyncio], shapely, pyproj, pydantic
"""

import asyncio
import hashlib
import json
import logging
import math
import random
from typing import Any

import redis.asyncio as aioredis
from fastapi import BackgroundTasks, FastAPI, HTTPException, Request
from pydantic import BaseModel, field_validator
from shapely.geometry import shape
from shapely.validation import explain_validity

# ---------------------------------------------------------------------------
# Configuration
# ---------------------------------------------------------------------------
IDEMPOTENCY_TTL_SECONDS = 86_400        # 24-hour retention for processed keys
DLQ_TTL_SECONDS = 604_800              # 7-day retention for dead-letter entries
MAX_RETRIES = 5
BASE_BACKOFF_SECONDS = 2.0
COORDINATE_PRECISION = 6               # ~11 cm; sufficient for all GIS use-cases

app = FastAPI(title="GIS Webhook Receiver")
redis_client = aioredis.Redis(host="localhost", port=6379, db=0, decode_responses=True)
logger = logging.getLogger("gis.at_least_once")

# ---------------------------------------------------------------------------
# Payload schema (GeoJSON Feature, strict)
# ---------------------------------------------------------------------------
class GeoJSONFeaturePayload(BaseModel):
    """Minimal GeoJSON Feature schema for webhook ingestion."""
    type: str
    geometry: dict[str, Any]
    properties: dict[str, Any] | None = None

    @field_validator("type")
    @classmethod
    def must_be_feature(cls, v: str) -> str:
        if v != "Feature":
            raise ValueError(f"Expected GeoJSON 'Feature', got '{v}'")
        return v

# ---------------------------------------------------------------------------
# Idempotency key derivation
# ---------------------------------------------------------------------------
def _round_coordinates(obj: Any) -> Any:
    """Recursively round all floats to COORDINATE_PRECISION decimal places.

    Floating-point serialization varies across sender libraries — 40.7128 and
    40.712800000001 are geographically identical but produce different SHA-256
    hashes. Rounding before hashing ensures duplicate payloads yield the same key.
    """
    if isinstance(obj, float):
        return round(obj, COORDINATE_PRECISION)
    if isinstance(obj, list):
        return [_round_coordinates(v) for v in obj]
    if isinstance(obj, dict):
        return {k: _round_coordinates(v) for k, v in sorted(obj.items())}
    return obj

def derive_idempotency_key(payload: dict[str, Any], event_id: str | None) -> str:
    """Return a Redis key that uniquely identifies this spatial event.

    Priority:
    1. X-Event-ID or Idempotency-Key header (sender-assigned, most reliable).
    2. SHA-256 hash of the coordinate-normalized, key-sorted payload JSON
       (used when the sender provides no event identifier).
    """
    if event_id:
        return f"idemp:{event_id}"

    normalized = json.dumps(_round_coordinates(payload), sort_keys=True, separators=(",", ":"))
    digest = hashlib.sha256(normalized.encode()).hexdigest()
    return f"idemp:{digest}"

# ---------------------------------------------------------------------------
# Backoff strategy
# ---------------------------------------------------------------------------
def full_jitter_backoff(attempt: int) -> float:
    """Full jitter: uniform sample in [0, min(cap, base * 2^attempt)].

    Full jitter outperforms equal-jitter and decorrelated jitter at preventing
    thundering-herd problems when many retrying workers recover simultaneously
    after a PostGIS or network outage.
    """
    cap = 60.0
    ceiling = min(cap, BASE_BACKOFF_SECONDS * math.pow(2, attempt))
    return random.uniform(0.0, ceiling)

# ---------------------------------------------------------------------------
# Geometry validation
# ---------------------------------------------------------------------------
def validate_geometry(geometry_dict: dict[str, Any]) -> None:
    """Raise ValueError for invalid or degenerate geometries before any DB write.

    Inserting an invalid polygon into PostGIS can corrupt spatial indexes and
    cause ST_Intersects / ST_Contains queries to return wrong results indefinitely.
    Validate with shapely before the geometry touches any spatial store.
    """
    try:
        geom = shape(geometry_dict)
    except Exception as exc:
        raise ValueError(f"Cannot parse geometry: {exc}") from exc

    if not geom.is_valid:
        raise ValueError(f"Invalid geometry: {explain_validity(geom)}")

    if geom.is_empty:
        raise ValueError("Geometry is empty — zero-area or zero-length feature rejected")

# ---------------------------------------------------------------------------
# Background worker
# ---------------------------------------------------------------------------
async def process_spatial_event(payload: dict[str, Any], event_id: str | None) -> None:
    """Deduplication + bounded-retry spatial processor.

    1. Derive the idempotency key.
    2. Atomic SET NX EX — skip immediately if key already exists (duplicate).
    3. Validate geometry before any spatial write.
    4. Attempt processing with exponential backoff and full jitter.
    5. On exhaustion, route to the dead-letter queue and release the key so
       manual replay is possible.
    """
    key = derive_idempotency_key(payload, event_id)

    # Atomic check-and-set: only ONE worker proceeds per unique event
    acquired = await redis_client.set(key, "processing", nx=True, ex=IDEMPOTENCY_TTL_SECONDS)
    if not acquired:
        logger.info("Duplicate spatial event skipped: %s", key)
        return

    # Validate geometry once, outside the retry loop — malformed geometries
    # will not succeed on retry, so fail fast and route directly to DLQ.
    try:
        validate_geometry(payload.get("geometry", {}))
    except ValueError as exc:
        logger.error("Geometry validation failed for %s: %s — routing to DLQ", key, exc)
        dlq_key = f"dlq:{key}"
        dlq_value = json.dumps({"payload": payload, "error": str(exc), "retries": 0})
        await redis_client.set(dlq_key, dlq_value, ex=DLQ_TTL_SECONDS)
        await redis_client.delete(key)
        return

    for attempt in range(MAX_RETRIES):
        try:
            # -------------------------------------------------------------------
            # SPATIAL PROCESSING — replace with your actual implementation:
            #   - asyncpg INSERT to PostGIS with ST_GeomFromGeoJSON
            #   - Shapely-based topology merge
            #   - Tile cache invalidation via HTTP to your tile server
            #   - Routing graph update in Neo4j / pgRouting
            # -------------------------------------------------------------------
            await asyncio.sleep(0.0)  # Remove; placeholder for real async call
            logger.info("Spatial event processed successfully: %s", key)
            return  # Success — idempotency key remains set in Redis as proof

        except Exception as exc:
            wait = full_jitter_backoff(attempt)
            logger.warning(
                "Attempt %d/%d failed for %s — retrying in %.2fs. Error: %s",
                attempt + 1, MAX_RETRIES, key, wait, exc,
            )
            await asyncio.sleep(wait)

    # Retries exhausted — move to dead-letter queue
    logger.error("Event %s exhausted %d retries. Routing to DLQ.", key, MAX_RETRIES)
    dlq_key = f"dlq:{key}"
    dlq_value = json.dumps({"payload": payload, "retries": MAX_RETRIES})
    await redis_client.set(dlq_key, dlq_value, ex=DLQ_TTL_SECONDS)

    # Delete processing key so the event can be replayed manually via DLQ consumer
    await redis_client.delete(key)

# ---------------------------------------------------------------------------
# HTTP endpoint
# ---------------------------------------------------------------------------
@app.post("/webhooks/gis", status_code=200)
async def receive_gis_webhook(
    request: Request,
    background_tasks: BackgroundTasks,
) -> dict[str, str]:
    """Accept a GeoJSON Feature webhook and acknowledge before processing.

    The immediate 200 OK stops the sender's retry timer. All spatial work
    happens in the background, decoupled from the HTTP response path.
    """
    try:
        raw = await request.json()
        payload = GeoJSONFeaturePayload(**raw).model_dump()
    except Exception as exc:
        raise HTTPException(status_code=400, detail=f"Schema validation failed: {exc}")

    # Prefer sender-assigned event identifier for stable deduplication
    event_id = (
        request.headers.get("X-Event-ID")
        or request.headers.get("Idempotency-Key")
    )

    background_tasks.add_task(process_spatial_event, payload, event_id)
    return {"status": "accepted"}

"""
test_at_least_once.py — Verify idempotency and DLQ routing end-to-end.

Run: pytest test_at_least_once.py -v
"""

import asyncio
import hashlib
import json
import pytest
import redis.asyncio as aioredis

from at_least_once_gis import (
    COORDINATE_PRECISION,
    derive_idempotency_key,
    full_jitter_backoff,
    validate_geometry,
    _round_coordinates,
)

# ---------------------------------------------------------------------------
# Unit: idempotency key stability
# ---------------------------------------------------------------------------
def test_coordinate_rounding_produces_stable_key():
    """Two payloads differing only in float noise must yield the same key."""
    payload_a = {
        "type": "Feature",
        "geometry": {"type": "Point", "coordinates": [13.404954, 52.520008]},
        "properties": {},
    }
    payload_b = {
        "type": "Feature",
        "geometry": {"type": "Point", "coordinates": [13.4049540000001, 52.5200080000001]},
        "properties": {},
    }
    key_a = derive_idempotency_key(payload_a, None)
    key_b = derive_idempotency_key(payload_b, None)
    assert key_a == key_b, "Coordinate noise must not produce distinct idempotency keys"

def test_event_id_header_takes_priority():
    """When X-Event-ID is provided it overrides the hash-based key."""
    payload = {"type": "Feature", "geometry": {"type": "Point", "coordinates": [0, 0]}}
    key = derive_idempotency_key(payload, "evt-abc-123")
    assert key == "idemp:evt-abc-123"

# ---------------------------------------------------------------------------
# Unit: geometry validation
# ---------------------------------------------------------------------------
def test_valid_point_passes():
    validate_geometry({"type": "Point", "coordinates": [13.4050, 52.5200]})  # no exception

def test_self_intersecting_polygon_rejected():
    # Bowtie polygon — invalid topology
    bowtie = {
        "type": "Polygon",
        "coordinates": [[[0, 0], [2, 2], [2, 0], [0, 2], [0, 0]]],
    }
    with pytest.raises(ValueError, match="Invalid geometry"):
        validate_geometry(bowtie)

def test_empty_geometry_rejected():
    from shapely.wkt import loads
    # Pass an empty geometry dict (missing type)
    with pytest.raises((ValueError, KeyError)):
        validate_geometry({})

# ---------------------------------------------------------------------------
# Unit: backoff stays within bounds
# ---------------------------------------------------------------------------
def test_backoff_never_exceeds_cap():
    for attempt in range(20):
        assert full_jitter_backoff(attempt) <= 60.0

def test_backoff_is_non_negative():
    for attempt in range(10):
        assert full_jitter_backoff(attempt) >= 0.0

# ---------------------------------------------------------------------------
# Integration: duplicate suppression via Redis
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_duplicate_event_suppressed():
    """Second call with the same key must be a no-op (key already set)."""
    r = aioredis.Redis(host="localhost", port=6379, db=15, decode_responses=True)
    key = "idemp:test-dedup-001"
    await r.delete(key)

    first = await r.set(key, "processing", nx=True, ex=60)
    second = await r.set(key, "processing", nx=True, ex=60)

    assert first is True,  "First acquisition must succeed"
    assert second is None, "Duplicate acquisition must be rejected by SET NX"

    await r.delete(key)
    await r.aclose()