Best Practices for Spatial Event Payload Schemas

Spatial event payload schemas for tile update pipelines must enforce explicit CRS declarations (EPSG:4326), pre-computed bounding boxes, isolated routing metadata, and CloudEvents-wrapped envelopes — validated by Pydantic v2 at the ingress layer before any event reaches a message broker. This page is part of Tile Update Event Pipelines within Core Event Fundamentals & Architecture.

When to apply this schema pattern

This approach is appropriate over a minimal ad-hoc JSON structure in three situations:

  • You route events through a message broker (Kafka, RabbitMQ, AWS EventBridge) where infrastructure must make routing decisions without deserializing full coordinate arrays.
  • Multiple downstream consumers — tile renderers, spatial analytics engines, audit logs — each need a different slice of the payload, making field isolation critical to avoid tight coupling.
  • Your pipeline requires safe broker retries, meaning consumers must detect and skip duplicate events without replaying spatial index rebuilds or tile regeneration jobs.
Spatial event payload schema: five isolated domains Diagram showing a GeoJSON feature mutation entering the ingress validator, which produces a CloudEvents envelope containing five domains — Geometry, BBox, Temporal, Routing, Precision — before delivery to the message broker and downstream consumers. GIS Source feature edit Ingress Validator Pydantic v2 coord bounds CloudEvents v1.0 envelope Geometry type + coordinates BBox [min_lon … max_lat] Temporal event_time · version Routing tenant · priority · cid Precision 6–8 decimal places crs: EPSG:4326 explicit on every event Message Broker Kafka / SQS Tile renderer Spatial index Audit log

Schema anatomy: five isolated domains

Spatial event payloads break down in production when transport concerns bleed into geometric data. A resilient schema isolates five distinct domains:

  1. Geometry and projection. Default to WGS84 (EPSG:4326) with explicit [longitude, latitude] ordering. Always include a crs string even when using the default, and never rely on implicit axis ordering. The RFC 7946 GeoJSON specification mandates longitude-first ordering to prevent coordinate inversion across parsers. For payloads mixing projections, apply CRS normalization before serialization.

  2. Bounding box (bbox). Require a fixed 4-element array [min_lon, min_lat, max_lon, max_lat]. Consumers use this for rapid spatial indexing, tile routing, and bounding-box intersection tests without deserializing full coordinate arrays — essential for the sub-millisecond routing decisions described in Tile Update Event Pipelines.

  3. Temporal and versioning. Include event_time (ISO 8601 UTC), schema_version, and update_operation (insert, update, delete). This enables idempotent replay, out-of-order event handling, and schema drift detection. Pair these fields with a deterministic idempotency key as described in Event Key Generation for Spatial Data.

  4. Routing and context. Keep tenant_id, priority, source_system, and correlation_id outside the geometry object. Message brokers can route, throttle, or drop events based on these flat keys before incurring the CPU cost of parsing nested spatial structures.

  5. Precision control. Cap coordinate precision at 6–8 decimal places (~1–10 cm accuracy on WGS84). Excess precision inflates JSON payloads, increases network latency, and triggers unnecessary tile cache invalidations when floating-point noise shifts coordinates by sub-millimeter margins.

Complete runnable implementation

The Pydantic v2 model below enforces strict GeoJSON validation, auto-computes fallback bounding boxes from any geometry type, and serializes to a CloudEvents v1.0 envelope. It rejects invalid payloads at the class boundary rather than letting errors propagate into consumer workers.

python 
from pydantic import BaseModel, Field, field_validator, model_validator
from datetime import datetime, timezone
from typing import Literal, Optional
import uuid


