From b0ac4e8a9b8a428092130633b413e60d2c3fa39b Mon Sep 17 00:00:00 2001 From: Curry Date: Sun, 12 Jul 2026 18:11:29 +0800 Subject: [PATCH] feat: notification ack fields + fix downstream NotificationSendResult consumers Cherry-pick of 1bdc0d8 (the real notification-ack commit), rebased onto current main's alembic head (d8e9f0a1b2c3 instead of the stale l2g3h4i5j6k7 parent, avoiding a second alembic head). Also fixes two downstream consumers broken by webhook_notifier.send() now returning NotificationSendResult instead of bool: - worker/tasks.py: success was a non-JSON-serializable dataclass instead of bool in the send_notification celery task's return value. - workflow/webhook_delivery.py: `if not delivered` never fired since a dataclass instance is always truthy, silently swallowing genuine webhook delivery failures. Both now reuse the existing _normalize_send_result() helper instead of new logic. Also fixes a stale assertion in test_notifiers.py (result is True -> result.success is True) and drops the stray backend/test_notification_ack.py scratch file with no references elsewhere. --- backend/api/v1/notifications.py | 65 ++++++++++++++++++- ...2n3o4p5q6r7_add_notification_ack_fields.py | 40 ++++++++++++ backend/models/notification.py | 18 +++-- backend/notifiers/base.py | 13 +++- backend/notifiers/webhook_notifier.py | 15 ++++- backend/pipeline/notifier_dispatch.py | 39 ++++++++--- backend/schemas/notification.py | 37 +++++++---- backend/worker/tasks.py | 3 +- backend/workflow/webhook_delivery.py | 3 +- tests/unit/test_notifiers.py | 2 +- 10 files changed, 195 insertions(+), 40 deletions(-) create mode 100644 backend/migrations/versions/m2n3o4p5q6r7_add_notification_ack_fields.py diff --git a/backend/api/v1/notifications.py b/backend/api/v1/notifications.py index 1cdbb3b..7e444b7 100644 --- a/backend/api/v1/notifications.py +++ b/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)) diff --git a/backend/migrations/versions/m2n3o4p5q6r7_add_notification_ack_fields.py b/backend/migrations/versions/m2n3o4p5q6r7_add_notification_ack_fields.py new file mode 100644 index 0000000..8a81cb7 --- /dev/null +++ b/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') diff --git a/backend/models/notification.py b/backend/models/notification.py index 63bbe4a..f0135f4 100644 --- a/backend/models/notification.py +++ b/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") diff --git a/backend/notifiers/base.py b/backend/notifiers/base.py index a91fb1e..a3f810e 100644 --- a/backend/notifiers/base.py +++ b/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: ... diff --git a/backend/notifiers/webhook_notifier.py b/backend/notifiers/webhook_notifier.py index 8f64288..f2a2006 100644 --- a/backend/notifiers/webhook_notifier.py +++ b/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], + }, + ) diff --git a/backend/pipeline/notifier_dispatch.py b/backend/pipeline/notifier_dispatch.py index ea9fb33..12357f6 100644 --- a/backend/pipeline/notifier_dispatch.py +++ b/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" diff --git a/backend/schemas/notification.py b/backend/schemas/notification.py index 91db1dc..1746b63 100644 --- a/backend/schemas/notification.py +++ b/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) diff --git a/backend/worker/tasks.py b/backend/worker/tasks.py index 8915160..b140394 100644 --- a/backend/worker/tasks.py +++ b/backend/worker/tasks.py @@ -141,6 +141,7 @@ async def _send_notification_async(rule_id: str, record_id: str) -> dict: from backend.models.record import CollectedRecord from backend.notifiers.base import NotificationPayload from backend.notifiers.registry import get_notifier + from backend.pipeline.notifier_dispatch import _normalize_send_result async with AsyncSessionLocal() as session: rule_result = await session.execute( @@ -165,5 +166,5 @@ async def _send_notification_async(rule_id: str, record_id: str) -> dict: data=record.normalized_data, ai_enrichment=record.ai_enrichment, ) - success = await notifier.send(rule.notifier_config, payload) + success, _ = _normalize_send_result(await notifier.send(rule.notifier_config, payload)) return {"success": success, "rule_id": rule_id, "record_id": record_id} diff --git a/backend/workflow/webhook_delivery.py b/backend/workflow/webhook_delivery.py index b450385..b669b31 100644 --- a/backend/workflow/webhook_delivery.py +++ b/backend/workflow/webhook_delivery.py @@ -6,6 +6,7 @@ from backend.notifiers.base import NotificationPayload from backend.notifiers.registry import get_notifier +from backend.pipeline.notifier_dispatch import _normalize_send_result WEBHOOK_DELIVERY_EVENT = "workflow.evidence_batch.ready" WEBHOOK_DELIVERY_PAYLOAD_SCHEMA = "workflow.webhook.evidence_batch.v1" @@ -44,7 +45,7 @@ async def execute_workflow_webhook_delivery( }, ) - delivered = await get_notifier("webhook").send(config, payload) + delivered, _ = _normalize_send_result(await get_notifier("webhook").send(config, payload)) if not delivered: raise WorkflowWebhookDeliveryError( code="webhook_delivery_failed", diff --git a/tests/unit/test_notifiers.py b/tests/unit/test_notifiers.py index 4211f39..241bace 100644 --- a/tests/unit/test_notifiers.py +++ b/tests/unit/test_notifiers.py @@ -61,7 +61,7 @@ async def test_webhook_notifier_send_success(): payload, ) - assert result is True + assert result.success is True @pytest.mark.asyncio