Sink Health Monitoring: Database Schema & Migrations

[joejohnson123]

Overview

Add the foundational database tables and Ecto schemas for the sink health monitoring feature.

Tables to Create

sink_monitors

Column	Type	Notes
id	binary_id	PK
sink_consumer_id	binary_id	FK → sink_consumers.id, on_delete: :delete_all
account_id	binary_id	FK → accounts
name	string	User-friendly label
enabled	boolean	Default true
metric	enum	:pending_messages | :failed_messages
monitor_type	enum	:threshold | :anomaly
threshold_value	integer	Only for :threshold — fire when metric ≥ this
threshold_window_seconds	integer	Debounce window. Default 60.
anomaly_sensitivity	enum	:low | :medium | :high — only for :anomaly
notify_on_recovery	boolean	Send "resolved" notification. Default true
cooldown_seconds	integer	Min time between re-alerts. Default 300 (5 min)
inserted_at / updated_at	timestamps

sink_monitor_notification_channels

Column	Type	Notes
id	binary_id	PK
sink_monitor_id	binary_id	FK → sink_monitors.id, on_delete: :delete_all
channel_type	enum	:webhook | :slack | :discord | :pagerduty | :incident_io
config	map (encrypted via Sequin.Encrypted.Map)	Channel-specific config
inserted_at / updated_at	timestamps

Channel config shapes:

• Webhook: %{url, headers, method}
• Slack: %{webhook_url}
• Discord: %{webhook_url}
• PagerDuty: %{routing_key}
• Incident.io: %{api_key, severity}

sink_monitor_alerts

Column	Type	Notes
id	binary_id	PK
sink_monitor_id	binary_id	FK → sink_monitors.id
status	enum	:triggered | :resolved
metric_value	integer	Value that triggered/resolved
message	text	Human-readable alert description
triggered_at	utc_datetime_usec
resolved_at	utc_datetime_usec	nullable
inserted_at / updated_at	timestamps

sink_monitor_metric_samples (for anomaly detection)

Column	Type	Notes
id	binary_id	PK
sink_monitor_id	binary_id	FK → sink_monitors.id, on_delete: :delete_all
value	integer	Sampled metric value
sampled_at	utc_datetime_usec

Auto-pruned to last 2 hours.

Files to Create

• priv/repo/migrations/YYYYMMDD_create_sink_monitors.exs
• lib/sequin/monitors/sink_monitor.ex
• lib/sequin/monitors/sink_monitor_notification_channel.ex
• lib/sequin/monitors/sink_monitor_alert.ex
• lib/sequin/monitors/sink_monitor_metric_sample.ex

Implementation Notes

• All tables in config_schema prefix (use Sequin.ConfigSchema)
• Use Ecto.Enum for enum fields
• Encrypt channel config with Sequin.Encrypted.Map
• Indexes: sink_consumer_id on monitors, sink_monitor_id on channels/alerts/samples, sampled_at on samples
• Unique index on (sink_consumer_id, name) for monitors
• Follow existing patterns (sink_consumer.ex, health_snapshot.ex)

Acceptance Criteria

• Migration runs cleanly (up and down)
• All 4 Ecto schemas compile with proper types
• Changesets validate required fields and enum values
• Threshold-specific fields validated only when monitor_type == :threshold
• Anomaly-specific fields validated only when monitor_type == :anomaly