105 lines
2.8 KiB
Elixir
105 lines
2.8 KiB
Elixir
defmodule Towerops.Workers.AlertDigestWorker do
|
|
@moduledoc """
|
|
Oban worker that sends batched notification digests for rate-limited alerts.
|
|
|
|
Two modes:
|
|
- Cron mode (no user_id or user_id="__cron__"): finds all users with pending digests
|
|
and enqueues individual delivery jobs
|
|
- Delivery mode (specific user_id): sends the actual digest to that user
|
|
"""
|
|
use Oban.Worker,
|
|
queue: :notifications,
|
|
max_attempts: 3
|
|
|
|
import Ecto.Query
|
|
|
|
alias Towerops.Alerts
|
|
alias Towerops.Alerts.NotificationRateLimiter
|
|
|
|
require Logger
|
|
|
|
@impl Oban.Worker
|
|
def perform(%Oban.Job{args: %{"user_id" => "__cron__"}}) do
|
|
# Cron mode: find all pending digests and enqueue delivery
|
|
|
|
pending =
|
|
Towerops.Repo.all(
|
|
from(d in Towerops.Alerts.NotificationDigest,
|
|
where: d.digest_sent == false,
|
|
where: fragment("array_length(?, 1) > 0", d.suppressed_alert_ids),
|
|
select: d.user_id,
|
|
distinct: true
|
|
)
|
|
)
|
|
|
|
Enum.each(pending, fn user_id ->
|
|
case %{user_id: user_id}
|
|
|> new()
|
|
|> Oban.insert() do
|
|
{:ok, _job} ->
|
|
:ok
|
|
|
|
{:error, changeset} ->
|
|
Logger.error("Failed to enqueue digest job for user_id=#{user_id}: #{inspect(changeset)}")
|
|
:error
|
|
end
|
|
end)
|
|
|
|
:ok
|
|
end
|
|
|
|
def perform(%Oban.Job{args: %{"user_id" => user_id}}) when user_id != "__cron__" do
|
|
case NotificationRateLimiter.get_pending_digest(user_id) do
|
|
nil ->
|
|
:ok
|
|
|
|
digest ->
|
|
alert_ids = digest.suppressed_alert_ids || []
|
|
|
|
with [_ | _] <- alert_ids,
|
|
alerts = Alerts.list_alerts_by_ids(alert_ids),
|
|
[_ | _] <- alerts do
|
|
send_digest_notification(alerts)
|
|
NotificationRateLimiter.mark_digest_sent(digest)
|
|
end
|
|
|
|
:ok
|
|
end
|
|
end
|
|
|
|
@doc """
|
|
Enqueue digest delivery for a specific user.
|
|
Schedules delivery 5 minutes from now to batch more alerts.
|
|
"""
|
|
def enqueue(user_id) do
|
|
case %{user_id: user_id}
|
|
|> new(schedule_in: 300, unique: [period: 300, keys: [:user_id]])
|
|
|> Oban.insert() do
|
|
{:ok, _job} = result ->
|
|
result
|
|
|
|
{:error, changeset} = error ->
|
|
Logger.error("Failed to enqueue digest job for user_id=#{user_id}: #{inspect(changeset)}")
|
|
error
|
|
end
|
|
end
|
|
|
|
defp send_digest_notification(alerts) do
|
|
count = length(alerts)
|
|
|
|
device_names =
|
|
alerts
|
|
|> Enum.map(fn a ->
|
|
case a.device do
|
|
nil -> "unknown"
|
|
device -> device.name || device.ip_address || "unknown"
|
|
end
|
|
end)
|
|
|> Enum.uniq()
|
|
|
|
Logger.info("Sending notification digest: #{count} suppressed alerts for devices: #{Enum.join(device_names, ", ")}",
|
|
alert_count: count,
|
|
alert_ids: Enum.map(alerts, & &1.id)
|
|
)
|
|
end
|
|
end
|