UN-3056 [FEAT] Failure-only notifications with batched dispatch

backend/api_v2/notification.py

@@ -1,6 +1,7 @@
 import logging
 
-from notification_v2.helper import NotificationHelper
+from notification_v2.enums import FAILURE_STATUSES
+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.models.execution import WorkflowExecution
@@ -16,11 +17,33 @@ def __init__(self, api: APIDeployment, workflow_execution: WorkflowExecution) ->
         self.api = api
         self.workflow_execution = workflow_execution
 
-    def send(self):
-        if not self.notifications.count():
-            logger.info(f"No notifications found for api {self.api}")
+    def send(self) -> None:
+        # Failure if the run hit a non-success terminal state OR any file errored.
+        # Partial-success runs land as status=COMPLETED with failed_files>0, so the
+        # status check alone misses them — see callback aggregation rules.
+        failed_files = self.workflow_execution.failed_files or 0
+        is_failure = (
+            self.workflow_execution.status in FAILURE_STATUSES or failed_files > 0
+        )
+        if not is_failure:
+            # Success path: skip rows that opted into failure-only alerts.
+            self.notifications = self.notifications.filter(notify_on_failures=False)
+
+        if not self.notifications.exists():
+            logger.info(
+                "No notifications to dispatch for api %s (status=%s, failed_files=%s)",
+                self.api,
+                self.workflow_execution.status,
+                failed_files,
+            )
             return
-        logger.info(f"Sending api status notification for api {self.api}")
+        logger.info(
+            "Sending api status notification for api %s (status=%s, successful=%s, failed=%s)",
+            self.api,
+            self.workflow_execution.status,
+            self.workflow_execution.successful_files or 0,
+            failed_files,
+        )
 
         payload_dto = PipelineStatusPayload(
             type="API",
@@ -29,8 +52,13 @@ def send(self):
             status=self.workflow_execution.status,
             execution_id=self.workflow_execution.id,
             error_message=self.workflow_execution.error_message,
+            total_files=self.workflow_execution.total_files,
+            successful_files=self.workflow_execution.successful_files,
+            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

@@ -219,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 300 (5 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", "300"))
+# 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")
 )

backend/configuration/enums.py

@@ -61,6 +61,14 @@ 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.",
+        min_value=60,
+        max_value=7200,
+    )
+
     def cast_value(self, raw_value: Any):
         converters = {
             ConfigType.INT: int,

backend/notification_v2/clubbed_renderer.py

@@ -0,0 +1,52 @@
+"""Backend dispatch entry for clubbed-notification rendering.
+
+Delegates the canonical envelope + Slack body to
+``unstract.core.notification_clubbed_renderer`` so backend and worker
+callbacks emit byte-identical receiver-visible payloads. This thin shim
+keeps the ``render_clubbed_message`` platform dispatcher (uses
+``PlatformType`` enum) backend-side; everything else lives in the shared
+module.
+"""
+
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+from notification_v2.enums import PlatformType
+from unstract.core.notification_clubbed_renderer import (
+    build_envelope,
+    render_slack_text,
+)
+
+logger = logging.getLogger(__name__)
+
+__all__ = ["build_envelope", "render_clubbed_message"]
+
+
+def _render_for_slack(envelope: dict[str, Any]) -> dict[str, Any]:
+    """Wrap the rendered Slack mrkdwn body in the dict shape Slack expects."""
+    return {"text": render_slack_text(envelope)}
+
+
+def render_clubbed_message(
+    payloads: list[dict[str, Any]],
+    platform: str,
+) -> dict[str, Any]:
+    """Top-level entry — returns the dispatch body for ``platform``.
+
+    Used by every dispatch site so the receiver-visible payload is
+    identical regardless of caller.
+    """
+    envelope = build_envelope(payloads)
+    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

@@ -1,5 +1,12 @@
 from enum import Enum
 
+from workflow_manager.workflow_v2.enums import ExecutionStatus
+
+# Single source of truth for "did this run fail for notification routing?".
+# STOPPED is intentionally a failure here per migrations/0002_…notify_on_failures
+# db_comment ("terminal status ERROR/STOPPED or any file in the run errored").
+FAILURE_STATUSES = frozenset({ExecutionStatus.ERROR.value, ExecutionStatus.STOPPED.value})
+
 
 class NotificationType(Enum):
     WEBHOOK = "WEBHOOK"
@@ -36,3 +43,41 @@ 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.
+
+    Product ships every notification as BATCHED — events buffer into
+    ``NotificationBuffer`` and flush as one clubbed message per
+    (org, webhook_url, auth_sig) every ``NOTIFICATION_CLUB_INTERVAL`` seconds.
+
+    The ``IMMEDIATE`` value is purely a historical DB value — no code reads
+    it anymore (the legacy synchronous-dispatch path was removed). The
+    column and enum value remain so existing rows don't break; both will be
+    dropped in a follow-up schema migration.
+    """
+
+    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]