class SpatialEventPayload(BaseModel):
    # Schema versioning — consumers gate on this before parsing geometry
    schema_version: Literal["1.0"] = "1.0"

    # Operational envelope — routable without touching geometry
    event_type: Literal["tile_update", "geometry_change", "attribute_sync"]
    update_operation: Literal["insert", "update", "delete"]
    correlation_id: str = Field(default_factory=lambda: str(uuid.uuid4()))
    tenant_id: Optional[str] = None
    priority: int = Field(default=5, ge=1, le=10)
    source_system: Optional[str] = None

    # Geometry domain — always GeoJSON, always explicit CRS
    geometry: dict
    crs: str = Field(default="EPSG:4326")

    # Pre-computed bbox: [min_lon, min_lat, max_lon, max_lat] — 4 elements exactly
    bbox: Optional[list[float]] = Field(default=None, min_length=4, max_length=4)

    # Temporal domain — ISO 8601 UTC, mandatory
    event_time: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))

    # Arbitrary attribute payload (properties, not geometry or routing)
    metadata: dict = Field(default_factory=dict)

    @field_validator("geometry")
    @classmethod
    def validate_geojson_structure(cls, v: dict) -> dict:
        valid_types = {
            "Point", "LineString", "Polygon",
            "MultiPoint", "MultiLineString", "MultiPolygon",
            "GeometryCollection",
        }
        if v.get("type") not in valid_types:
            raise ValueError(f"Invalid GeoJSON geometry type: {v.get('type')!r}")
        if "coordinates" not in v and v["type"] != "GeometryCollection":
            raise ValueError("GeoJSON object missing required 'coordinates' array")
        # Validate coordinate bounds: lon in [-180, 180], lat in [-90, 90]
        for lon, lat in cls._flatten_coordinates(v):
            if not (-180 <= lon <= 180):
                raise ValueError(f"Longitude {lon} out of WGS84 range [-180, 180]")
            if not (-90 <= lat <= 90):
                raise ValueError(f"Latitude {lat} out of WGS84 range [-90, 90]")
        return v

    @model_validator(mode="before")
    @classmethod
    def compute_bbox_if_missing(cls, data: dict) -> dict:
        """Auto-derive bbox from geometry coordinates if not supplied by producer."""
        if not data.get("bbox") and data.get("geometry"):
            coords = cls._flatten_coordinates(data["geometry"])
            if coords:
                lons = [c[0] for c in coords]
                lats = [c[1] for c in coords]
                # Round to 7 decimal places to match coordinate precision policy
                data["bbox"] = [
                    round(min(lons), 7),
                    round(min(lats), 7),
                    round(max(lons), 7),
                    round(max(lats), 7),
                ]
        return data

    @staticmethod
    def _flatten_coordinates(geom: dict) -> list[list[float]]:
        """Recursively extract all [lon, lat] pairs for bbox computation."""
        t = geom.get("type")
        c = geom.get("coordinates")
        if t == "Point":
            return [c]
        if t in {"LineString", "MultiPoint"}:
            return list(c)
        if t in {"Polygon", "MultiLineString"}:
            return [pt for ring in c for pt in ring]
        if t == "MultiPolygon":
            return [pt for poly in c for ring in poly for pt in ring]
        if t == "GeometryCollection":
            return [
                pt
                for g in geom.get("geometries", [])
                for pt in SpatialEventPayload._flatten_coordinates(g)
            ]
        return []

    def round_coordinates(self, precision: int = 7) -> "SpatialEventPayload":
        """
        Return a copy with coordinates rounded to `precision` decimal places.
        Call before to_cloudevent() to enforce the precision policy.
        """
        import json

        def round_coords(obj):
            if isinstance(obj, float):
                return round(obj, precision)
            if isinstance(obj, list):
                return [round_coords(x) for x in obj]
            if isinstance(obj, dict):
                return {k: round_coords(v) for k, v in obj.items()}
            return obj

        rounded_geom = round_coords(self.geometry)
        return self.model_copy(update={"geometry": rounded_geom})

    def to_cloudevent(self, source: str = "geospatial-ingest") -> dict:
        """
        Serialize to a CloudEvents v1.0 envelope.
        Brokers route on specversion / type / source / subject without touching 'data'.
        """
        return {
            "specversion": "1.0",
            "type": f"com.spatial.{self.event_type}",
            "source": source,
            "id": self.correlation_id,
            "time": self.event_time.isoformat(),
            "subject": self.tenant_id or "default",
            "datacontenttype": "application/json",
            "data": self.model_dump(mode="json"),
        }

