Merge pull request #1321 from BudEcosystem/fix/budpipeline-notify-publish-pattern

fix(budpipeline): use flat payload for pub/sub and fix stale DB constraint

Author: dittops

services/budpipeline/budpipeline/alembic/versions/20260116_2040_003_fix_delivery_status_constraint.py

@@ -20,7 +20,11 @@
 
 def upgrade() -> None:
     """Update check constraint to use uppercase enum names."""
-    # Drop the old constraint using raw SQL
+    # Drop both possible constraint names: the SQLAlchemy-generated prefixed
+    # name and the short name used in migration 001.
+    op.execute(
+        'ALTER TABLE execution_subscription DROP CONSTRAINT IF EXISTS "ck_execution_subscription_chk_execution_subscription_status"'
+    )
     op.execute(
         'ALTER TABLE execution_subscription DROP CONSTRAINT IF EXISTS "chk_execution_subscription_status"'
     )

services/budpipeline/budpipeline/alembic/versions/20260216_1200_009_drop_stale_delivery_status_constraint.py

@@ -0,0 +1,41 @@
+"""Drop stale lowercase delivery_status check constraint.
+
+Revision ID: 009_fix_status_constraint
+Revises: 008_add_notification_workflow_id
+Create Date: 2026-02-16 12:00:00.000000
+
+Migration 003 intended to replace the lowercase delivery_status constraint
+with an uppercase one, but used the wrong constraint name. The SQLAlchemy
+naming convention prefixed the original constraint as
+'ck_execution_subscription_chk_execution_subscription_status' (lowercase),
+while migration 003 only dropped 'chk_execution_subscription_status'.
+
+This left two constraints: the old lowercase one (blocking inserts of 'ACTIVE')
+and the new uppercase one. This migration drops the stale lowercase constraint.
+"""
+
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = "009_fix_status_constraint"
+down_revision = "008_add_notification_workflow_id"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    """Drop the stale lowercase constraint that blocks ACTIVE inserts."""
+    op.execute(
+        'ALTER TABLE execution_subscription DROP CONSTRAINT IF EXISTS "ck_execution_subscription_chk_execution_subscription_status"'
+    )
+
+
+def downgrade() -> None:
+    """Re-add the lowercase constraint (not recommended)."""
+    op.execute(
+        """
+        ALTER TABLE execution_subscription
+        ADD CONSTRAINT ck_execution_subscription_chk_execution_subscription_status
+        CHECK (delivery_status IN ('active', 'expired', 'failed'))
+        """
+    )

services/budpipeline/budpipeline/progress/publisher.py

@@ -350,10 +350,10 @@ async def _publish_single_topic(
     ) -> bool:
         """Publish event to a single topic.
 
-        Constructs a full CloudEvent envelope and publishes with
-        application/cloudevents+json content type. This prevents Dapr from
-        injecting its own id/datacontenttype fields, which budnotify's
-        CloudEventBase(extra="forbid") would reject with 422.
+        Sends flat payload data with CloudEvent attributes passed via
+        publish_metadata, matching the budmicroframe publish_to_topic
+        pattern. This avoids injecting extra fields (id, datacontenttype)
+        into the JSON body that downstream subscribers would reject.
 
         Args:
             topic: Target topic.
@@ -370,30 +370,25 @@ async def _publish_single_topic(
 
             # Copy payload to avoid mutating the shared dict across topics
             event_payload = payload.copy()
-            correlation_id = event_payload.pop("correlation_id", None)
-
-            # Construct a full CloudEvent envelope so Dapr does not inject
-            # its own id/datacontenttype fields (which budnotify rejects).
-            # Extension attributes (source_topic, correlation_id) go at the
-            # envelope level, not inside data.
-            cloud_event: dict[str, Any] = {
-                "specversion": "1.0",
-                "id": str(uuid4()),
-                "source": settings.name,
-                "type": event_type,
-                "datacontenttype": "application/json",
-                "source_topic": topic,
-                "data": event_payload,
+            event_payload["source"] = settings.name
+            event_payload["source_topic"] = topic
+
+            # Pass CloudEvent attributes via publish_metadata so Dapr
+            # does not inject them into the JSON body.
+            event_id = str(uuid4())
+            publish_metadata = {
+                "cloudevent.id": event_id,
+                "cloudevent.source": settings.name,
+                "cloudevent.type": event_type,
             }
-            if correlation_id:
-                cloud_event["correlation_id"] = correlation_id
 
             async with DaprClient() as client:
                 await client.publish_event(
                     pubsub_name=self.pubsub_name,
                     topic_name=topic,
-                    data=json.dumps(cloud_event, cls=_DecimalEncoder),
+                    data=json.dumps(event_payload, cls=_DecimalEncoder),
                     data_content_type="application/cloudevents+json",
+                    publish_metadata=publish_metadata,
                 )
 
             record_event_published(event_type)