notifications: dispatch newly-opened findings to email and webhook channels (#60)

.env.example

@@ -66,6 +66,28 @@ ICEBERG_BOOTSTRAP_ADMIN_EMAIL=you@example.test
 # corpus, lower it to see what the rules are nearly matching.
 ICEBERG_CONFIDENCE_THRESHOLD=0.5
 
+# ─── Notifications (api role) ─────────────────────────────────────────────────
+# Channels are configured in the console; these are the deployment facts behind
+# them (docs/notifications.md). Delivery runs in the API's maintenance loop, so
+# none of this reaches an engine.
+#
+# Leave ICEBERG_SMTP_HOST empty to disable email: an email channel's deliveries
+# then fail loudly naming this variable, rather than quietly going nowhere.
+ICEBERG_SMTP_HOST=
+ICEBERG_SMTP_PORT=587
+# Omit both for an unauthenticated relay.
+ICEBERG_SMTP_USERNAME=
+ICEBERG_SMTP_PASSWORD=
+# Certificates are verified. Only turn this off for a relay on localhost.
+ICEBERG_SMTP_STARTTLS=true
+ICEBERG_SMTP_FROM=icebergsst@localhost
+
+# Attempts before a delivery is given up on. The row is kept either way — giving
+# up is recorded, never silent.
+ICEBERG_NOTIFICATION_MAX_ATTEMPTS=5
+# First retry delay in seconds; doubles each attempt.
+ICEBERG_NOTIFICATION_RETRY_BACKOFF_SECONDS=60
+
 # ─── Engine role ──────────────────────────────────────────────────────────────
 # The per-engine API credential. In production this is minted at deploy time
 # (docs/security.md § Bootstrap), never defaulted into an image.

apps/api/src/iceberg_api/maintenance.py

@@ -24,27 +24,40 @@
 import structlog
 from iceberg_core.config import ApiSettings, get_api_settings
 from iceberg_core.db import session_scope
+from iceberg_core.secrets import SecretStore, build_secret_store
 
 from iceberg_api import suppressions
 from iceberg_api.dispatch import Dispatcher, build_dispatcher
+from iceberg_api.notifications import dispatch as notification_dispatch
 from iceberg_api.scans import service
 from iceberg_api.scheduler import postgres_advisory_lock, tick
 from iceberg_api.scheduler_launcher import build_launcher
 
 logger = structlog.get_logger()
 
 
-def run_once(dispatcher: Dispatcher, *, now: datetime | None = None) -> None:
-    """One maintenance round: schedules, reclaim, then the safety sweeps.
+def run_once(
+    dispatcher: Dispatcher,
+    *,
+    now: datetime | None = None,
+    settings: ApiSettings | None = None,
+    store: SecretStore | None = None,
+) -> None:
+    """One maintenance round: schedules, reclaim, the safety sweeps, then alerts.
 
     Leadership is held on a session of its own for the whole round. Holding it on
     the working session would not work: ``pg_try_advisory_xact_lock`` releases at
     transaction end, and the scan launcher commits mid-tick — so the lock would be
     gone after the first schedule fired, and every replica would run the rest of
     the round at once. The guard session runs no other statements, so its
     transaction — and the lock — spans everything below.
+
+    ``settings``/``store`` are injectable so a test can drive a round without a
+    configured SMTP relay; both default to the process configuration.
     """
     at = now or datetime.now(UTC)
+    resolved = settings or get_api_settings()
+    secret_store = store or build_secret_store(resolved)
     with session_scope() as guard:
         if not postgres_advisory_lock(guard):
             # Another replica is doing this round — the whole round, sweeps
@@ -69,6 +82,12 @@ def run_once(dispatcher: Dispatcher, *, now: datetime | None = None) -> None:
         # that source, which for a weekly cadence is six days of silence (ADR 0008).
         with session_scope() as db:
             suppressions.release_lapsed(db, now=at)
+        # Announcements queued by reconciliation (#60). Sending here rather than at
+        # ingest means a webhook that hangs for its full timeout delays an alert
+        # instead of an engine's result submission, and a receiver that is down
+        # gets retried on the next beat instead of losing the alert.
+        with session_scope() as db:
+            notification_dispatch.deliver_pending(db, resolved, secret_store, now=at)
 
 
 def _already_leader(db: object) -> bool:

apps/api/src/iceberg_api/migrations/versions/0006_notification_delivery.py

@@ -0,0 +1,100 @@
+"""Notification delivery outbox
+
+Dispatch (#60) writes its intention to announce a finding in the same transaction
+that opens the finding, and sends afterwards from the maintenance loop. That needs
+a table: a row per (channel, finding, scan), carrying attempt count, when it next
+becomes due, and the error that ended it if it never went out.
+
+The unique constraint is the load-bearing part. Enqueueing is re-run whenever a
+stalled scan is re-finalized, so without it the safety sweep would announce the
+same secret again every beat.
+
+Revision ID: 0006
+Revises: 0005
+Create Date: 2026-07-31 07:10:00.000000+00:00
+"""
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+import sqlmodel.sql.sqltypes  # AutoString appears in autogenerated column ops
+from alembic import op
+
+revision: str = "0006"
+down_revision: str | None = "0005"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "notification_delivery",
+        sa.Column("id", sa.Uuid(), nullable=False),
+        sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
+        sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False),
+        sa.Column("channel_id", sa.Uuid(), nullable=False),
+        sa.Column("finding_id", sa.Uuid(), nullable=False),
+        sa.Column("scan_id", sa.Uuid(), nullable=False),
+        sa.Column(
+            "status",
+            sa.Enum(
+                "pending",
+                "delivered",
+                "failed",
+                name="notification_delivery_status",
+                native_enum=False,
+                length=32,
+            ),
+            nullable=False,
+        ),
+        sa.Column("attempts", sa.Integer(), nullable=False),
+        sa.Column("next_attempt_at", sa.DateTime(timezone=True), nullable=False),
+        sa.Column("delivered_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column("last_error", sqlmodel.sql.sqltypes.AutoString(length=500), nullable=True),
+        # CASCADE throughout: a delivery record is about a finding, not part of
+        # its history, so removing the finding, the channel, or the scan should
+        # take the record rather than be blocked by it (contrast finding's own
+        # RESTRICT scan references, added in 0004).
+        sa.ForeignKeyConstraint(
+            ["channel_id"],
+            ["notification_channel.id"],
+            name=op.f("fk_notification_delivery_channel_id_notification_channel"),
+            ondelete="CASCADE",
+        ),
+        sa.ForeignKeyConstraint(
+            ["finding_id"],
+            ["finding.id"],
+            name=op.f("fk_notification_delivery_finding_id_finding"),
+            ondelete="CASCADE",
+        ),
+        sa.ForeignKeyConstraint(
+            ["scan_id"],
+            ["scan.id"],
+            name=op.f("fk_notification_delivery_scan_id_scan"),
+            ondelete="CASCADE",
+        ),
+        sa.PrimaryKeyConstraint("id", name=op.f("pk_notification_delivery")),
+        # What makes re-enqueueing idempotent, and therefore what stops the
+        # stalled-scan sweep re-announcing every secret it re-finalizes.
+        sa.UniqueConstraint(
+            "channel_id",
+            "finding_id",
+            "scan_id",
+            name="uq_notification_delivery_channel_finding_scan",
+        ),
+    )
+    # The delivery loop's only query: pending rows that are due.
+    op.create_index(
+        "ix_notification_delivery_status_next_attempt_at",
+        "notification_delivery",
+        ["status", "next_attempt_at"],
+        unique=False,
+    )
+
+
+def downgrade() -> None:
+    op.drop_index(
+        "ix_notification_delivery_status_next_attempt_at",
+        table_name="notification_delivery",
+    )
+    op.drop_table("notification_delivery")