Storing Failed Geometries in a PostGIS Dead-Letter Queue

CREATE TABLE geometry_dlq (
    id            bigserial PRIMARY KEY,
    work_key      text        NOT NULL,      -- the pipeline's idempotency key
    source_uri    text        NOT NULL,
    feature_id    text,                      -- the source's own identifier, if any
    -- Untyped, unvalidated, nullable ON PURPOSE. This column exists to hold the
    -- geometries that a geometry(MultiPolygon, 25832) NOT NULL column threw out.
    geom          geometry,
    geom_wkb      bytea,                     -- the raw bytes, even when geom will not parse
    declared_crs  text,                      -- what the source SAID, verbatim
    detected_crs  text,                      -- what the coordinate ranges suggest
    validity      text,                      -- ST_IsValidReason() output, or NULL
    failure_class text        NOT NULL,
    error_text    text,
    params        jsonb       NOT NULL DEFAULT '{}'::jsonb,
    attempts      integer     NOT NULL DEFAULT 1,
    failed_at     timestamptz NOT NULL DEFAULT now(),
    resolved_at   timestamptz,
    UNIQUE (work_key, feature_id)
);

-- A GiST index still works on a mixed-SRID column, but only within one SRID at a
-- time; triage queries must filter by declared_crs before using it spatially.
CREATE INDEX geometry_dlq_gist ON geometry_dlq USING gist (geom)
    WHERE geom IS NOT NULL AND resolved_at IS NULL;

CREATE INDEX geometry_dlq_open ON geometry_dlq (failure_class, failed_at)
    WHERE resolved_at IS NULL;

from __future__ import annotations

import json
from typing import Any, Mapping, Optional

import psycopg
import shapely
from shapely.errors import GEOSException, ShapelyError

INSERT = """
INSERT INTO geometry_dlq
       (work_key, source_uri, feature_id, geom, geom_wkb, declared_crs,
        detected_crs, validity, failure_class, error_text, params)
VALUES (%(work_key)s, %(source_uri)s, %(feature_id)s,
        CASE WHEN %(wkb)s IS NULL THEN NULL
             ELSE ST_SetSRID(ST_GeomFromWKB(%(wkb)s), %(srid)s) END,
        %(wkb)s, %(declared_crs)s, %(detected_crs)s, %(validity)s,
        %(failure_class)s, %(error_text)s, %(params)s::jsonb)
ON CONFLICT (work_key, feature_id) DO UPDATE
   SET attempts = geometry_dlq.attempts + 1,
       error_text = EXCLUDED.error_text,
       failed_at = now()
"""


def detect_crs(geom: Any) -> Optional[str]:
    """A cheap sanity read of the coordinate ranges — not a substitute for metadata."""
    try:
        minx, miny, maxx, maxy = geom.bounds
    except (AttributeError, ValueError):
        return None
    if -180.5 <= minx and maxx <= 180.5 and -90.5 <= miny and maxy <= 90.5:
        return "looks-geographic"
    if 100_000 <= minx and maxx <= 1_000_000 and 6_000_000 <= miny:
        return "looks-utm-northern"
    return "unknown"


def dead_letter_geometry(
    conn: psycopg.Connection,
    work_key: str,
    source_uri: str,
    feature_id: Optional[str],
    geom: Any,
    declared_crs: str,
    failure_class: str,
    error_text: str,
    params: Mapping[str, Any],
    srid: int = 0,
) -> None:
    """Store one failed feature with everything needed to reproduce it."""
    wkb: Optional[bytes] = None
    validity: Optional[str] = None
    try:
        wkb = shapely.to_wkb(geom)
        # is_valid_reason answers "why not", which is the column triage groups on.
        validity = None if shapely.is_valid(geom) else shapely.is_valid_reason(geom)
    except (GEOSException, ShapelyError, TypeError):
        # Unserialisable geometry: keep the row, lose only the geometry column.
        validity = "unserialisable"

    with conn.cursor() as cur:
        cur.execute(
            INSERT,
            {
                "work_key": work_key,
                "source_uri": source_uri,
                "feature_id": feature_id,
                "wkb": wkb,
                "srid": srid,                       # 0 = unknown, and that is allowed here
                "declared_crs": declared_crs,
                "detected_crs": detect_crs(geom),
                "validity": validity,
                "failure_class": failure_class,
                "error_text": error_text[:2000],
                "params": json.dumps(dict(params), sort_keys=True),
            },
        )

import shapely


def test_dlq_accepts_what_the_main_table_rejects(conn) -> None:
    # A classic bow-tie: valid WKB, invalid polygon.
    bowtie = shapely.from_wkt("POLYGON((0 0, 2 2, 2 0, 0 2, 0 0))")
    assert not shapely.is_valid(bowtie)

    dead_letter_geometry(
        conn, work_key="k1", source_uri="s3://drop/parcels.gpkg", feature_id="42",
        geom=bowtie, declared_crs="EPSG:25832", failure_class="invalid_geometry",
        error_text="ST_MakeValid produced an empty geometry", params={"nodata": None},
        srid=25832,
    )

    with conn.cursor() as cur:
        cur.execute("SELECT validity, ST_SRID(geom) FROM geometry_dlq WHERE work_key = 'k1'")
        validity, srid = cur.fetchone()
    assert "Self-intersection" in validity
    assert srid == 25832

-- What is broken, and is it one source or many?
SELECT failure_class, count(*), count(DISTINCT source_uri) AS sources,
       min(failed_at) AS first_seen
FROM   geometry_dlq
WHERE  resolved_at IS NULL
GROUP  BY failure_class
ORDER  BY 2 DESC;

-- Where is it broken? Failures that cluster in one municipality are a data
-- problem; failures spread evenly are a code problem.
SELECT m.name, count(*) AS failures
FROM   geometry_dlq d
JOIN   municipalities m ON ST_Intersects(m.geom, ST_Transform(d.geom, 25833))
WHERE  d.resolved_at IS NULL
  AND  d.declared_crs = 'EPSG:25832'      -- one SRID at a time on a mixed column
GROUP  BY m.name
ORDER  BY 2 DESC
LIMIT  10;