Spatial Overlap Deduplication for Geospatial Webhooks

from shapely.geometry import shape
from shapely.ops import transform
from shapely.validation import make_valid
import pyproj
from pydantic import BaseModel, field_validator
from typing import Any

class SpatialWebhookPayload(BaseModel):
    event_id: str
    source_crs: str = "EPSG:4326"
    geometry: dict[str, Any]

    @field_validator("geometry")
    @classmethod
    def geometry_must_be_valid_geojson(cls, v: dict) -> dict:
        allowed = {"Point", "MultiPoint", "LineString", "MultiLineString",
                   "Polygon", "MultiPolygon", "GeometryCollection"}
        if v.get("type") not in allowed:
            raise ValueError(f"Unsupported geometry type: {v.get('type')}")
        return v


TARGET_CRS = "EPSG:4326"


def normalize_geometry(payload: SpatialWebhookPayload):
    """
    Return a valid Shapely geometry normalised to EPSG:4326.
    Projects from the payload's declared source CRS when needed.
    """
    geom = shape(payload.geometry)

    # Repair self-intersections and unclosed rings before any predicate
    if not geom.is_valid:
        geom = make_valid(geom)

    if payload.source_crs != TARGET_CRS:
        transformer = pyproj.Transformer.from_crs(
            payload.source_crs, TARGET_CRS, always_xy=True
        )
        geom = transform(transformer.transform, geom)

    return geom

import hashlib
from datetime import datetime, timezone

# Spatial grid precision in decimal degrees.
# Precision 4 = ~11 m grid cell; precision 3 = ~111 m grid cell.
GRID_PRECISION = 4
# Temporal window width in minutes (events within the same window share a bucket).
WINDOW_MINUTES = 60


def spatial_cache_key(geom, event_time: datetime | None = None) -> str:
    """
    Derive a deterministic cache key from a geometry's centroid grid cell
    and an hourly time bucket.
    """
    if event_time is None:
        event_time = datetime.now(tz=timezone.utc)

    centroid = geom.centroid
    grid_lat = round(centroid.y, GRID_PRECISION)
    grid_lon = round(centroid.x, GRID_PRECISION)
    # Bucket by hour (or subdivide by WINDOW_MINUTES for sub-hour windows)
    bucket = event_time.strftime("%Y%m%dT%H")

    raw = f"{grid_lat}:{grid_lon}:{bucket}"
    return "sodp:" + hashlib.sha256(raw.encode()).hexdigest()[:20]

import redis.asyncio as aioredis

CACHE_TTL_SECONDS = 3600  # 1-hour deduplication window


async def is_duplicate_via_cache(
    client: aioredis.Redis,
    key: str,
) -> bool:
    """
    Atomically claim the cache key.
    Returns True  → key already existed  → event is a duplicate.
    Returns False → key was absent       → event may be novel (proceed to predicate).
    """
    # SET key value NX EX ttl — atomic; no race between GET and SET
    acquired = await client.set(key, "1", nx=True, ex=CACHE_TTL_SECONDS)
    return not bool(acquired)

from shapely.strtree import STRtree
from shapely.geometry.base import BaseGeometry
import pyproj
from shapely.ops import transform as shp_transform

# Minimum fractional area overlap to treat two geometries as duplicates.
OVERLAP_THRESHOLD = 0.85

# Project to equal-area CRS for meaningful area ratios.
# EPSG:6933 covers the full globe; swap for a local UTM zone for higher accuracy.
_AREA_CRS_TRANSFORMER = pyproj.Transformer.from_crs(
    "EPSG:4326", "EPSG:6933", always_xy=True
)


def _project_for_area(geom: BaseGeometry) -> BaseGeometry:
    return shp_transform(_AREA_CRS_TRANSFORMER.transform, geom)


def is_duplicate_via_overlap(
    new_geom: BaseGeometry,
    candidates: list[BaseGeometry],
) -> bool:
    """
    Return True if any candidate overlaps new_geom by >= OVERLAP_THRESHOLD.
    All geometries must be in EPSG:4326; area comparison projects to EPSG:6933.
    """
    if not candidates:
        return False

    tree = STRtree(candidates)
    # R-tree pre-filter: only intersecting bounding boxes proceed to exact check
    hits = tree.query(new_geom, predicate="intersects")

    new_geom_area = _project_for_area(new_geom).area
    if new_geom_area == 0:
        # Point or line — fall back to exact-contains check
        return any(candidates[i].contains(new_geom) for i in hits)

    for idx in hits:
        candidate = candidates[idx]
        intersection = new_geom.intersection(candidate)
        if intersection.is_empty:
            continue
        overlap_ratio = _project_for_area(intersection).area / new_geom_area
        if overlap_ratio >= OVERLAP_THRESHOLD:
            return True

    return False

from fastapi import FastAPI, Request, BackgroundTasks, HTTPException
from fastapi.responses import JSONResponse
import redis.asyncio as aioredis
from contextlib import asynccontextmanager
import json