Parameter reference

Field Type Constraint Default
schema_version Literal["1.0"] Must match consumer version gate "1.0"
event_type str (enum) tile_update, geometry_change, attribute_sync required
update_operation str (enum) insert, update, delete required
correlation_id str UUID4; used as CloudEvents id auto-generated
tenant_id str | None Routable flat key; must not be inside geometry None
priority int 1 (highest) to 10 (lowest) 5
geometry dict Valid RFC 7946 GeoJSON geometry object required
crs str Always include EPSG code, e.g. "EPSG:4326" "EPSG:4326"
bbox list[float] (4) [min_lon, min_lat, max_lon, max_lat]; auto-computed if absent auto
event_time datetime ISO 8601, timezone-aware UTC utcnow()
metadata dict Feature properties; never coordinates {}

Gotchas and spatial edge cases

  1. Coordinate axis ordering silently inverts across systems. The RFC 7946 GeoJSON format requires [longitude, latitude], but legacy WFS services and many GIS desktop exports emit [latitude, longitude] (EPSG geographic ordering). Always validate that the first coordinate component stays in [-180, 180] before accepting a payload — your field_validator above does this. For payloads arriving in non-WGS84 projections, apply CRS normalization strategies before the schema validator runs.

  2. Floating-point noise causes spurious tile invalidations. IEEE-754 representation drift can shift a coordinate by 1e-15 degrees between serializer versions. If you hash raw coordinates for cache keys or idempotency checks without rounding first, structurally identical geometries produce different hashes. Round to 7 decimal places before serialization; this is the responsibility of round_coordinates() in the model above.

  3. Polygon ring orientation is not enforced by JSON parsers. RFC 7946 requires exterior rings to be counter-clockwise and holes to be clockwise, but no standard JSON library checks this. Some spatial databases (PostGIS) silently reorient rings on insert; others (SQLite/SpatiaLite) do not. If downstream consumers compare geometries directly, mismatched orientations cause false non-equality. Use shapely.geometry.shape(geom).normalize() before serialization if ring orientation must be canonical.

  4. GeometryCollection breaks naive bbox computation. The coordinates key does not exist on GeometryCollection — child geometries live under geometries. The _flatten_coordinates method handles this recursively, but any third-party bbox utility you call must also support it; many do not.

  5. bbox that does not contain the geometry’s extents fails spatial routing. If a producer manually supplies a stale bbox from a cached feature and then edits the geometry, the bbox may no longer enclose the new shape. The model_validator auto-computes bbox only when the field is absent; it does not re-validate a supplied bbox against the geometry. Add an explicit containment check in production: assert bbox[0] <= min(lons) and bbox[2] >= max(lons).

  6. CloudEvents id must be unique per event, not per feature. Reusing the same correlation_id across retries is intentional for idempotency at the consumer. However, the broker may deduplicate based on id, which would silently drop legitimate re-deliveries. Use the correlation_id as the idempotency key inside your consumer’s Redis store; let the broker see unique id values if it performs producer-side deduplication.

Verification snippet

This pytest confirms that bbox auto-computation, coordinate rounding, and CloudEvents envelope structure all behave correctly against a representative GeoJSON polygon:

python 
import pytest
from datetime import timezone
from your_module import SpatialEventPayload  # replace with your import path


POLYGON_PAYLOAD = {
    "event_type": "tile_update",
    "update_operation": "update",
    "geometry": {
        "type": "Polygon",
        "coordinates": [[
            [-122.41941550000001, 37.77492950000001],
            [-122.41800000000000, 37.77492950000001],
            [-122.41800000000000, 37.77350000000000],
            [-122.41941550000001, 37.77350000000000],
            [-122.41941550000001, 37.77492950000001],
        ]]
    },
    "tenant_id": "acme-maps",
    "crs": "EPSG:4326",
}


