id: postgis-point-to-zone-enrichment
namespace: company.data
inputs:
- id: points
type: FILE
description: RFC 7946 FeatureCollection of 1-1000 Point features with unique
string feature.id values.
- id: zones
type: FILE
description: RFC 7946 FeatureCollection of 1-100 Polygon/MultiPolygon features
with unique string feature.id values.
tasks:
- id: input_sizes
type: io.kestra.plugin.core.flow.Parallel
tasks:
- id: points_size
type: io.kestra.plugin.core.storage.Size
uri: '{{ inputs.points }}'
- id: zones_size
type: io.kestra.plugin.core.storage.Size
uri: '{{ inputs.zones }}'
- id: bound_inputs
type: io.kestra.plugin.core.execution.Assert
conditions:
- '{{ outputs.points_size.size > 0 and outputs.points_size.size <= 1048576
}}'
- '{{ outputs.zones_size.size > 0 and outputs.zones_size.size <= 1048576
}}'
- id: stage_inputs
type: io.kestra.plugin.core.flow.Parallel
description: 'Copy file URI inputs into this execution: read() cannot read files
owned by a producer execution.'
tasks:
- id: stage_points
type: io.kestra.plugin.core.storage.Concat
files:
- '{{ inputs.points }}'
- id: stage_zones
type: io.kestra.plugin.core.storage.Concat
files:
- '{{ inputs.zones }}'
- id: validate_geometry
type: io.kestra.plugin.core.flow.Parallel
tasks:
- id: validate_points
type: io.kestra.plugin.jdbc.postgresql.Query
description: Validate points geometry, coordinates and stable unique feature IDs
without writing database tables.
url: '{{ secret(''POSTGRES_JDBC_URL'') }}'
username: '{{ secret(''POSTGRES_USERNAME'') }}'
password: '{{ secret(''POSTGRES_PASSWORD'') }}'
fetchType: FETCH_ONE
parameters:
payload: '{{ read(outputs.stage_points.uri) }}'
sql: |
WITH document AS (SELECT CAST(:payload AS jsonb) AS payload),
features AS (
SELECT f, ST_GeomFromGeoJSON(f->'geometry') AS geom
FROM document, jsonb_array_elements(payload->'features') AS f
)
SELECT (SELECT (jsonb_typeof(payload) = 'object'
AND payload->>'type' = 'FeatureCollection'
AND jsonb_typeof(payload->'features') = 'array'
AND NOT jsonb_exists(payload, 'crs')) IS TRUE FROM document) AS collection_valid,
count(*) AS feature_count,
count(*) BETWEEN 1 AND 1000 AS count_valid,
count(DISTINCT f->>'id') = count(*) AS ids_unique,
COALESCE(bool_and((f->>'type' = 'Feature'
AND jsonb_typeof(f->'id') = 'string'
AND char_length(f->>'id') BETWEEN 1 AND 128
AND NOT jsonb_exists(f->'geometry', 'crs')
AND geom IS NOT NULL AND NOT ST_IsEmpty(geom)
AND ST_GeometryType(geom) IN ('ST_Point')
AND ST_SRID(geom) = 4326 AND ST_NDims(geom) = 2
AND ST_IsValid(geom) AND ST_NPoints(geom) <= 10000
AND ST_XMin(Box3D(geom)) >= -180 AND ST_XMax(Box3D(geom)) <= 180
AND ST_YMin(Box3D(geom)) >= -90 AND ST_YMax(Box3D(geom)) <= 90
) IS TRUE), false) AS features_valid
FROM features;
- id: validate_zones
type: io.kestra.plugin.jdbc.postgresql.Query
description: Validate zones geometry, coordinates and stable unique feature IDs
without writing database tables.
url: '{{ secret(''POSTGRES_JDBC_URL'') }}'
username: '{{ secret(''POSTGRES_USERNAME'') }}'
password: '{{ secret(''POSTGRES_PASSWORD'') }}'
fetchType: FETCH_ONE
parameters:
payload: '{{ read(outputs.stage_zones.uri) }}'
sql: |
WITH document AS (SELECT CAST(:payload AS jsonb) AS payload),
features AS (
SELECT f, ST_GeomFromGeoJSON(f->'geometry') AS geom
FROM document, jsonb_array_elements(payload->'features') AS f
)
SELECT (SELECT (jsonb_typeof(payload) = 'object'
AND payload->>'type' = 'FeatureCollection'
AND jsonb_typeof(payload->'features') = 'array'
AND NOT jsonb_exists(payload, 'crs')) IS TRUE FROM document) AS collection_valid,
count(*) AS feature_count,
count(*) BETWEEN 1 AND 100 AS count_valid,
count(DISTINCT f->>'id') = count(*) AS ids_unique,
COALESCE(bool_and((f->>'type' = 'Feature'
AND jsonb_typeof(f->'id') = 'string'
AND char_length(f->>'id') BETWEEN 1 AND 128
AND NOT jsonb_exists(f->'geometry', 'crs')
AND geom IS NOT NULL AND NOT ST_IsEmpty(geom)
AND ST_GeometryType(geom) IN ('ST_Polygon', 'ST_MultiPolygon')
AND ST_SRID(geom) = 4326 AND ST_NDims(geom) = 2
AND ST_IsValid(geom) AND ST_NPoints(geom) <= 10000
AND ST_XMin(Box3D(geom)) >= -180 AND ST_XMax(Box3D(geom)) <= 180
AND ST_YMin(Box3D(geom)) >= -90 AND ST_YMax(Box3D(geom)) <= 90
) IS TRUE), false) AS features_valid
FROM features;
- id: require_valid_inputs
type: io.kestra.plugin.core.execution.Assert
description: Reject invalid geometries and duplicate IDs before running any
point-in-polygon predicates.
conditions:
- '{{ outputs.validate_points.row.collection_valid and
outputs.validate_points.row.count_valid and
outputs.validate_points.row.ids_unique and
outputs.validate_points.row.features_valid }}'
- '{{ outputs.validate_zones.row.collection_valid and
outputs.validate_zones.row.count_valid and
outputs.validate_zones.row.ids_unique and
outputs.validate_zones.row.features_valid }}'
- id: assign_zones
type: io.kestra.plugin.jdbc.postgresql.Query
description: Include polygon boundaries; preserve every event and report
zero/multiple matches instead of choosing an arbitrary zone.
url: '{{ secret(''POSTGRES_JDBC_URL'') }}'
username: '{{ secret(''POSTGRES_USERNAME'') }}'
password: '{{ secret(''POSTGRES_PASSWORD'') }}'
fetchType: STORE
fetchSize: 1000
parameters:
points: '{{ read(outputs.stage_points.uri) }}'
zones: '{{ read(outputs.stage_zones.uri) }}'
sql: |
WITH
points AS MATERIALIZED (
SELECT f->>'id' AS event_id, ST_GeomFromGeoJSON(f->'geometry') AS geom
FROM jsonb_array_elements(CAST(:points AS jsonb)->'features') AS f
),
zones AS MATERIALIZED (
SELECT f->>'id' AS zone_id, ST_GeomFromGeoJSON(f->'geometry') AS geom
FROM jsonb_array_elements(CAST(:zones AS jsonb)->'features') AS f
),
matches AS (
SELECT p.event_id, ST_X(p.geom) AS longitude, ST_Y(p.geom) AS latitude,
count(z.zone_id) AS match_count,
min(z.zone_id) AS only_zone,
COALESCE(jsonb_agg(z.zone_id ORDER BY z.zone_id) FILTER (WHERE z.zone_id IS NOT NULL),
CAST('[]' AS jsonb)) AS candidate_zones
FROM points p LEFT JOIN zones z ON ST_Covers(z.geom, p.geom)
GROUP BY p.event_id, p.geom
)
SELECT event_id, longitude, latitude,
CASE WHEN match_count = 1 THEN only_zone ELSE NULL END AS assigned_zone,
match_count,
CASE WHEN match_count = 1 THEN 'MATCHED'
WHEN match_count = 0 THEN 'UNMATCHED' ELSE 'AMBIGUOUS' END AS status,
CAST(candidate_zones AS text) AS candidate_zones
FROM matches ORDER BY event_id;
- id: export_assignments
type: io.kestra.plugin.serdes.csv.IonToCsv
from: '{{ outputs.assign_zones.uri }}'
header: true
outputs:
- id: assignments
type: FILE
value: '{{ outputs.export_assignments.uri ?? null }}'
required: false
triggers:
- id: incoming_batch
type: io.kestra.plugin.core.trigger.Webhook
key: '{{ secret(''GEO_WEBHOOK_KEY'') }}'
inputs:
points: '{{ trigger.body.points }}'
zones: '{{ trigger.body.zones }}'