app = FastAPI()
redis_client: aioredis.Redis | None = None

@asynccontextmanager
async def lifespan(app_: FastAPI):
    global redis_client
    redis_client = aioredis.from_url("redis://localhost:6379", decode_responses=True)
    yield
    await redis_client.aclose()

app = FastAPI(lifespan=lifespan)

# In production, load recent event geometries from a PostGIS query or
# a shared in-memory spatial index refreshed by a background worker.
recent_geometries: list = []


async def process_spatial_event(raw_body: bytes) -> None:
    payload_dict = json.loads(raw_body)
    try:
        payload = SpatialWebhookPayload(**payload_dict)
    except Exception as exc:
        # Schema validation failure — route to DLQ
        print(f"[DLQ] invalid payload: {exc}")
        return

    geom = normalize_geometry(payload)
    key = spatial_cache_key(geom)

    if await is_duplicate_via_cache(redis_client, key):
        print(f"[DEDUP] cache hit: {key}")
        return

    if is_duplicate_via_overlap(geom, recent_geometries):
        print(f"[DEDUP] overlap threshold exceeded for event {payload.event_id}")
        # Persist audit record: original event_id, overlap ratio, canonical event ref
        return

    # Novel event — forward to downstream processor
    recent_geometries.append(geom)  # update in-memory index
    print(f"[PUBLISH] forwarding unique event {payload.event_id}")


@app.post("/webhook/spatial")
async def receive_spatial_event(request: Request, background_tasks: BackgroundTasks):
    body = await request.body()
    if not body:
        raise HTTPException(status_code=400, detail="Empty body")
    background_tasks.add_task(process_spatial_event, body)
    return JSONResponse({"status": "accepted"}, status_code=200)

from shapely.errors import TopologicalError, GEOSException

async def safe_normalize(payload: SpatialWebhookPayload):
    try:
        return normalize_geometry(payload)
    except (TopologicalError, GEOSException, ValueError) as exc:
        # Route to dead-letter queue with original payload and error context
        print(f"[DLQ] geometry error for event {payload.event_id}: {exc}")
        return None

import asyncio
import random

MAX_RETRIES = 4
BASE_DELAY_SECONDS = 0.25


async def publish_with_backoff(event_id: str, payload: dict) -> None:
    """
    Retry forwarding a novel event to the downstream processor with
    exponential backoff and full jitter to avoid retry storms.
    """
    for attempt in range(MAX_RETRIES):
        try:
            # Replace with your broker client (aiokafka, aio-pika, etc.)
            await downstream_publish(event_id, payload)
            return
        except Exception as exc:
            if attempt == MAX_RETRIES - 1:
                print(f"[DLQ] exhausted retries for {event_id}: {exc}")
                await send_to_dlq(event_id, payload, reason=str(exc))
                return
            delay = BASE_DELAY_SECONDS * (2 ** attempt) * random.random()
            await asyncio.sleep(delay)


async def downstream_publish(event_id: str, payload: dict) -> None:
    # Placeholder for aiokafka / aio-pika / SQS publish call
    pass


async def send_to_dlq(event_id: str, payload: dict, reason: str) -> None:
    # Persist to append-only DLQ with geometry snapshot and reason code
    pass

import asyncio
import pytest
from shapely.geometry import Polygon, mapping

# A base polygon covering a 0.01° × 0.01° cell
BASE = Polygon([(0, 0), (0.01, 0), (0.01, 0.01), (0, 0.01), (0, 0)])
# 90 % overlap — should be suppressed
NEAR_DUPE = Polygon([(0.001, 0.001), (0.011, 0.001), (0.011, 0.011), (0.001, 0.011), (0.001, 0.001)])
# Distant polygon — should be forwarded
DISTANT = Polygon([(1, 1), (1.01, 1), (1.01, 1.01), (1, 1.01), (1, 1)])


@pytest.mark.asyncio
async def test_overlap_suppression():
    """Events with >= 85 % area overlap are classified as duplicates."""
    result = is_duplicate_via_overlap(NEAR_DUPE, [BASE])
    assert result is True, "Near-duplicate polygon should be suppressed"


@pytest.mark.asyncio
async def test_distant_event_passes():
    """Events with no spatial intersection are classified as novel."""
    result = is_duplicate_via_overlap(DISTANT, [BASE])
    assert result is False, "Distant polygon should be forwarded"


@pytest.mark.asyncio
async def test_cache_deduplication(monkeypatch):
    """Second call with the same key returns duplicate=True."""
    seen = {}

    async def fake_set(key, val, *, nx, ex):
        if key in seen:
            return None  # NX fails — key exists
        seen[key] = val
        return True

    class FakeRedis:
        set = fake_set

    client = FakeRedis()
    key = "sodp:test-key-abc123"
    first = await is_duplicate_via_cache(client, key)
    second = await is_duplicate_via_cache(client, key)
    assert first is False, "First call should not be a duplicate"
    assert second is True, "Second call with same key should be a duplicate"