def test_bbox_auto_computed():
    event = SpatialEventPayload(**POLYGON_PAYLOAD)
    assert event.bbox is not None, "bbox must be auto-computed when absent"
    assert len(event.bbox) == 4, "bbox must be a 4-element list"
    min_lon, min_lat, max_lon, max_lat = event.bbox
    assert min_lon <= max_lon, "bbox min_lon must not exceed max_lon"
    assert min_lat <= max_lat, "bbox min_lat must not exceed max_lat"


def test_coordinate_precision_capped():
    event = SpatialEventPayload(**POLYGON_PAYLOAD).round_coordinates(precision=7)
    # After rounding, no coordinate should have more than 7 decimal places
    for lon, lat in SpatialEventPayload._flatten_coordinates(event.geometry):
        assert len(str(lon).split(".")[-1]) <= 7, f"lon {lon} exceeds 7 decimal places"
        assert len(str(lat).split(".")[-1]) <= 7, f"lat {lat} exceeds 7 decimal places"


def test_cloudevent_envelope_structure():
    event = SpatialEventPayload(**POLYGON_PAYLOAD)
    envelope = event.to_cloudevent(source="test-ingest")
    assert envelope["specversion"] == "1.0"
    assert envelope["type"] == "com.spatial.tile_update"
    assert envelope["subject"] == "acme-maps"
    assert "data" in envelope
    assert envelope["data"]["crs"] == "EPSG:4326"


def test_out_of_bounds_longitude_rejected():
    bad = dict(POLYGON_PAYLOAD)
    bad["geometry"] = {
        "type": "Point",
        "coordinates": [200.0, 37.77]  # longitude 200 is invalid
    }
    with pytest.raises(ValueError, match="Longitude"):
        SpatialEventPayload(**bad)


def test_event_time_is_utc():
    event = SpatialEventPayload(**POLYGON_PAYLOAD)
    assert event.event_time.tzinfo is not None, "event_time must be timezone-aware"
    assert event.event_time.tzinfo == timezone.utc or str(event.event_time.tzinfo) in ("+00:00", "UTC")

FAQ

Why must I include a CRS field even when using the WGS84 default?

Implicit CRS assumptions break the moment a second data source enters the pipeline with a different projection. An explicit crs: "EPSG:4326" on every event eliminates axis-ordering ambiguity and makes payloads self-describing for consumers that apply a transformation before geometry comparison. The cost is a few bytes; the alternative is silent coordinate corruption when a producer in EPSG:3857 joins the topic. When mixed projections are unavoidable, normalize them at ingress per CRS Normalization Strategies.

How many decimal places should I use for GeoJSON coordinates?

Cap at 6–8 decimal places. Six decimals gives roughly 0.11 m horizontal accuracy on WGS84 (EPSG:4326), which is more than enough for tile invalidation. Excess precision inflates payload size, raises network latency, and triggers false cache invalidations when IEEE-754 noise shifts a coordinate by sub-millimeter amounts. The round_coordinates() method in the model above enforces this before serialization.

When should I send delta geometries instead of full feature state?

Use delta encoding for update operations on large polygons or multipolygons where only one ring changes — it keeps payloads small and brokers fast. Prefer full-state payloads for insert and delete, where consumers need the complete geometry to rebuild spatial indexes. Whichever you choose, set update_operation explicitly so each consumer can branch correctly rather than inferring intent from the geometry.

How do I detect duplicate spatial events from broker retries?

Derive a deterministic idempotency key from correlation_id + update_operation + a SHA-256 hash of the canonicalized geometry, then store it in Redis or DynamoDB with a TTL matching your retry window. Consumers skip processing if the key already exists. The canonicalization (coordinate rounding plus ring normalization) is what makes the hash stable across retries — see Generating Deterministic Idempotency Keys for GeoJSON Events.