Added notification triggers in database.

Author: pantierra

.github/workflows/tests/test_pgstac_notifications.py

@@ -0,0 +1,315 @@
+"""Test pgstac notification triggers."""
+import json
+import os
+import psycopg2
+import psycopg2.extensions
+import pytest
+import time
+from datetime import datetime, timezone
+
+
+@pytest.fixture(scope='session')
+def db_connection():
+    """Create database connection for testing."""
+    # Use the same environment variables as the pgstac bootstrap
+    connection_params = {
+        'host': os.getenv('PGHOST', 'localhost'),
+        'port': int(os.getenv('PGPORT', '5432')),
+        'database': os.getenv('PGDATABASE', 'postgres'),
+        'user': os.getenv('PGUSER', 'postgres'),
+        'password': os.getenv('PGPASSWORD', 'password')
+    }
+
+    conn = psycopg2.connect(**connection_params)
+    conn.set_isolation_level(psycopg2.extensions.ISOLATION_LEVEL_AUTOCOMMIT)
+    yield conn
+    conn.close()
+
+
+@pytest.fixture
+def notification_listener(db_connection):
+    """Set up notification listener for pgstac_items_change."""
+    cursor = db_connection.cursor()
+    cursor.execute("LISTEN pgstac_items_change;")
+    yield cursor
+    cursor.execute("UNLISTEN pgstac_items_change;")
+    cursor.close()
+
+
+def test_notification_triggers_exist(db_connection):
+    """Test that notification triggers and function are properly installed."""
+    cursor = db_connection.cursor()
+
+    # Check that the notification function exists
+    cursor.execute("""
+        SELECT EXISTS(
+            SELECT 1 FROM pg_proc
+            WHERE proname = 'notify_items_change_func'
+        );
+    """)
+    assert cursor.fetchone()[0], "notify_items_change_func function should exist"
+
+    # Check that all three triggers exist
+    trigger_names = [
+        'notify_items_change_insert',
+        'notify_items_change_update',
+        'notify_items_change_delete'
+    ]
+
+    for trigger_name in trigger_names:
+        cursor.execute("""
+            SELECT EXISTS(
+                SELECT 1 FROM pg_trigger
+                WHERE tgname = %s
+                AND tgrelid = 'pgstac.items'::regclass
+            );
+        """, (trigger_name,))
+        assert cursor.fetchone()[0], f"Trigger {trigger_name} should exist on pgstac.items"
+
+    cursor.close()
+
+
+def test_insert_notification(db_connection, notification_listener):
+    """Test that INSERT operations trigger notifications."""
+    cursor = db_connection.cursor()
+
+    # Clear any pending notifications
+    db_connection.poll()
+    while db_connection.notifies:
+        db_connection.notifies.pop(0)
+
+    # Use existing collection
+    test_collection_id = "noaa-emergency-response"
+
+    # Insert a test item using pgstac.create_item
+    test_item_id = f"test-item-{int(time.time())}"
+    item_data = json.dumps({
+        "id": test_item_id,
+        "type": "Feature",
+        "stac_version": "1.0.0",
+        "collection": test_collection_id,
+        "geometry": {"type": "Point", "coordinates": [0, 0]},
+        "bbox": [0, 0, 0, 0],
+        "properties": {"datetime": "2020-01-01T00:00:00Z"},
+        "assets": {}
+    })
+
+    cursor.execute("SELECT pgstac.create_item(%s);", (item_data,))
+
+    # Wait for notification
+    timeout = 5
+    start_time = time.time()
+    received_notification = False
+
+    while time.time() - start_time < timeout:
+        db_connection.poll()
+        if db_connection.notifies:
+            notify = db_connection.notifies.pop(0)
+            assert notify.channel == "pgstac_items_change"
+
+            # Parse the notification payload
+            payload = json.loads(notify.payload)
+            assert payload["operation"] == "INSERT"
+            assert "items" in payload
+            assert len(payload["items"]) == 1
+            assert payload["items"][0]["id"] == test_item_id
+            assert payload["items"][0]["collection"] == test_collection_id
+
+            received_notification = True
+            break
+        time.sleep(0.1)
+
+    assert received_notification, "Should have received INSERT notification"
+
+    # Cleanup
+    cursor.execute("SELECT pgstac.delete_item(%s);", (test_item_id,))
+    cursor.close()
+
+
+def test_update_notification(db_connection, notification_listener):
+    """Test that UPDATE operations trigger notifications."""
+    cursor = db_connection.cursor()
+
+    # Clear any pending notifications
+    db_connection.poll()
+    while db_connection.notifies:
+        db_connection.notifies.pop(0)
+
+    test_collection_id = "noaa-emergency-response"
+
+    # Insert a test item first using pgstac.create_item
+    test_item_id = f"test-item-update-{int(time.time())}"
+    item_data = json.dumps({
+        "id": test_item_id,
+        "type": "Feature",
+        "stac_version": "1.0.0",
+        "collection": test_collection_id,
+        "geometry": {"type": "Point", "coordinates": [0, 0]},
+        "bbox": [0, 0, 0, 0],
+        "properties": {"datetime": "2020-01-01T00:00:00Z"},
+        "assets": {}
+    })
+
+    cursor.execute("SELECT pgstac.create_item(%s);", (item_data,))
+
+    # Clear INSERT notification
+    db_connection.poll()
+    while db_connection.notifies:
+        db_connection.notifies.pop(0)
+
+    # Update the item using pgstac.update_item
+    updated_item_data = json.dumps({
+        "id": test_item_id,
+        "type": "Feature",
+        "stac_version": "1.0.0",
+        "collection": test_collection_id,
+        "geometry": {"type": "Point", "coordinates": [0, 0]},
+        "bbox": [0, 0, 0, 0],
+        "properties": {"datetime": "2020-01-01T00:00:00Z", "updated": True},
+        "assets": {}
+    })
+
+    cursor.execute("SELECT pgstac.update_item(%s);", (updated_item_data,))
+
+    # Wait for notification
+    timeout = 5
+    start_time = time.time()
+    received_notification = False
+
+    while time.time() - start_time < timeout:
+        db_connection.poll()
+        if db_connection.notifies:
+            notify = db_connection.notifies.pop(0)
+            assert notify.channel == "pgstac_items_change"
+
+            # Parse the notification payload - PgSTAC update uses DELETE+INSERT, so accept both
+            payload = json.loads(notify.payload)
+            assert payload["operation"] in ["DELETE", "INSERT", "UPDATE"], f"Operation should be DELETE, INSERT, or UPDATE, got {payload['operation']}"
+            assert "items" in payload
+            assert len(payload["items"]) == 1
+            assert payload["items"][0]["id"] == test_item_id
+            assert payload["items"][0]["collection"] == test_collection_id
+
+            received_notification = True
+            break
+        time.sleep(0.1)
+
+    assert received_notification, "Should have received UPDATE notification"
+
+    # Cleanup
+    cursor.execute("SELECT pgstac.delete_item(%s);", (test_item_id,))
+    cursor.close()
+
+
+def test_delete_notification(db_connection, notification_listener):
+    """Test that DELETE operations trigger notifications."""
+    cursor = db_connection.cursor()
+
+    # Clear any pending notifications
+    db_connection.poll()
+    while db_connection.notifies:
+        db_connection.notifies.pop(0)
+
+    test_collection_id = "noaa-emergency-response"
+
+    # Insert a test item first using pgstac.create_item
+    test_item_id = f"test-item-delete-{int(time.time())}"
+    item_data = json.dumps({
+        "id": test_item_id,
+        "type": "Feature",
+        "stac_version": "1.0.0",
+        "collection": test_collection_id,
+        "geometry": {"type": "Point", "coordinates": [0, 0]},
+        "bbox": [0, 0, 0, 0],
+        "properties": {"datetime": "2020-01-01T00:00:00Z"},
+        "assets": {}
+    })
+
+    cursor.execute("SELECT pgstac.create_item(%s);", (item_data,))
+
+    # Clear INSERT notification
+    db_connection.poll()
+    while db_connection.notifies:
+        db_connection.notifies.pop(0)
+
+    # Delete the item using pgstac.delete_item
+    cursor.execute("SELECT pgstac.delete_item(%s);", (test_item_id,))
+
+    # Wait for notification
+    timeout = 5
+    start_time = time.time()
+    received_notification = False
+
+    while time.time() - start_time < timeout:
+        db_connection.poll()
+        if db_connection.notifies:
+            notify = db_connection.notifies.pop(0)
+            assert notify.channel == "pgstac_items_change"
+
+            # Parse the notification payload
+            payload = json.loads(notify.payload)
+            assert payload["operation"] == "DELETE"
+            assert "items" in payload
+            assert len(payload["items"]) == 1
+            assert payload["items"][0]["id"] == test_item_id
+            assert payload["items"][0]["collection"] == test_collection_id
+
+            received_notification = True
+            break
+        time.sleep(0.1)
+
+    assert received_notification, "Should have received DELETE notification"
+    cursor.close()
+
+
+def test_bulk_operations_notification(db_connection, notification_listener):
+    """Test that bulk operations send single notifications with multiple items."""
+    cursor = db_connection.cursor()
+
+    # Clear any pending notifications
+    db_connection.poll()
+    while db_connection.notifies:
+        db_connection.notifies.pop(0)
+
+    test_collection_id = "noaa-emergency-response"
+
+    # Insert multiple items using pgstac.create_item
+    test_items = [f"bulk-item-{i}-{int(time.time())}" for i in range(3)]
+
+    for item_id in test_items:
+        item_data = json.dumps({
+            "id": item_id,
+            "type": "Feature",
+            "stac_version": "1.0.0",
+            "collection": test_collection_id,
+            "geometry": {"type": "Point", "coordinates": [0, 0]},
+            "bbox": [0, 0, 0, 0],
+            "properties": {"datetime": "2020-01-01T00:00:00Z"},
+            "assets": {}
+        })
+
+        cursor.execute("SELECT pgstac.create_item(%s);", (item_data,))
+
+    # Wait for notifications (should get one per insert since we're doing separate statements)
+    timeout = 10
+    start_time = time.time()
+    notifications_received = 0
+
+    while time.time() - start_time < timeout and notifications_received < len(test_items):
+        db_connection.poll()
+        while db_connection.notifies:
+            notify = db_connection.notifies.pop(0)
+            assert notify.channel == "pgstac_items_change"
+
+            payload = json.loads(notify.payload)
+            assert payload["operation"] == "INSERT"
+            assert "items" in payload
+            notifications_received += len(payload["items"])
+
+    assert notifications_received >= len(test_items), f"Should have received notifications for all {len(test_items)} items"
+
+    # Cleanup
+    for item_id in test_items:
+        cursor.execute("SELECT pgstac.delete_item(%s);", (item_id,))
+
+    cursor.close()

