Idempotent Consumers for Out-of-Order Spatial Events

import asyncio
import json
from dataclasses import dataclass
from datetime import datetime, timezone

import asyncpg
import redis.asyncio as redis

# --- Event model -----------------------------------------------------------

@dataclass(frozen=True)
class FeatureEvent:
    feature_id: str          # stable identity of the real-world feature
    idempotency_key: str     # content digest, unique per exact payload
    observed_at: datetime    # SOURCE observation time (NOT arrival time)
    geometry: dict           # GeoJSON geometry, EPSG:4326 (WGS84), RFC 7946
    properties: dict         # non-spatial attributes


# --- Schema (run once) -----------------------------------------------------
# CREATE TABLE features (
#     feature_id  text PRIMARY KEY,
#     observed_at timestamptz NOT NULL,
#     geom        geometry(Geometry, 4326) NOT NULL,
#     properties  jsonb NOT NULL DEFAULT '{}'::jsonb,
#     updated_at  timestamptz NOT NULL DEFAULT now()
# );

UPSERT_SQL = """
INSERT INTO features (feature_id, observed_at, geom, properties)
VALUES ($1, $2, ST_SetSRID(ST_GeomFromGeoJSON($3), 4326), $4::jsonb)
ON CONFLICT (feature_id) DO UPDATE
    SET observed_at = EXCLUDED.observed_at,
        geom        = EXCLUDED.geom,
        properties  = EXCLUDED.properties,
        updated_at  = now()
    -- The guard: only overwrite when the incoming observation is STRICTLY
    -- newer. A stale (older) or tied event leaves the stored row untouched.
    WHERE features.observed_at < EXCLUDED.observed_at
RETURNING feature_id;
"""


async def is_duplicate(r: redis.Redis, key: str, ttl_seconds: int = 259200) -> bool:
    """Return True if this exact payload was already seen (at-least-once guard).

    SET NX is atomic: the first caller sets the key and gets True back from
    Redis; any concurrent or later redelivery finds the key present and is
    told it is a duplicate. TTL should cover the broker's redelivery window
    (72h default here).
    """
    # `set(..., nx=True)` returns True only if the key did NOT exist.
    was_set = await r.set(f"idem:{key}", "1", nx=True, ex=ttl_seconds)
    return not was_set


async def handle_event(pool: asyncpg.Pool, r: redis.Redis, ev: FeatureEvent) -> str:
    """Process one event idempotently and order-safely. Returns the outcome."""
    # 1. Drop exact redeliveries before touching the database.
    if await is_duplicate(r, ev.idempotency_key):
        return "duplicate"

    # 2. Conditional upsert: newest observation wins, regardless of arrival order.
    async with pool.acquire() as conn:
        row = await conn.fetchrow(
            UPSERT_SQL,
            ev.feature_id,
            ev.observed_at,
            json.dumps(ev.geometry),
            json.dumps(ev.properties),
        )
    # RETURNING yields a row only when the WHERE guard passed (insert or newer update).
    return "applied" if row is not None else "stale"


async def consume(pool: asyncpg.Pool, r: redis.Redis, queue: asyncio.Queue) -> None:
    """Main consumer loop. Acknowledge on every outcome so the broker
    does not redeliver a message we have already reasoned about."""
    while True:
        ev, ack = await queue.get()
        try:
            outcome = await handle_event(pool, r, ev)
            print(f"{ev.feature_id} @ {ev.observed_at.isoformat()} -> {outcome}")
            ack()  # duplicate, stale, and applied are all terminal — ack all three
        except Exception:
            # Do NOT ack: let the broker redeliver. The idempotency + version
            # guards make redelivery safe, so transient errors self-heal.
            raise
        finally:
            queue.task_done()


