Uns 611 clubbed notification dispatch

.github/workflows/uv-lock-automation.yaml

@@ -36,8 +36,6 @@ jobs:
           version: "0.6.14"
           python-version: 3.12.9
 
-      - run: uv pip install --python=3.12.9 pip
-
       - name: Generate UV lockfiles
         env:
           GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}

backend/api_v2/notification.py

@@ -1,6 +1,6 @@
 import logging
 
-from notification_v2.helper import NotificationHelper
+from notification_v2.helper import dispatch_with_delivery_mode
 from notification_v2.models import Notification
 from pipeline_v2.dto import PipelineStatusPayload
 from workflow_manager.workflow_v2.enums import ExecutionStatus
@@ -60,6 +60,8 @@ def send(self) -> None:
             failed_files=failed_files,
         )
 
-        NotificationHelper.send_notification(
-            notifications=self.notifications, payload=payload_dto.to_dict()
+        dispatch_with_delivery_mode(
+            list(self.notifications),
+            payload_dto.to_dict(),
+            error_context=f"api={self.api.id}",
         )

backend/backend/settings/base.py

@@ -12,11 +12,12 @@
 import logging
 import os
 from pathlib import Path
-from urllib.parse import quote, urlparse
+from urllib.parse import quote
 
 import httpx
 from dotenv import find_dotenv, load_dotenv
 from utils.common_utils import CommonUtils
+from utils.cors_origin import normalize_web_app_origin
 
 missing_settings = []
 
@@ -68,10 +69,18 @@ def get_required_setting(setting_key: str, default: str | None = None) -> str |
 )
 # Maximum number of files allowed per workflow page execution
 WORKFLOW_PAGE_MAX_FILES = int(os.environ.get("WORKFLOW_PAGE_MAX_FILES", 2))
-WEB_APP_ORIGIN_URL = os.environ.get("WEB_APP_ORIGIN_URL", "http://localhost:3000")
-parsed_url = urlparse(WEB_APP_ORIGIN_URL)
-WEB_APP_ORIGIN_URL_WITH_WILD_CARD = f"{parsed_url.scheme}://*.{parsed_url.netloc}"
+(
+    WEB_APP_ORIGIN_URL,
+    WEB_APP_ORIGIN_URL_WITH_WILD_CARD,
+    _CORS_SUBDOMAIN_REGEX,
+) = normalize_web_app_origin(
+    os.environ.get("WEB_APP_ORIGIN_URL", "http://localhost:3000")
+)
 CORS_ALLOWED_ORIGINS = [WEB_APP_ORIGIN_URL]
+# Wildcard subdomain regex consumed by django-cors-headers and (via the
+# RegexOrigin wrapper in utils/log_events.py) by the SocketIO engine.io
+# handshake — a single source of truth so HTTP and WS CORS cannot diverge.
+CORS_ALLOWED_ORIGIN_REGEXES = [_CORS_SUBDOMAIN_REGEX]
 
 DJANGO_APP_BACKEND_URL = os.environ.get("DJANGO_APP_BACKEND_URL", "http://localhost:8000")
 INTERNAL_SERVICE_API_KEY = os.environ.get("INTERNAL_SERVICE_API_KEY")
@@ -210,6 +219,15 @@ def get_required_setting(setting_key: str, default: str | None = None) -> str |
 
 INDEXING_FLAG_TTL = int(get_required_setting("INDEXING_FLAG_TTL"))
 NOTIFICATION_TIMEOUT = int(get_required_setting("NOTIFICATION_TIMEOUT", "5"))
+# Window for clubbing BATCHED notifications — also the flush cadence (seconds).
+# Default 1800 (30 min). Per-notification buffer rows precompute flush_after at
+# enqueue time, so changing this only affects rows enqueued after the restart.
+NOTIFICATION_CLUB_INTERVAL = int(os.environ.get("NOTIFICATION_CLUB_INTERVAL", "1800"))
+# Retention for terminal NotificationBuffer rows (DISPATCHED / DEAD_LETTER).
+# PENDING rows are never GC'd regardless of age.
+NOTIFICATION_BUFFER_RETENTION_DAYS = int(
+    os.environ.get("NOTIFICATION_BUFFER_RETENTION_DAYS", "7")
+)
 ATOMIC_REQUESTS = CommonUtils.str_to_bool(
     os.environ.get("DJANGO_ATOMIC_REQUESTS", "False")
 )
