id: postgis-polygon-repair-quarantine-approval
namespace: company.data
labels:
team: data-platform
use-case: geospatial-quality
inputs:
- id: geojson
type: FILE
required: false
description: Optional GeoJSON file; omitted uses the four-feature example.
Webhooks accept geojson in the JSON body.
- id: dataset_id
type: STRING
defaults: service-zones-demo
description: Label attached to the published snapshot; publication never
replaces an existing dataset.
- id: mode
type: SELECT
values:
- PREVIEW
- REVIEW_AND_PUBLISH
defaults: PREVIEW
- id: allow_partial
type: BOOL
defaults: false
description: Allow publication excluding quarantined features; approval must
also acknowledge_partial.
variables:
sample_geojson: '{"type": "FeatureCollection", "features": [{"type": "Feature",
"id": "north", "properties": {"name": "Valid zone"}, "geometry": {"type":
"Polygon", "coordinates": [[[0, 0], [2, 0], [2, 2], [0, 2], [0, 0]]]}},
{"type": "Feature", "id": "central", "properties": {"name": "Bow tie"},
"geometry": {"type": "Polygon", "coordinates": [[[3, 0], [5, 2], [3, 2], [5,
0], [3, 0]]]}}, {"type": "Feature", "id": "south", "properties": {"name":
"Collapsed zone"}, "geometry": {"type": "Polygon", "coordinates": [[[6, 0],
[7, 0], [8, 0], [6, 0]]]}}, {"type": "Feature", "id": "west", "properties":
{"name": "Missing zone"}, "geometry": null}]}'
tasks:
- id: example_or_webhook
type: io.kestra.plugin.core.storage.Write
extension: .geojson
content: '{{ trigger.body.geojson is defined ? trigger.body.geojson | toJson :
vars.sample_geojson }}'
- id: source_size
type: io.kestra.plugin.core.storage.Size
uri: '{{ inputs.geojson ?? outputs.example_or_webhook.uri }}'
- id: bound_source
type: io.kestra.plugin.core.execution.Assert
conditions:
- '{{ outputs.source_size.size > 0 and outputs.source_size.size <= 1048576
}}'
- id: validate_input
type: io.kestra.plugin.scripts.python.Script
description: Validate the GeoJSON envelope, ring structure and resource limits;
PostGIS handles validity and repair.
containerImage: python:3.13-slim@sha256:3dd7cc108ec1493442514f5c2a871af6af0ec31d768ff6e378a93340c3b3db5f
taskRunner:
type: io.kestra.plugin.scripts.runner.docker.Docker
inputFiles:
source.geojson: '{{ inputs.geojson ?? outputs.example_or_webhook.uri }}'
outputFiles:
- normalized.json
script: |
import json
import math
from pathlib import Path
source = Path('source.geojson').read_bytes()
if not 0 < len(source) <= 1048576:
raise ValueError('Provide a GeoJSON file between 1 byte and 1 MiB')
document = json.loads(source)
if document.get('type') != 'FeatureCollection' or 'crs' in document:
raise ValueError('Use an RFC 7946 FeatureCollection without a CRS member')
features = document.get('features')
if not isinstance(features, list) or not 1 <= len(features) <= 100:
raise ValueError('Provide 1–100 features')
identifiers = set()
positions = 0
for feature in features:
identifier = feature.get('id')
if (feature.get('type') != 'Feature' or not isinstance(identifier, str)
or not 1 <= len(identifier) <= 128 or identifier in identifiers):
raise ValueError('Each Feature needs a unique string id (1–128 characters)')
identifiers.add(identifier)
properties = feature.get('properties')
if properties is not None and not isinstance(properties, dict):
raise ValueError('Feature properties must be an object or null')
geometry = feature.get('geometry')
if geometry is None:
continue # PostGIS staging records this feature as quarantined.
if not isinstance(geometry, dict) or 'crs' in geometry:
raise ValueError('Geometry must be an object without a CRS member')
if geometry.get('type') not in ('Polygon', 'MultiPolygon'):
continue # Unsupported types are quarantined without parsing them.
coordinates = geometry.get('coordinates')
polygons = [coordinates] if geometry['type'] == 'Polygon' else coordinates
if not isinstance(polygons, list) or not polygons:
raise ValueError('Polygon coordinates must contain rings')
for polygon in polygons:
if not isinstance(polygon, list) or not polygon:
raise ValueError('Each polygon must contain rings')
for ring in polygon:
if not isinstance(ring, list) or len(ring) < 4 or ring[0] != ring[-1]:
raise ValueError('Each ring needs at least four positions and must close')
for position in ring:
if (not isinstance(position, list) or len(position) != 2
or any(isinstance(n, bool) or not isinstance(n, (int, float))
or not math.isfinite(n) for n in position)):
raise ValueError('Use finite two-dimensional coordinates')
longitude, latitude = position
if not -180 <= longitude <= 180 or not -90 <= latitude <= 90:
raise ValueError('Coordinates exceed longitude/latitude bounds')
positions += 1
if positions > 10000:
raise ValueError('At most 10,000 coordinate positions are supported')
Path('normalized.json').write_text(json.dumps(document, allow_nan=False))
- id: initialize_schema
type: io.kestra.plugin.jdbc.postgresql.Queries
description: Create only two blueprint-owned tables; PostGIS must already be installed.
url: '{{ secret(''POSTGRES_JDBC_URL'') }}'
username: '{{ secret(''POSTGRES_USERNAME'') }}'
password: '{{ secret(''POSTGRES_PASSWORD'') }}'
sql: |
CREATE SCHEMA IF NOT EXISTS blueprint_polygon_review;
CREATE TABLE IF NOT EXISTS blueprint_polygon_review.stage (
batch_id text PRIMARY KEY, dataset_id text NOT NULL, report jsonb NOT NULL,
created_at timestamptz NOT NULL DEFAULT now()
);
CREATE TABLE IF NOT EXISTS blueprint_polygon_review.published (
batch_id text PRIMARY KEY, dataset_id text NOT NULL, features jsonb NOT NULL,
manifest_sha256 text NOT NULL, review_note text NOT NULL,
published_at timestamptz NOT NULL DEFAULT now()
);
retry:
type: constant
interval: PT2S
maxAttempts: 3
timeout: PT1M
- id: batch
type: io.kestra.plugin.core.flow.Sequential
tasks:
- id: stage
type: io.kestra.plugin.jdbc.postgresql.Query
url: '{{ secret(''POSTGRES_JDBC_URL'') }}'
username: '{{ secret(''POSTGRES_USERNAME'') }}'
password: '{{ secret(''POSTGRES_PASSWORD'') }}'
fetchType: FETCH_ONE
sql: |
WITH source AS (
SELECT feature, feature->>'id' AS id,
CASE WHEN feature->'geometry'->>'type' IN ('Polygon', 'MultiPolygon')
THEN ST_GeomFromGeoJSON(feature->'geometry') END AS geom
FROM jsonb_array_elements(CAST(:payload AS jsonb)->'features') AS feature
), repaired AS (
SELECT *, ST_MakeValid(geom) AS candidate FROM source
), classified AS (
SELECT *, COALESCE(NOT ST_IsEmpty(candidate) AND ST_IsValid(candidate)
AND ST_GeometryType(candidate) IN ('ST_Polygon', 'ST_MultiPolygon')
AND ST_NPoints(candidate) <= 20000, false) AS eligible
FROM repaired
), report AS (
SELECT jsonb_build_object('dataset_id', :dataset, 'features',
jsonb_agg(jsonb_build_object(
'id', id, 'original', feature,
'candidate', CASE WHEN eligible THEN jsonb_set(feature, '{geometry}',
ST_AsGeoJSON(candidate, 17)::jsonb) END,
'disposition', CASE WHEN NOT eligible THEN 'QUARANTINED'
WHEN ST_IsValid(geom) THEN 'UNCHANGED' ELSE 'REPAIRED' END,
'reason', CASE WHEN geom IS NULL THEN 'NULL_OR_UNSUPPORTED_GEOMETRY'
WHEN NOT eligible THEN 'REPAIR_OUTSIDE_POLYGON_CONTRACT'
ELSE ST_IsValidReason(geom) END
) ORDER BY id)) AS document FROM classified
), stored AS (
INSERT INTO blueprint_polygon_review.stage(batch_id, dataset_id, report)
SELECT :batch, :dataset, document FROM report
ON CONFLICT (batch_id) DO UPDATE SET batch_id = EXCLUDED.batch_id
RETURNING report
)
SELECT report::text AS report,
encode(sha256(convert_to(report::text, 'UTF8')), 'hex') AS manifest_sha256,
(SELECT count(*) FROM jsonb_array_elements(report->'features') AS f
WHERE f->>'disposition' <> 'QUARANTINED') AS eligible,
(SELECT count(*) FROM jsonb_array_elements(report->'features') AS f
WHERE f->>'disposition' = 'QUARANTINED') AS quarantined
FROM stored;
retry:
type: constant
interval: PT2S
maxAttempts: 3
timeout: PT1M
parameters:
payload: '{{ read(outputs.validate_input.outputFiles[''normalized.json'']) }}'
batch: '{{ execution.id }}'
dataset: '{{ inputs.dataset_id }}'
description: Repair with real ST_MakeValid; preserve originals and quarantine
entire non-polygon or oversized repairs.
- id: bound_report
type: io.kestra.plugin.core.execution.Assert
conditions:
- '{{ outputs.stage.row.report | length <= 1048576 }}'
- id: review_report
type: io.kestra.plugin.core.storage.Write
extension: .json
content: '{{ outputs.stage.row.report }}'
- id: publication_mode
type: io.kestra.plugin.core.flow.If
condition: '{{ inputs.mode == ''REVIEW_AND_PUBLISH'' and trigger.body is not
defined }}'
then:
- id: require_eligible
type: io.kestra.plugin.core.execution.Assert
conditions:
- '{{ outputs.stage.row.eligible > 0 }}'
- '{{ inputs.allow_partial or outputs.stage.row.quarantined == 0
}}'
- id: review
type: io.kestra.plugin.core.flow.Pause
description: Inspect the report before approving. Rejection is the default and
timeout fails after 24 hours.
behavior: FAIL
onResume:
- id: approved
type: BOOL
defaults: false
- id: reviewed_manifest_sha256
type: STRING
- id: acknowledge_partial
type: BOOL
defaults: false
- id: review_note
type: STRING
pauseDuration: PT24H
- id: approved
type: io.kestra.plugin.core.flow.If
condition: '{{ outputs.review.onResume.approved }}'
then:
- id: require_review
type: io.kestra.plugin.core.execution.Assert
conditions:
- '{{ outputs.review.onResume.review_note | trim | length > 0
}}'
- id: publish
type: io.kestra.plugin.jdbc.postgresql.Query
url: '{{ secret(''POSTGRES_JDBC_URL'') }}'
username: '{{ secret(''POSTGRES_USERNAME'') }}'
password: '{{ secret(''POSTGRES_PASSWORD'') }}'
fetchType: FETCH_ONE
sql: |
WITH approved AS (
SELECT batch_id, report->>'dataset_id' AS dataset_id, report,
encode(sha256(convert_to(report::text, 'UTF8')), 'hex') AS fingerprint
FROM blueprint_polygon_review.stage
WHERE batch_id = :batch AND created_at > now() - interval '24 hours'
FOR UPDATE
), snapshot AS (
SELECT batch_id, dataset_id, fingerprint,
jsonb_build_object('type', 'FeatureCollection', 'features',
jsonb_agg(f->'candidate' ORDER BY f->>'id')
FILTER (WHERE f->>'disposition' <> 'QUARANTINED')) AS features
FROM approved, jsonb_array_elements(report->'features') AS f
WHERE fingerprint = :fingerprint
GROUP BY batch_id, dataset_id, fingerprint
HAVING count(*) FILTER (WHERE f->>'disposition' <> 'QUARANTINED') > 0
AND (bool_and(f->>'disposition' <> 'QUARANTINED')
OR (CAST(:partial AS boolean) AND CAST(:ack AS boolean)))
)
INSERT INTO blueprint_polygon_review.published
(batch_id, dataset_id, features, manifest_sha256, review_note)
SELECT batch_id, dataset_id, features, fingerprint, :note FROM snapshot
ON CONFLICT (batch_id) DO UPDATE SET batch_id = EXCLUDED.batch_id
WHERE published.manifest_sha256 = EXCLUDED.manifest_sha256
RETURNING batch_id, dataset_id, manifest_sha256,
jsonb_array_length(features->'features') AS published_features;
retry:
type: constant
interval: PT2S
maxAttempts: 3
timeout: PT1M
parameters:
batch: '{{ execution.id }}'
fingerprint: '{{ outputs.review.onResume.reviewed_manifest_sha256 }}'
partial: '{{ inputs.allow_partial }}'
ack: '{{ outputs.review.onResume.acknowledge_partial }}'
note: '{{ outputs.review.onResume.review_note }}'
description: Atomically save an approved snapshot; recheck fingerprint, expiry
and partial acknowledgement in the database.
- id: require_published
type: io.kestra.plugin.core.execution.Assert
conditions:
- '{{ outputs.publish.row.published_features ==
outputs.stage.row.eligible }}'
- id: publication_receipt
type: io.kestra.plugin.core.storage.Write
extension: .json
content: '{{ outputs.publish.row | toJson }}'
else:
- id: rejected
type: io.kestra.plugin.core.log.Log
message: Review rejected; no snapshot published.
finally:
- id: cleanup_stage
type: io.kestra.plugin.jdbc.postgresql.Query
url: '{{ secret(''POSTGRES_JDBC_URL'') }}'
username: '{{ secret(''POSTGRES_USERNAME'') }}'
password: '{{ secret(''POSTGRES_PASSWORD'') }}'
fetchType: FETCH_ONE
sql: |
DELETE FROM blueprint_polygon_review.stage WHERE batch_id = :batch RETURNING batch_id;
retry:
type: constant
interval: PT2S
maxAttempts: 3
timeout: PT1M
parameters:
batch: '{{ execution.id }}'
description: Delete only this execution’s stage; retained reports and published
snapshots remain.
errors:
- id: failure
type: io.kestra.plugin.core.log.Log
level: ERROR
message: 'PostGIS review failed: {{ flow.namespace }}.{{ flow.id }}, execution
{{ execution.id }}. Inspect task errors; a saved publication remains
committed.'
triggers:
- id: preview_webhook
type: io.kestra.plugin.core.trigger.Webhook
key: '{{ secret(''POSTGIS_REVIEW_WEBHOOK_KEY'') }}'
description: POST a GeoJSON FeatureCollection in geojson; webhook executions
only preview.
outputs:
- id: report
type: FILE
value: '{{ outputs.review_report.uri ?? null }}'
- id: manifest_sha256
type: STRING
value: '{{ outputs.stage.row.manifest_sha256 ?? null }}'
- id: receipt
type: STRING
value: '{{ outputs.publication_receipt.uri ?? '''' }}'
