Sink Health Monitoring: Notification Dispatcher & Webhook Channel

[joejohnson123]

Overview

Implement the notification dispatch system that sends alerts to configured channels when monitors fire. Start with the generic webhook channel as the foundation.

Depends on: #1 (Schema), #2 (Context Module)

Dispatcher: Sequin.Monitors.NotificationDispatcher

Core Function

@spec dispatch(SinkMonitorAlert.t(), [SinkMonitorNotificationChannel.t()]) :: :ok
def dispatch(alert, channels)

• Takes an alert and list of notification channels
• Dispatches to each channel concurrently via Task.Supervisor
• Logs success/failure for each channel
• Does NOT fail if individual channels fail (best-effort delivery)

Alert Payload (shared across all channels)

%{
  monitor_name: "High failure rate",
  sink_name: "my-webhook-sink",
  sink_id: "uuid",
  metric: "failed_messages",
  metric_value: 150,
  threshold_value: 100,  # nil for anomaly monitors
  status: "triggered",   # or "resolved"
  message: "Failed messages (150) exceeded threshold (100) for sink my-webhook-sink",
  triggered_at: "2026-02-26T13:00:00Z",
  resolved_at: nil,       # populated for resolution notifications
  sink_url: "https://app.sequinstream.com/sinks/http_push/uuid"
}

Channel Behaviour

@callback send_notification(alert_payload :: map(), config :: map()) :: :ok | {:error, term()}

Each channel type implements this behaviour.

Webhook Channel: Sequin.Monitors.Channels.Webhook

Implementation

• HTTP POST to configured url
• JSON body with the alert payload
• Custom headers from config (e.g., auth tokens)
• Configurable HTTP method (default POST)
• Timeout: 10 seconds
• Use Req (already in deps)

Request Format

POST https://example.com/alerts
Content-Type: application/json
X-Custom-Header: value

{
  "monitor_name": "High failure rate",
  "sink_name": "my-webhook-sink",
  "metric": "failed_messages",
  "metric_value": 150,
  "status": "triggered",
  "message": "Failed messages (150) exceeded threshold (100)...",
  "triggered_at": "2026-02-26T13:00:00Z"
}

Files to Create

• lib/sequin/monitors/notification_dispatcher.ex
• lib/sequin/monitors/channels/webhook.ex
• test/sequin/monitors/notification_dispatcher_test.exs
• test/sequin/monitors/channels/webhook_test.exs

Implementation Notes

• Use a Task.Supervisor (register one in application.ex) for concurrent dispatch
• Each channel module lives under Sequin.Monitors.Channels.*
• Dispatcher builds the payload from the alert + preloaded monitor + consumer
• Log at info level for successful dispatches, error for failures
• Consider adding a last_notified_at field to alerts for debugging

Acceptance Criteria

• Dispatcher sends to all configured channels concurrently
• Individual channel failure doesn't affect others
• Webhook channel sends correct JSON payload
• Custom headers are included
• Timeout handling (doesn't hang forever)
• Both triggered and resolved alert types are formatted correctly
• Tests use Bypass or Req.Test for HTTP assertions