Feature Change Triggers for Geospatial Webhooks

import asyncpg

REPLICATION_DSN = "postgresql://user:pass@localhost/gis?replication=database"

async def stream_feature_changes(slot_name: str, publication: str):
    conn = await asyncpg.connect(dsn=REPLICATION_DSN)
    await conn.execute(
        "SELECT pg_create_logical_replication_slot($1, 'pgoutput')", slot_name
    )
    async with conn.replication(slot_name, output_plugin="pgoutput",
                                options={"publication_names": publication}) as stream:
        async for message in stream:
            yield parse_wal_message(message)  # returns dict with op, pk, old_geom, new_geom

from shapely.geometry import shape, mapping
from shapely.validation import explain_validity, make_valid
from shapely.ops import transform as shapely_transform
from shapely import set_precision
import pyproj

def normalize_geometry(geojson_geometry: dict, source_epsg: int = 4326) -> dict:
    """Validate, repair, reproject, and reduce precision of a GeoJSON geometry."""
    geom = shape(geojson_geometry)

    if not geom.is_valid:
        geom = make_valid(geom)
        if not geom.is_valid:
            raise ValueError(f"Irreparable topology: {explain_validity(geom)}")

    if source_epsg != 4326:
        transformer = pyproj.Transformer.from_crs(
            f"EPSG:{source_epsg}", "EPSG:4326", always_xy=True
        )
        # Use shapely.ops.transform, not the deprecated pyproj.transform
        geom = shapely_transform(transformer.transform, geom)

    # Snap coordinates to 7 decimal places (~1 cm precision at equator)
    geom = set_precision(geom, grid_size=1e-7)
    return mapping(geom)

from pydantic import BaseModel, field_validator
from typing import Literal

class GeoJSONGeometry(BaseModel):
    type: Literal["Point", "LineString", "Polygon", "MultiPolygon",
                  "MultiPoint", "MultiLineString", "GeometryCollection"]
    coordinates: list

    @field_validator("coordinates")
    @classmethod
    def coordinates_not_empty(cls, v: list) -> list:
        if not v:
            raise ValueError("coordinates must not be empty")
        return v

class SpatialFeatureEvent(BaseModel):
    feature_id: str
    operation: Literal["INSERT", "UPDATE", "DELETE"]
    geometry: GeoJSONGeometry
    crs: str = "EPSG:4326"
    properties: dict = {}

import uuid
import hashlib
import json
from datetime import datetime, timezone

def build_event_envelope(
    feature_event: SpatialFeatureEvent,
    source: str = "/postgis/features",
) -> dict:
    # Deterministic ID: same feature + operation + geometry hash → same UUID
    geom_hash = hashlib.sha256(
        json.dumps(feature_event.geometry.model_dump(), sort_keys=True).encode()
    ).hexdigest()[:16]
    event_id = str(uuid.uuid5(
        uuid.NAMESPACE_URL,
        f"{feature_event.feature_id}:{feature_event.operation}:{geom_hash}"
    ))

    geom_dict = feature_event.geometry.model_dump()

    return {
        "specversion": "1.0",
        "type": f"com.geospatialwebhook.feature.{feature_event.operation.lower()}",
        "source": source,
        "id": event_id,
        "time": datetime.now(timezone.utc).isoformat(),
        "datacontenttype": "application/geo+json",
        # Spatial CloudEvents extensions
        "crs": feature_event.crs,
        "spatialoperation": feature_event.operation,
        "data": {
            "type": "Feature",
            "geometry": geom_dict,
            "properties": {
                "feature_id": feature_event.feature_id,
                **feature_event.properties,
            },
        },
    }

import aiokafka
import json

async def publish_to_broker(
    envelope: dict,
    topic: str = "spatial-feature-events",
    partition_key: str | None = None,
) -> None:
    producer = aiokafka.AIOKafkaProducer(
        bootstrap_servers="localhost:9092",
        value_serializer=lambda v: json.dumps(v).encode("utf-8"),
        key_serializer=lambda k: k.encode("utf-8") if k else None,
        compression_type="gzip",  # spatial payloads compress well
    )
    await producer.start()
    try:
        key = partition_key or envelope.get("data", {}).get("properties", {}).get("feature_id")
        await producer.send_and_wait(topic, value=envelope, key=key)
    finally:
        await producer.stop()

import asyncio
import hashlib
import hmac
import json
import httpx
import random

def sign_payload(payload: dict, secret: str) -> str:
    body = json.dumps(payload, sort_keys=True, separators=(",", ":")).encode("utf-8")
    digest = hmac.new(secret.encode("utf-8"), body, hashlib.sha256).hexdigest()
    return f"sha256={digest}"

async def dispatch_webhook(
    url: str,
    envelope: dict,
    secret: str,
    max_retries: int = 5,
    base_delay: float = 1.0,
) -> int:
    headers = {
        "Content-Type": "application/json",
        "X-Webhook-Signature": sign_payload(envelope, secret),
        "X-Event-Id": envelope["id"],
    }
    async with httpx.AsyncClient(timeout=httpx.Timeout(10.0)) as client:
        for attempt in range(max_retries):
            try:
                resp = await client.post(url, json=envelope, headers=headers)
                resp.raise_for_status()
                return resp.status_code
            except httpx.HTTPStatusError as exc:
                if 400 <= exc.response.status_code < 500:
                    raise  # 4xx: client error, do not retry
                # 5xx: jittered exponential backoff
                delay = min(base_delay * (2 ** attempt), 60.0) + random.uniform(0, 1)
                await asyncio.sleep(delay)
            except (httpx.TimeoutException, httpx.ConnectError):
                delay = min(base_delay * (2 ** attempt), 60.0) + random.uniform(0, 1)
                await asyncio.sleep(delay)
    # Exhausted retries — caller should route to DLQ
    raise RuntimeError(f"Webhook delivery failed after {max_retries} attempts: {url}")

import asyncio
import pytest
from aiohttp import web

received: list[dict] = []

async def mock_consumer(request: web.Request) -> web.Response:
    received.append(await request.json())
    return web.Response(status=200)

@pytest.mark.asyncio
async def test_feature_change_triggers_webhook():
    app = web.Application()
    app.router.add_post("/hook", mock_consumer)
    runner = web.AppRunner(app)
    await runner.setup()
    site = web.TCPSite(runner, "localhost", 9999)
    await site.start()

    # Insert a test feature into PostGIS (requires a test DB)
    await insert_test_parcel(pg_dsn="postgresql://user:pass@localhost/test_gis")

    # Allow pipeline to propagate
    await asyncio.sleep(2)

    assert len(received) == 1
    env = received[0]
    assert env["type"].startswith("com.geospatialwebhook.feature.")
    assert env["specversion"] == "1.0"
    assert env["crs"] == "EPSG:4326"

    # Verify HMAC — re-sign with known secret and compare
    expected = sign_payload(env, secret="test-secret")
    assert env.get("_headers", {}).get("X-Webhook-Signature", "") == expected

    await runner.cleanup()