@@ -689,4 +707,6 @@ def filter(self, record):
     )
     raise ValueError(ERROR_MESSAGE)
 
-ENABLE_HIGHLIGHT_API_DEPLOYMENT = os.environ.get("ENABLE_HIGHLIGHT_API_DEPLOYMENT", False)
+ENABLE_HIGHLIGHT_API_DEPLOYMENT = CommonUtils.str_to_bool(
+    os.environ.get("ENABLE_HIGHLIGHT_API_DEPLOYMENT", "False")
+)

backend/configuration/enums.py

@@ -61,6 +61,12 @@ class ConfigKey(Enum):
         max_value=settings.MAX_PARALLEL_FILE_BATCHES_MAX_VALUE,
     )
 
+    NOTIFICATION_CLUB_INTERVAL = ConfigSpec(
+        default=settings.NOTIFICATION_CLUB_INTERVAL,
+        value_type=ConfigType.INT,
+        help_text="Window (seconds) for clubbing BATCHED notifications.",
+    )
+
     def cast_value(self, raw_value: Any):
         converters = {
             ConfigType.INT: int,

backend/notification_v2/clubbed_renderer.py

@@ -0,0 +1,138 @@
+"""Clubbed notification renderer.
+
+Builds one canonical JSON envelope from a group of buffered execution events
+and emits the platform-appropriate dispatch payload. Stays separate from the
+single-event SlackWebhook / APIWebhook providers so immediate-dispatch behavior
+stays untouched.
+
+Envelope shape (always the same — single-event groups use this too so consumers
+never need to branch on "is this batched?"):
+
+    {
+        "kind": "batch",
+        "summary": {
+            "pipeline": "<name>",
+            "interval_minutes": 30,
+            "total": N, "succeeded": S, "failed": F
+        },
+        "events": [{"execution_id": ..., "status": ..., "error": ...?}, ...]
+    }
+"""
+
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+from notification_v2.enums import PlatformType
+
+logger = logging.getLogger(__name__)
+
+# Hard cap on events per dispatch — extras roll over to the next flush tick.
+# Bounds memory + payload size and prevents a runaway backlog from creating an
+# unbounded HTTP body.
+MAX_BATCH_SIZE = 500
+# How many events Slack renders inline before collapsing the rest under a
+# "… and K more" footer. Slack tolerates much larger payloads, but readability
+# tanks past ~25 lines.
+SLACK_MAX_DISPLAY_EVENTS = 25
+
+_SUCCESS_STATUSES = {"COMPLETED", "SUCCESS"}
+
+
+def _is_success(status: str | None) -> bool:
+    if not status:
+        return False
+    return status.upper() in _SUCCESS_STATUSES
+
+
+def _event_from_payload(payload: dict[str, Any]) -> dict[str, Any]:
+    event: dict[str, Any] = {
+        "execution_id": payload.get("execution_id"),
+        "status": payload.get("status"),
+    }
+    error_message = payload.get("error_message")
+    if error_message:
+        event["error"] = error_message
+    return event
+
+
+def build_envelope(
+    payloads: list[dict[str, Any]], interval_seconds: int
+) -> dict[str, Any]:
+    """Build the canonical batch envelope.
+
+    Caps the events list at MAX_BATCH_SIZE; oldest-first ordering is the
+    caller's responsibility (the flush job sorts by created_at).
+    """
+    capped = payloads[:MAX_BATCH_SIZE]
+    succeeded = sum(1 for p in capped if _is_success(p.get("status")))
+    failed = len(capped) - succeeded
+    # Multiple pipelines can share an (org, url, auth_sig) group; we surface
+    # the first one's name as a representative. Mixed-pipeline batches are
+    # rare in practice and a v2 enhancement would aggregate distinct names.
+    pipeline_name = capped[0].get("pipeline_name") if capped else None
+    return {
+        "kind": "batch",
+        "summary": {
+            "pipeline": pipeline_name,
+            "interval_minutes": max(1, interval_seconds // 60),
+            "total": len(capped),
+            "succeeded": succeeded,
+            "failed": failed,
+        },
+        "events": [_event_from_payload(p) for p in capped],
+    }
+
+
+def _slack_event_line(event: dict[str, Any]) -> str:
+    parts = [f"— {event.get('execution_id') or 'unknown'}: {event.get('status')}"]
+    if event.get("error"):
+        parts.append(f"({event['error']})")
+    return " ".join(parts)
+
+
+def render_for_slack(envelope: dict[str, Any]) -> dict[str, Any]:
+    """Format the envelope as a Slack-compatible payload dict.
+
+    Returns the body shape Slack incoming webhooks expect (`text` field with
+    mrkdwn). Truncates inline events at SLACK_MAX_DISPLAY_EVENTS.
+    """
+    summary = envelope["summary"]
+    events: list[dict[str, Any]] = envelope["events"]
+    pipeline = summary.get("pipeline") or "pipeline"
+
+    header = f"*[Unstract] {summary['total']} executions for `{pipeline}`*"
+    counts = f"✅ {summary['succeeded']} succeeded  ❌ {summary['failed']} failed"
+
+    visible = events[:SLACK_MAX_DISPLAY_EVENTS]
+    lines = [_slack_event_line(e) for e in visible]
+    overflow = len(events) - len(visible)
+    if overflow > 0:
+        lines.append(f"… and {overflow} more executions")
+
+    body = "\n".join([header, counts, *lines])
+    return {"text": body}
+
+
+def render_clubbed_message(
+    payloads: list[dict[str, Any]], platform: str, interval_seconds: int
+) -> dict[str, Any]:
+    """Top-level entry point — returns the dispatch body for ``platform``.
+
+    Slack receives the rendered text payload; raw API webhooks receive the
+    canonical envelope unchanged so downstream consumers can parse it
+    programmatically.
+    """
+    envelope = build_envelope(payloads, interval_seconds)
+    if platform == PlatformType.SLACK.value:
+        return render_for_slack(envelope)
+    if platform == PlatformType.API.value:
+        return envelope
+    # Unknown platform — fall back to the raw envelope and warn so misrouted
+    # rows don't drop silently.
+    logger.warning(
+        "Unknown platform %s for clubbed dispatch; returning raw envelope",
+        platform,
+    )
+    return envelope

backend/notification_v2/enums.py

@@ -36,3 +36,37 @@ class PlatformType(Enum):
     @classmethod
     def choices(cls):
         return [(e.value, e.name.replace("_", " ").capitalize()) for e in cls]
+
+
+class DeliveryMode(Enum):
+    """Per-notification dispatch mode.
+
+    IMMEDIATE fires on every workflow completion (pre-existing behavior).
+    BATCHED buffers events into NotificationBuffer and flushes them as one
+    clubbed message per (org, webhook_url, auth_sig) every
+    NOTIFICATION_CLUB_INTERVAL seconds.
+    """
+
+    IMMEDIATE = "IMMEDIATE"
+    BATCHED = "BATCHED"
+
+    @classmethod
+    def choices(cls):
+        return [(e.value, e.name.replace("_", " ").capitalize()) for e in cls]
+
+
+class BufferStatus(Enum):
+    """Lifecycle states for a NotificationBuffer row.
+
+    PENDING       — waiting for the next flush tick.
+    DISPATCHED    — successfully sent as part of a clubbed message.
+    DEAD_LETTER   — Celery exhausted retries; terminal, never re-picked.
+    """
+
+    PENDING = "PENDING"
+    DISPATCHED = "DISPATCHED"
+    DEAD_LETTER = "DEAD_LETTER"
+
+    @classmethod
+    def choices(cls):
+        return [(e.value, e.name.replace("_", " ").capitalize()) for e in cls]