Graham McIntire
cc0a4f4ed6
Replaced fire-and-forget Task.start calls with reliable Oban workers for all alert notifications. **Why Oban instead of Task.start:** - ✅ Persistent - jobs survive crashes/restarts - ✅ Automatic retries with backoff (3 attempts) - ✅ Monitoring via Oban dashboard - ✅ Rate limiting via queue concurrency - ✅ Distributed coordination across servers **Changes:** - Created AlertNotificationWorker for trigger/acknowledge/resolve - Added notifications queue (concurrency: 10) to Oban config - Replaced 4 Task.start calls in alerts.ex and device_monitor_worker.ex - Updated 2 test assertions for string alert_type **Files changed:** - lib/towerops/workers/alert_notification_worker.ex (new) - lib/towerops/alerts.ex - lib/towerops/workers/device_monitor_worker.ex - config/dev.exs - config/runtime.exs - test/towerops/alerts_test.exs Notifications are now guaranteed to be delivered with automatic retry.
101 lines
2.7 KiB
Elixir
101 lines
2.7 KiB
Elixir
defmodule Towerops.Workers.AlertNotificationWorker do
|
|
@moduledoc """
|
|
Oban worker for sending alert notifications reliably.
|
|
|
|
Handles trigger, acknowledge, and resolve notifications with automatic
|
|
retries and error tracking.
|
|
"""
|
|
use Oban.Worker,
|
|
queue: :notifications,
|
|
max_attempts: 3,
|
|
priority: 1
|
|
|
|
alias Towerops.Alerts
|
|
alias Towerops.Devices
|
|
alias Towerops.PagerDuty.Notifier
|
|
|
|
require Logger
|
|
|
|
@impl Oban.Worker
|
|
def perform(%Oban.Job{args: %{"action" => "trigger", "alert_id" => alert_id}}) do
|
|
with {:ok, alert} <- fetch_alert(alert_id),
|
|
{:ok, device} <- fetch_device(alert.device_id) do
|
|
Notifier.notify_trigger(alert, device)
|
|
else
|
|
{:error, :alert_not_found} ->
|
|
Logger.warning("Alert #{alert_id} not found for notification")
|
|
:ok
|
|
|
|
{:error, :device_not_found} ->
|
|
Logger.warning("Device not found for alert #{alert_id}")
|
|
:ok
|
|
|
|
{:error, reason} ->
|
|
Logger.error("Failed to send trigger notification for alert #{alert_id}: #{inspect(reason)}")
|
|
{:error, reason}
|
|
end
|
|
end
|
|
|
|
def perform(%Oban.Job{args: %{"action" => "acknowledge", "alert_id" => alert_id}}) do
|
|
case fetch_alert(alert_id) do
|
|
{:ok, alert} ->
|
|
Notifier.notify_acknowledge(alert)
|
|
|
|
{:error, :alert_not_found} ->
|
|
Logger.warning("Alert #{alert_id} not found for acknowledge notification")
|
|
:ok
|
|
|
|
{:error, reason} ->
|
|
Logger.error("Failed to send acknowledge notification for alert #{alert_id}: #{inspect(reason)}")
|
|
{:error, reason}
|
|
end
|
|
end
|
|
|
|
def perform(%Oban.Job{args: %{"action" => "resolve", "alert_id" => alert_id}}) do
|
|
case fetch_alert(alert_id) do
|
|
{:ok, alert} ->
|
|
Notifier.notify_resolve(alert)
|
|
|
|
{:error, :alert_not_found} ->
|
|
Logger.warning("Alert #{alert_id} not found for resolve notification")
|
|
:ok
|
|
|
|
{:error, reason} ->
|
|
Logger.error("Failed to send resolve notification for alert #{alert_id}: #{inspect(reason)}")
|
|
{:error, reason}
|
|
end
|
|
end
|
|
|
|
# Helper to enqueue notification jobs
|
|
def enqueue_trigger(alert_id) do
|
|
%{action: "trigger", alert_id: alert_id}
|
|
|> new()
|
|
|> Oban.insert()
|
|
end
|
|
|
|
def enqueue_acknowledge(alert_id) do
|
|
%{action: "acknowledge", alert_id: alert_id}
|
|
|> new()
|
|
|> Oban.insert()
|
|
end
|
|
|
|
def enqueue_resolve(alert_id) do
|
|
%{action: "resolve", alert_id: alert_id}
|
|
|> new()
|
|
|> Oban.insert()
|
|
end
|
|
|
|
defp fetch_alert(alert_id) do
|
|
case Alerts.get_alert(alert_id) do
|
|
nil -> {:error, :alert_not_found}
|
|
alert -> {:ok, alert}
|
|
end
|
|
end
|
|
|
|
defp fetch_device(device_id) do
|
|
case Devices.get_device(device_id) do
|
|
nil -> {:error, :device_not_found}
|
|
device -> {:ok, device}
|
|
end
|
|
end
|
|
end
|