feat: notification ack fields (cherry-pick, isolated from unrelated stacked commits)

backend/api/v1/notifications.py

@@ -1,13 +1,17 @@
-from typing import Optional
+import hashlib
+import hmac
+from datetime import UTC, datetime
 
-from fastapi import APIRouter, Depends, HTTPException, Query
+from fastapi import APIRouter, Depends, HTTPException, Query, Request
 from sqlalchemy import func, select
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from backend.database import get_db
 from backend.models.notification import NotificationLog, NotificationRule
+from backend.models.record import CollectedRecord
 from backend.schemas.common import ApiResponse, PaginationMeta
 from backend.schemas.notification import (
+    NotificationAckRequest,
     NotificationLogRead,
     NotificationRuleCreate,
     NotificationRuleRead,
@@ -17,6 +21,15 @@
 router = APIRouter(prefix="/notifications", tags=["notifications"])
 
 
+def _verify_hmac(body: bytes, signature: str, secret: str) -> bool:
+    expected = "sha256=" + hmac.new(secret.encode(), body, hashlib.sha256).hexdigest()
+    return hmac.compare_digest(expected, signature)
+
+
+def _ack_secret(rule: NotificationRule) -> str:
+    return str(rule.notifier_config.get("ack_secret") or "")
+
+
 @router.get("/rules", response_model=ApiResponse[list[NotificationRuleRead]])
 async def list_rules(
     page: int = Query(1, ge=1),
@@ -81,7 +94,7 @@ async def delete_rule(rule_id: str, db: AsyncSession = Depends(get_db)) -> ApiRe
 
 @router.get("/logs", response_model=ApiResponse[list[NotificationLogRead]])
 async def list_logs(
-    rule_id: Optional[str] = None,
+    rule_id: str | None = None,
     page: int = Query(1, ge=1),
     limit: int = Query(20, ge=1, le=100),
     db: AsyncSession = Depends(get_db),
@@ -98,3 +111,49 @@ async def list_logs(
         data=[NotificationLogRead.model_validate(log) for log in logs],
         meta=PaginationMeta(total=total, page=page, limit=limit, pages=max(1, -(-total // limit))),
     )
+
+
+@router.post("/logs/{log_id}/ack", response_model=ApiResponse[NotificationLogRead])
+async def ack_log(
+    log_id: str,
+    body: NotificationAckRequest,
+    request: Request,
+    db: AsyncSession = Depends(get_db),
+) -> ApiResponse:
+    result = await db.execute(select(NotificationLog).where(NotificationLog.id == log_id))
+    log = result.scalar_one_or_none()
+    if not log:
+        raise HTTPException(status_code=404, detail="Notification log not found")
+
+    rule = await db.get(NotificationRule, log.rule_id)
+    if not rule:
+        raise HTTPException(status_code=404, detail="Notification rule not found")
+
+    secret = _ack_secret(rule)
+    if not secret:
+        raise HTTPException(
+            status_code=400, detail="Notification rule has no ack_secret configured"
+        )
+
+    signature = request.headers.get("X-Signature-256", "")
+    raw_body = await request.body()
+    if not signature or not _verify_hmac(raw_body, signature, secret):
+        raise HTTPException(status_code=401, detail="Invalid ACK signature")
+
+    log.ack_status = body.status
+    log.ack_data = body.ack_data
+    log.acked_at = datetime.now(UTC)
+
+    if log.record_id:
+        record = await db.get(CollectedRecord, log.record_id)
+        if record:
+            if body.status == "acked":
+                record.status = "notified"
+                record.error_message = None
+            else:
+                record.status = "error"
+                record.error_message = str(body.ack_data.get("error") or "Downstream ACK failed")
+
+    await db.flush()
+    await db.refresh(log)
+    return ApiResponse.ok(NotificationLogRead.model_validate(log))

backend/migrations/versions/m2n3o4p5q6r7_add_notification_ack_fields.py

@@ -0,0 +1,40 @@
+"""add notification ack fields
+
+Revision ID: m2n3o4p5q6r7
+Revises: d8e9f0a1b2c3
+Create Date: 2026-06-21
+
+"""
+import sqlalchemy as sa
+from alembic import op
+
+revision = 'm2n3o4p5q6r7'
+down_revision = 'd8e9f0a1b2c3'
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.add_column(
+        'notification_logs',
+        sa.Column(
+            'ack_status',
+            sa.String(length=50),
+            nullable=False,
+            server_default='not_required',
+        ),
+    )
+    op.add_column(
+        'notification_logs',
+        sa.Column('ack_data', sa.JSON(), nullable=True),
+    )
+    op.add_column(
+        'notification_logs',
+        sa.Column('acked_at', sa.DateTime(timezone=True), nullable=True),
+    )
+
+
+def downgrade() -> None:
+    op.drop_column('notification_logs', 'acked_at')
+    op.drop_column('notification_logs', 'ack_data')
+    op.drop_column('notification_logs', 'ack_status')

backend/models/notification.py

@@ -1,6 +1,6 @@
-from typing import Optional
+from datetime import datetime
 
-from sqlalchemy import JSON, Boolean, ForeignKey, String, Text
+from sqlalchemy import JSON, Boolean, DateTime, ForeignKey, String, Text
 from sqlalchemy.orm import Mapped, mapped_column, relationship
 
 from backend.models.base import TimestampMixin
@@ -13,7 +13,7 @@ class NotificationRule(TimestampMixin):
 
     name: Mapped[str] = mapped_column(String(255), nullable=False)
     # Source filter: null means all sources
-    source_id: Mapped[Optional[str]] = mapped_column(
+    source_id: Mapped[str | None] = mapped_column(
         String(36), ForeignKey("data_sources.id", ondelete="SET NULL"), nullable=True
     )
     # Trigger: on_new_record | on_ai_processed | on_task_failed
@@ -22,7 +22,7 @@ class NotificationRule(TimestampMixin):
     notifier_type: Mapped[str] = mapped_column(String(50), nullable=False)
     notifier_config: Mapped[dict] = mapped_column(JSON, nullable=False, default=dict)
     # Optional filter conditions as JSON logic
-    filter_conditions: Mapped[Optional[dict]] = mapped_column(JSON, nullable=True)
+    filter_conditions: Mapped[dict | None] = mapped_column(JSON, nullable=True)
     enabled: Mapped[bool] = mapped_column(Boolean, default=True, nullable=False)
 
     logs: Mapped[list["NotificationLog"]] = relationship(
@@ -38,10 +38,14 @@ class NotificationLog(TimestampMixin):
     rule_id: Mapped[str] = mapped_column(
         String(36), ForeignKey("notification_rules.id", ondelete="CASCADE"), nullable=False
     )
-    record_id: Mapped[Optional[str]] = mapped_column(String(36), nullable=True)
+    record_id: Mapped[str | None] = mapped_column(String(36), nullable=True)
     # sent | failed
     status: Mapped[str] = mapped_column(String(50), nullable=False)
-    response_data: Mapped[Optional[dict]] = mapped_column(JSON, nullable=True)
-    error_message: Mapped[Optional[str]] = mapped_column(Text, nullable=True)
+    response_data: Mapped[dict | None] = mapped_column(JSON, nullable=True)
+    error_message: Mapped[str | None] = mapped_column(Text, nullable=True)
+    # not_required | pending | acked | failed
+    ack_status: Mapped[str] = mapped_column(String(50), nullable=False, default="not_required")
+    ack_data: Mapped[dict | None] = mapped_column(JSON, nullable=True)
+    acked_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
 
     rule: Mapped["NotificationRule"] = relationship("NotificationRule", back_populates="logs")

backend/notifiers/base.py

@@ -7,13 +7,24 @@
 class NotificationPayload:
     event: str
     source_id: str
+    delivery_id: str | None = None
     record_id: str | None = None
     data: dict[str, Any] = field(default_factory=dict)
     ai_enrichment: dict[str, Any] | None = None
 
 
+@dataclass
+class NotificationSendResult:
+    success: bool
+    response_data: dict[str, Any] | None = None
+
+
 class AbstractNotifier(ABC):
     notifier_type: str
 
     @abstractmethod
-    async def send(self, config: dict[str, Any], payload: NotificationPayload) -> bool: ...
+    async def send(
+        self,
+        config: dict[str, Any],
+        payload: NotificationPayload,
+    ) -> bool | NotificationSendResult: ...

backend/notifiers/webhook_notifier.py

@@ -6,7 +6,7 @@
 import time
 from typing import Any
 
-from backend.notifiers.base import AbstractNotifier, NotificationPayload
+from backend.notifiers.base import AbstractNotifier, NotificationPayload, NotificationSendResult
 from backend.notifiers.registry import register_notifier
 from backend.security.url_guard import SSRFValidationError, guarded_async_client
 
@@ -15,7 +15,9 @@
 class WebhookNotifier(AbstractNotifier):
     notifier_type = "webhook"
 
-    async def send(self, config: dict[str, Any], payload: NotificationPayload) -> bool:
+    async def send(
+        self, config: dict[str, Any], payload: NotificationPayload
+    ) -> NotificationSendResult:
         url: str = config.get("url", "")
         secret: str = config.get("secret", "")
         timeout: int = config.get("timeout", 15)
@@ -33,6 +35,7 @@ async def send(self, config: dict[str, Any], payload: NotificationPayload) -> bo
         body = {
             "event": payload.event,
             "source_id": payload.source_id,
+            "delivery_id": payload.delivery_id,
             "record_id": payload.record_id,
             "data": payload.data,
             "ai_enrichment": payload.ai_enrichment,
@@ -50,4 +53,10 @@ async def send(self, config: dict[str, Any], payload: NotificationPayload) -> bo
         # loopback/fleet address (SSRF via redirect).
         async with client as opened_client:
             response = await opened_client.post(url, content=body_bytes, headers=headers)
-            return response.is_success
+            return NotificationSendResult(
+                success=response.is_success,
+                response_data={
+                    "status_code": response.status_code,
+                    "body": response.text[:1000],
+                },
+            )

backend/pipeline/notifier_dispatch.py

@@ -5,10 +5,20 @@
 
 from backend.models.notification import NotificationLog, NotificationRule
 from backend.models.record import CollectedRecord
-from backend.notifiers.base import NotificationPayload
+from backend.notifiers.base import NotificationPayload, NotificationSendResult
 from backend.notifiers.registry import get_notifier
 
 
+def _ack_secret(config: dict) -> str:
+    return str(config.get("ack_secret") or "")
+
+
+def _normalize_send_result(result: bool | NotificationSendResult) -> tuple[bool, dict | None]:
+    if isinstance(result, NotificationSendResult):
+        return result.success, result.response_data
+    return bool(result), None
+
+
 async def dispatch_notifications(
     session: AsyncSession,
     source_id: str,
@@ -36,25 +46,36 @@ async def dispatch_notifications(
             continue
 
         for record in records:
+            log = NotificationLog(
+                rule_id=rule.id,
+                record_id=record.id,
+                status="pending",
+                ack_status="not_required",
+            )
+            session.add(log)
+            await session.flush()
+
             payload = NotificationPayload(
                 event=trigger_event,
                 source_id=source_id,
+                delivery_id=log.id,
                 record_id=record.id,
                 data=record.normalized_data,
                 ai_enrichment=record.ai_enrichment,
             )
             try:
-                success = await notifier.send(rule.notifier_config, payload)
+                success, response_data = _normalize_send_result(
+                    await notifier.send(rule.notifier_config, payload)
+                )
                 status = "sent" if success else "failed"
                 error_msg = None
             except Exception as exc:
                 status = "failed"
+                response_data = None
                 error_msg = str(exc)
 
-            log = NotificationLog(
-                rule_id=rule.id,
-                record_id=record.id,
-                status=status,
-                error_message=error_msg,
-            )
-            session.add(log)
+            log.status = status
+            log.response_data = response_data
+            log.error_message = error_msg
+            if status == "sent" and _ack_secret(rule.notifier_config):
+                log.ack_status = "pending"

backend/schemas/notification.py

@@ -1,37 +1,38 @@
 from datetime import datetime
-from typing import Any, Optional
+from typing import Any, Literal
 
 from pydantic import BaseModel, Field
+
 from backend.schemas.common import UTCModel
 
 
 class NotificationRuleCreate(BaseModel):
     name: str = Field(..., min_length=1, max_length=255)
-    source_id: Optional[str] = None
+    source_id: str | None = None
     trigger_event: str
     notifier_type: str
     notifier_config: dict[str, Any] = Field(default_factory=dict)
-    filter_conditions: Optional[dict[str, Any]] = None
+    filter_conditions: dict[str, Any] | None = None
     enabled: bool = True
 
 
 class NotificationRuleUpdate(BaseModel):
-    name: Optional[str] = None
-    trigger_event: Optional[str] = None
-    notifier_type: Optional[str] = None
-    notifier_config: Optional[dict[str, Any]] = None
-    filter_conditions: Optional[dict[str, Any]] = None
-    enabled: Optional[bool] = None
+    name: str | None = None
+    trigger_event: str | None = None
+    notifier_type: str | None = None
+    notifier_config: dict[str, Any] | None = None
+    filter_conditions: dict[str, Any] | None = None
+    enabled: bool | None = None
 
 
 class NotificationRuleRead(UTCModel):
     id: str
     name: str
-    source_id: Optional[str]
+    source_id: str | None
     trigger_event: str
     notifier_type: str
     notifier_config: dict[str, Any]
-    filter_conditions: Optional[dict[str, Any]]
+    filter_conditions: dict[str, Any] | None
     enabled: bool
     created_at: datetime
     updated_at: datetime
@@ -42,10 +43,18 @@ class NotificationRuleRead(UTCModel):
 class NotificationLogRead(UTCModel):
     id: str
     rule_id: str
-    record_id: Optional[str]
+    record_id: str | None
     status: str
-    response_data: Optional[dict[str, Any]]
-    error_message: Optional[str]
+    response_data: dict[str, Any] | None
+    error_message: str | None
+    ack_status: str
+    ack_data: dict[str, Any] | None
+    acked_at: datetime | None
     created_at: datetime
 
     model_config = {"from_attributes": True}
+
+
+class NotificationAckRequest(BaseModel):
+    status: Literal["acked", "failed"]
+    ack_data: dict[str, Any] = Field(default_factory=dict)