Conflict Resolution Strategies for Geospatial Webhooks

from pyproj import Transformer
from shapely.geometry import shape, mapping
from shapely.validation import make_valid
from typing import Any

_transformer_cache: dict[str, Transformer] = {}

def _get_transformer(source_epsg: int) -> Transformer:
    key = f"{source_epsg}:4326"
    if key not in _transformer_cache:
        _transformer_cache[key] = Transformer.from_crs(
            source_epsg, 4326, always_xy=True
        )
    return _transformer_cache[key]

def normalize_geometry(geom_dict: dict[str, Any], source_epsg: int = 4326) -> dict[str, Any]:
    """Return a valid GeoJSON geometry dict in EPSG:4326."""
    geom = make_valid(shape(geom_dict))
    if source_epsg != 4326:
        transformer = _get_transformer(source_epsg)
        # shapely.ops.transform applies the projection coordinate-wise
        from shapely.ops import transform as shapely_transform
        geom = make_valid(shapely_transform(transformer.transform, geom))
    return mapping(geom)

import hashlib
import json
from redis.asyncio import Redis

async def is_duplicate(redis: Redis, payload: dict[str, Any]) -> bool:
    """Return True if this event was already processed."""
    # Sort keys for a stable hash across providers that reorder JSON fields
    event_key = hashlib.sha256(
        json.dumps(payload, sort_keys=True).encode()
    ).hexdigest()
    idem_key = f"idem:{event_key}"
    # SET NX returns None if key exists, "OK" if newly set
    registered = await redis.set(idem_key, "1", nx=True, ex=300)
    return registered is None  # None means the key was already present

from shapely.geometry import shape
from shapely.validation import make_valid

def compute_overlap(
    current_geom_dict: dict[str, Any],
    incoming_geom_dict: dict[str, Any]
) -> bool:
    """Return True when the two geometries share area — trigger resolution."""
    current = make_valid(shape(current_geom_dict))
    incoming = make_valid(shape(incoming_geom_dict))
    # Bounding-box envelope check is O(1); skip full intersection if boxes miss
    if not current.envelope.intersects(incoming.envelope):
        return False
    return current.intersects(incoming)

from shapely.geometry import shape, mapping
from shapely.validation import make_valid

def resolve_conflict(
    current_geom_dict: dict[str, Any],
    current_ts: float,
    incoming_geom_dict: dict[str, Any],
    incoming_ts: float,
    strategy: str = "lww",
) -> dict[str, Any]:
    """
    Resolve a spatial conflict between two geometry versions.

    strategy: "lww" (last-write-wins) or "merge" (spatial union).
    Returns the winning GeoJSON geometry dict.
    """
    current = make_valid(shape(current_geom_dict))
    incoming = make_valid(shape(incoming_geom_dict))

    if strategy == "merge":
        # Union preserves both edits — use for additive / collaborative domains
        merged = make_valid(current.union(incoming))
        return mapping(merged)

    # Default: last-write-wins on timestamp
    if incoming_ts >= current_ts:
        return incoming_geom_dict
    return current_geom_dict

import asyncio
from redis.asyncio import Redis

async def acquire_lock(redis: Redis, feature_id: str, ttl_seconds: int = 10) -> bool:
    """Attempt atomic lock acquisition. Returns False if already locked."""
    lock_key = f"lock:feature:{feature_id}"
    result = await redis.set(lock_key, "1", nx=True, ex=ttl_seconds)
    return result is not None

async def release_lock(redis: Redis, feature_id: str) -> None:
    await redis.delete(f"lock:feature:{feature_id}")

import asyncio
from fastapi import FastAPI, HTTPException, Request
from redis.asyncio import Redis

app = FastAPI()
redis_client = Redis.from_url("redis://localhost:6379/0", decode_responses=True)