CHANGELOG.md

@@ -10,6 +10,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ### Added
 
 - Enforcement of `CHANGELOG.md` entries for PRs and Conventional Commits for PR titles [#288](https://github.com/developmentseed/eoapi-k8s/pull/288)
+- Notification triggers in database [#289](https://github.com/developmentseed/eoapi-k8s/pull/289)
 
 ## [0.7.8] - 2025-09-10
 

charts/eoapi/README.md

@@ -12,6 +12,7 @@ A Helm chart for deploying Earth Observation APIs with integrated STAC, raster,
 - Multidimensional data support
 - Built-in STAC Browser interface
 - Flexible database configuration
+- Real-time PostgreSQL notifications for STAC item changes
 - Unified ingress system
 
 ## TL;DR
@@ -75,6 +76,7 @@ pgstacBootstrap:
 | `ingress.className` | Ingress controller class | `nginx` |
 | `browser.enabled` | Enable STAC Browser interface | `true` |
 | `pgstacBootstrap.enabled` | Enable database initialization | `true` |
+| `notifications.sources.pgstac` | Enable PostgreSQL notification triggers for STAC item changes | `false` |
 
 Refer to the [values.schema.json](./values.schema.json) for the complete list of configurable parameters.
 

charts/eoapi/initdb-data/settings/pgstac-notification-triggers.sql

@@ -0,0 +1,42 @@
+-- Create the notification function
+CREATE OR REPLACE FUNCTION notify_items_change_func()
+RETURNS TRIGGER AS $$
+DECLARE
+
+BEGIN
+    PERFORM pg_notify('pgstac_items_change'::text, json_build_object(
+            'operation', TG_OP,
+            'items', jsonb_agg(
+                jsonb_build_object(
+                    'collection', data.collection,
+                    'id', data.id
+                )
+            )
+        )::text
+        )
+        FROM data
+    ;
+    RETURN NULL;
+END;
+$$ LANGUAGE plpgsql;
+
+-- Create triggers for INSERT operations
+CREATE OR REPLACE TRIGGER notify_items_change_insert
+    AFTER INSERT ON pgstac.items
+    REFERENCING NEW TABLE AS data
+    FOR EACH STATEMENT EXECUTE FUNCTION notify_items_change_func()
+;
+
+-- Create triggers for UPDATE operations
+CREATE OR REPLACE TRIGGER notify_items_change_update
+    AFTER UPDATE ON pgstac.items
+    REFERENCING NEW TABLE AS data
+    FOR EACH STATEMENT EXECUTE FUNCTION notify_items_change_func()
+;
+
+-- Create triggers for DELETE operations
+CREATE OR REPLACE TRIGGER notify_items_change_delete
+    AFTER DELETE ON pgstac.items
+    REFERENCING OLD TABLE AS data
+    FOR EACH STATEMENT EXECUTE FUNCTION notify_items_change_func()
+;

charts/eoapi/initdb-data/settings/pgstac-settings.sql

@@ -10,3 +10,5 @@ INSERT INTO pgstac.pgstac_settings (name, value) VALUES ('context_estimated_cost
 
 DELETE FROM pgstac.pgstac_settings WHERE name = 'context_stats_ttl';
 INSERT INTO pgstac.pgstac_settings (name, value) VALUES ('context_stats_ttl', '1 day');
+
+

charts/eoapi/templates/pgstacbootstrap/configmap.yaml

@@ -14,6 +14,9 @@ metadata:
 data:
   pgstac-settings.sql: |
     {{ $.Files.Get "initdb-data/settings/pgstac-settings.sql" | nindent 4 }}
+    {{- if .Values.notifications.sources.pgstac }}
+    {{ $.Files.Get "initdb-data/settings/pgstac-notification-triggers.sql" | nindent 4 }}
+    {{- end }}
 ---
 {{- if .Values.pgstacBootstrap.settings.loadSamples }}
 apiVersion: v1