Sink Health Monitoring: YAML Config Support

[joejohnson123]

Overview

Extend Sequin's YAML configuration system to support declaring monitors as part of sink definitions, enabling infrastructure-as-code workflows.

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

YAML Schema

Monitors are nested under sink definitions:

sinks:
  - name: my-webhook-sink
    type: http_push
    # ... existing sink config ...

    monitors:
      - name: "High failure rate"
        metric: failed_messages       # pending_messages | failed_messages
        type: threshold               # threshold | anomaly
        threshold: 100                # required for threshold type
        window: 60s                   # threshold debounce window (default: 60s)
        cooldown: 5m                  # min time between re-alerts (default: 5m)
        notify_on_recovery: true      # send resolved notification (default: true)
        channels:
          - type: slack
            webhook_url: "https://hooks.slack.com/services/..."
          - type: discord
            webhook_url: "https://discord.com/api/webhooks/..."
          - type: webhook
            url: "https://my-app.com/alerts"
            headers:
              Authorization: "Bearer token"
          - type: pagerduty
            routing_key: "your-key"
          - type: incident_io
            api_key: "inc_..."
            severity: critical

      - name: "Pending anomaly"
        metric: pending_messages
        type: anomaly
        sensitivity: medium            # low | medium | high
        cooldown: 15m
        channels:
          - type: slack
            webhook_url: "https://hooks.slack.com/services/..."

Duration Parsing

Support human-readable durations:

• 30s → 30 seconds
• 1m → 60 seconds
• 5m → 300 seconds
• 1h → 3600 seconds
• 60 → 60 seconds (plain integer = seconds)

Config Parser Changes

In Sequin.ConfigParser

1. After parsing a sink consumer, check for monitors key
2. For each monitor definition:
   • Validate required fields based on type
   • Parse duration strings to seconds
   • Create SinkMonitor changeset
   • Create nested SinkMonitorNotificationChannel changesets for each channel
3. On apply:
   • Upsert monitors by (sink_consumer_id, name) unique key
   • Remove monitors that exist in DB but not in YAML (declarative sync)
   • Upsert channels within each monitor

Validation Rules

• name is required
• metric must be one of pending_messages, failed_messages
• type must be one of threshold, anomaly
• If type: threshold: threshold is required (positive integer)
• If type: anomaly: sensitivity is required (low/medium/high)
• channels must have at least one entry
• Each channel must have type + required config fields for that type

Export Support

Also support exporting monitors back to YAML (for sequin export or API export):

• Add monitors key to sink export
• Convert seconds back to human-readable durations
• Mask secrets in channel configs (show **** for webhook URLs, keys)

Files to Modify

• lib/sequin/config_parser.ex — add monitor parsing within sink parsing
• test/sequin_web/controllers/yaml_controller_test.exs — add monitor YAML test cases

Implementation Notes

• Follow existing YAML parsing patterns in config_parser.ex
• Monitors should be processed AFTER the sink is created/found (they reference sink_consumer_id)
• Use the same Repo.transact pattern as other YAML entities
• Channel config validation should match the schema validation in the Ecto changeset
• Secret values in YAML should be handled consistently with existing secret handling (env var references, etc.)

Acceptance Criteria

• Monitors can be defined in YAML under sinks
• All monitor types and channel types parse correctly
• Duration strings parsed to seconds
• Validation errors are clear and reference the YAML path
• Declarative sync: monitors in DB but not YAML are removed
• Existing monitors are updated (not duplicated) on re-apply
• Export includes monitors with masked secrets
• Integration test: full YAML with monitors → apply → verify DB state