async def main() -> None:
    pool = await asyncpg.create_pool(dsn="postgresql://localhost/geo")
    r = redis.from_url("redis://localhost/0")
    queue: asyncio.Queue = asyncio.Queue()

    # Enqueue an OUT-OF-ORDER pair: the newer correction arrives FIRST.
    t_new = datetime(2026, 7, 13, 12, 0, 3, tzinfo=timezone.utc)
    t_old = datetime(2026, 7, 13, 12, 0, 0, tzinfo=timezone.utc)
    events = [
        FeatureEvent("parcel-42", "key-new", t_new,
                     {"type": "Point", "coordinates": [-73.9857, 40.7484]}, {"v": 2}),
        FeatureEvent("parcel-42", "key-old", t_old,
                     {"type": "Point", "coordinates": [-73.9800, 40.7400]}, {"v": 1}),
    ]
    for ev in events:
        await queue.put((ev, lambda: None))

    worker = asyncio.create_task(consume(pool, r, queue))
    await queue.join()
    worker.cancel()
    await pool.close()
    await r.aclose()


if __name__ == "__main__":
    asyncio.run(main())

import pytest
from datetime import datetime, timezone


class FakeFeatureStore:
    """Emulates the PostGIS conditional upsert: newest observed_at wins."""
    def __init__(self):
        self.rows = {}

    def upsert_if_newer(self, feature_id, observed_at, geom):
        existing = self.rows.get(feature_id)
        if existing is None or existing["observed_at"] < observed_at:
            self.rows[feature_id] = {"observed_at": observed_at, "geom": geom}
            return "applied"
        return "stale"      # strict '<' => older OR tied is skipped


class FakeIdem:
    def __init__(self):
        self.seen = set()

    def is_duplicate(self, key):
        if key in self.seen:
            return True
        self.seen.add(key)
        return False


def process(store, idem, feature_id, key, observed_at, geom):
    if idem.is_duplicate(key):
        return "duplicate"
    return store.upsert_if_newer(feature_id, observed_at, geom)


T_OLD = datetime(2026, 7, 13, 12, 0, 0, tzinfo=timezone.utc)
T_NEW = datetime(2026, 7, 13, 12, 0, 3, tzinfo=timezone.utc)
GEOM_OLD = {"type": "Point", "coordinates": [-73.9800, 40.7400]}
GEOM_NEW = {"type": "Point", "coordinates": [-73.9857, 40.7484]}


def test_out_of_order_converges_to_newest():
    """Newer event arrives first, stale event second: newest geometry must win."""
    store, idem = FakeFeatureStore(), FakeIdem()
    assert process(store, idem, "p-42", "k2", T_NEW, GEOM_NEW) == "applied"
    assert process(store, idem, "p-42", "k1", T_OLD, GEOM_OLD) == "stale"
    assert store.rows["p-42"]["geom"] == GEOM_NEW      # stale did NOT overwrite


def test_in_order_also_converges():
    """Same two events in natural order converge to the same final state."""
    store, idem = FakeFeatureStore(), FakeIdem()
    assert process(store, idem, "p-42", "k1", T_OLD, GEOM_OLD) == "applied"
    assert process(store, idem, "p-42", "k2", T_NEW, GEOM_NEW) == "applied"
    assert store.rows["p-42"]["geom"] == GEOM_NEW


def test_exact_redelivery_is_dropped():
    """At-least-once: the same key twice is a no-op the second time."""
    store, idem = FakeFeatureStore(), FakeIdem()
    assert process(store, idem, "p-42", "k2", T_NEW, GEOM_NEW) == "applied"
    assert process(store, idem, "p-42", "k2", T_NEW, GEOM_NEW) == "duplicate"


def test_tie_does_not_overwrite():
    """Equal observation times: first write wins, tie is skipped."""
    store, idem = FakeFeatureStore(), FakeIdem()
    process(store, idem, "p-42", "a", T_NEW, GEOM_NEW)
    assert process(store, idem, "p-42", "b", T_NEW, GEOM_OLD) == "stale"
    assert store.rows["p-42"]["geom"] == GEOM_NEW