@app.post("/webhooks/spatial-update")
async def handle_spatial_webhook(request: Request):
    payload: dict = await request.json()
    feature_id: str = payload["feature_id"]

    # Step 1: normalize CRS to EPSG:4326
    payload["geometry"] = normalize_geometry(
        payload["geometry"],
        source_epsg=payload.get("crs_epsg", 4326)
    )

    # Step 2: idempotency gate
    if await is_duplicate(redis_client, payload):
        return {"status": "duplicate_acknowledged"}

    # Step 3: check for spatial overlap with current persisted state
    current_state = await fetch_from_db(feature_id)  # returns {"geometry": ..., "updated_at": ...}
    has_conflict = compute_overlap(current_state["geometry"], payload["geometry"])

    if not has_conflict:
        # Disjoint update — no resolution required
        await commit_to_db(feature_id, payload["geometry"], payload["updated_at"])
        return {"status": "applied_no_conflict"}

    # Step 4: apply resolution policy
    resolved_geom = resolve_conflict(
        current_state["geometry"],
        current_state["updated_at"],
        payload["geometry"],
        payload["updated_at"],
        strategy=payload.get("resolution_hint", "lww"),
    )

    # Step 5: acquire lock before writing
    locked = await acquire_lock(redis_client, feature_id)
    if not locked:
        raise HTTPException(status_code=429, detail="Feature locked — retry later")

    try:
        await commit_to_db(feature_id, resolved_geom, payload["updated_at"])
        await emit_audit_event(feature_id, payload, current_state, resolved_geom)
    finally:
        await release_lock(redis_client, feature_id)

    return {"status": "resolved_and_committed"}

from pydantic import BaseModel, field_validator
from shapely.geometry import shape
from shapely.validation import make_valid, explain_validity

class SpatialWebhookPayload(BaseModel):
    feature_id: str
    geometry: dict
    updated_at: float
    crs_epsg: int = 4326
    resolution_hint: str = "lww"

    @field_validator("geometry")
    @classmethod
    def geometry_must_be_valid_geojson(cls, v: dict) -> dict:
        allowed = {"Point", "LineString", "Polygon", "MultiPolygon",
                   "MultiPoint", "MultiLineString", "GeometryCollection"}
        if v.get("type") not in allowed:
            raise ValueError(f"Unknown geometry type: {v.get('type')}")
        try:
            geom = shape(v)
        except Exception as exc:
            raise ValueError(f"Cannot parse geometry: {exc}") from exc
        if not geom.is_valid:
            explanation = explain_validity(geom)
            # Attempt auto-repair; raise only if repair fails
            repaired = make_valid(geom)
            if not repaired.is_valid:
                raise ValueError(f"Geometry invalid and unrepairable: {explanation}")
        return v

import asyncio
import random

async def acquire_lock_with_backoff(
    redis: Redis,
    feature_id: str,
    max_attempts: int = 5,
    base_delay_ms: int = 50,
    ttl_seconds: int = 10,
) -> bool:
    """
    Exponential backoff with full jitter for distributed lock acquisition.
    Returns True if the lock was acquired within max_attempts.
    """
    for attempt in range(max_attempts):
        if await acquire_lock(redis, feature_id, ttl_seconds):
            return True
        # Full jitter: sleep between 0 and 2^attempt * base_delay_ms
        ceiling_ms = (2 ** attempt) * base_delay_ms
        delay_s = random.uniform(0, ceiling_ms) / 1000.0
        await asyncio.sleep(delay_s)
    return False

import asyncio
import pytest
from unittest.mock import AsyncMock, patch
from shapely.geometry import mapping
from shapely.geometry import Polygon

# Two overlapping squares in EPSG:4326
CURRENT_GEOM = mapping(Polygon([(0, 0), (1, 0), (1, 1), (0, 1)]))
INCOMING_GEOM = mapping(Polygon([(0.5, 0.5), (1.5, 0.5), (1.5, 1.5), (0.5, 1.5)]))

def test_overlap_detected():
    from your_module import compute_overlap
    assert compute_overlap(CURRENT_GEOM, INCOMING_GEOM) is True

def test_lww_incoming_wins():
    from your_module import resolve_conflict
    result = resolve_conflict(CURRENT_GEOM, 1000.0, INCOMING_GEOM, 1001.0, strategy="lww")
    assert result == INCOMING_GEOM

def test_lww_current_wins():
    from your_module import resolve_conflict
    result = resolve_conflict(CURRENT_GEOM, 1002.0, INCOMING_GEOM, 1001.0, strategy="lww")
    assert result == CURRENT_GEOM

def test_merge_union_contains_both():
    from your_module import resolve_conflict
    from shapely.geometry import shape
    result = resolve_conflict(CURRENT_GEOM, 1000.0, INCOMING_GEOM, 1001.0, strategy="merge")
    merged = shape(result)
    assert shape(CURRENT_GEOM).within(merged)
    assert shape(INCOMING_GEOM).within(merged)

@pytest.mark.asyncio
async def test_duplicate_is_short_circuited(redis_mock):
    from your_module import is_duplicate
    payload = {"feature_id": "f1", "geometry": INCOMING_GEOM, "updated_at": 1000.0}
    # First call should register the key
    assert await is_duplicate(redis_mock, payload) is False
    # Second call with same payload should detect duplicate
    assert await is_duplicate(redis_mock, payload) is True