towerops/lib/towerops/alerts/site_correlation.ex

defmodule Towerops.Alerts.SiteCorrelation do
  @moduledoc """
  Correlates device-down alerts at the same site into a single "Site Outage".

  When >N devices at the same site go down within T seconds, creates a
  single site outage record with one notification instead of N individual ones.

  ## Thresholds

  - Site correlation threshold: 3+ devices at same site within 120 seconds
  - These are configurable via application env

  ## Flow

  1. DeviceMonitorWorker detects device down
  2. Before creating individual alert, calls `check_site_correlation/2`
  3. If site outage threshold met, returns `{:site_outage, outage}` — caller
     links alert to outage and suppresses individual notification
  4. If not, returns `:no_correlation` — normal alert flow
  """

  import Ecto.Query

  alias Towerops.Alerts.Alert
  alias Towerops.Alerts.SiteOutage
  alias Towerops.Devices.Device
  alias Towerops.Repo

  require Logger

  # Default: 3+ devices at same site within 120s = site outage
  @default_device_threshold 3
  @default_time_window_seconds 120

  @doc """
  Check if this device going down should be correlated into a site outage.

  Returns:
  - `{:site_outage, outage}` — device is part of a site outage, suppress individual notification
  - `:no_correlation` — proceed with normal alert flow
  """
  @spec check_site_correlation(Device.t(), DateTime.t()) ::
          {:site_outage, SiteOutage.t()} | :no_correlation
  def check_site_correlation(%Device{site_id: nil}, _now), do: :no_correlation

  def check_site_correlation(%Device{} = device, now) do
    threshold = device_threshold()
    window = time_window_seconds()
    cutoff = DateTime.add(now, -window, :second)

    # Check for existing active site outage
    case get_active_outage(device.site_id) do
      %SiteOutage{} = outage ->
        # Already in site outage mode — just increment
        update_outage_count(outage, device)
        {:site_outage, outage}

      nil ->
        # Count recent device_down alerts at this site
        recent_down_count = count_recent_site_alerts(device.site_id, cutoff)

        # +1 for the current device about to go down
        if recent_down_count + 1 >= threshold do
          # Threshold met — create site outage
          case create_site_outage(device, now, recent_down_count + 1) do
            {:ok, outage} ->
              # Retroactively link existing alerts to this outage
              link_existing_alerts(device.site_id, cutoff, outage.id)
              {:site_outage, outage}

            {:error, reason} ->
              Logger.error("Failed to create site outage: #{inspect(reason)}")
              :no_correlation
          end
        else
          :no_correlation
        end
    end
  end

  @doc """
  Check if a site outage should be resolved (all devices back up).
  Called when a device comes back up.
  """
  @spec check_outage_resolution(Device.t()) :: :ok
  def check_outage_resolution(%Device{site_id: nil}), do: :ok

  def check_outage_resolution(%Device{} = device) do
    case get_active_outage(device.site_id) do
      nil ->
        :ok

      outage ->
        # Check if any devices at this site are still down
        still_down = count_active_down_at_site(device.site_id)

        if still_down <= 0 do
          resolve_outage(outage)
        else
          # Update the outage message with current count
          outage
          |> SiteOutage.changeset(%{
            device_count: still_down,
            message: "Site outage: #{still_down} device(s) still down at site"
          })
          |> Repo.update()
        end

        :ok
    end
  end

  @doc """
  Get the active site outage for a site, if any.
  """
  @spec get_active_outage(String.t()) :: SiteOutage.t() | nil
  def get_active_outage(site_id) do
    Repo.one(
      from(o in SiteOutage,
        where: o.site_id == ^site_id,
        where: is_nil(o.resolved_at),
        limit: 1
      )
    )
  end

  ## Private

  defp count_recent_site_alerts(site_id, cutoff) do
    Repo.aggregate(
      from(a in Alert,
        join: d in Device,
        on: d.id == a.device_id,
        where: d.site_id == ^site_id,
        where: a.alert_type == "device_down",
        where: is_nil(a.resolved_at),
        where: a.triggered_at >= ^cutoff
      ),
      :count
    )
  end

  defp count_active_down_at_site(site_id) do
    Repo.aggregate(
      from(a in Alert,
        join: d in Device,
        on: d.id == a.device_id,
        where: d.site_id == ^site_id,
        where: a.alert_type == "device_down",
        where: is_nil(a.resolved_at)
      ),
      :count
    )
  end

  defp create_site_outage(device, now, device_count) do
    total_at_site = count_devices_at_site(device.site_id)
    site = Repo.get(Towerops.Sites.Site, device.site_id)
    site_name = if site, do: site.name, else: "Unknown Site"

    %SiteOutage{}
    |> SiteOutage.changeset(%{
      site_id: device.site_id,
      organization_id: device.organization_id,
      started_at: DateTime.truncate(now, :second),
      device_count: device_count,
      total_devices_at_site: total_at_site,
      message: "Site outage at #{site_name}: #{device_count}/#{total_at_site} devices down"
    })
    |> Repo.insert()
  end

  defp update_outage_count(outage, _device) do
    new_count = (outage.device_count || 0) + 1

    outage
    |> SiteOutage.changeset(%{device_count: new_count})
    |> Repo.update()
  end

  defp link_existing_alerts(site_id, cutoff, outage_id) do
    Repo.update_all(
      from(a in Alert,
        join: d in Device,
        on: d.id == a.device_id,
        where: d.site_id == ^site_id,
        where: a.alert_type == "device_down",
        where: is_nil(a.resolved_at),
        where: a.triggered_at >= ^cutoff
      ),
      set: [site_outage_id: outage_id, storm_suppressed: true]
    )
  end

  defp resolve_outage(outage) do
    now = DateTime.truncate(DateTime.utc_now(), :second)

    outage
    |> SiteOutage.changeset(%{resolved_at: now})
    |> Repo.update()

    Logger.info("Site outage resolved: #{outage.id} at site #{outage.site_id}")

    # Broadcast resolution
    Phoenix.PubSub.broadcast(
      Towerops.PubSub,
      "alerts:org:#{outage.organization_id}:site_outage",
      {:site_outage_resolved, outage}
    )
  end

  defp count_devices_at_site(site_id) do
    Repo.aggregate(
      from(d in Device, where: d.site_id == ^site_id),
      :count
    )
  end

  defp device_threshold do
    :towerops
    |> Application.get_env(:alert_storm, [])
    |> Keyword.get(:site_correlation_threshold, @default_device_threshold)
  end

  defp time_window_seconds do
    :towerops
    |> Application.get_env(:alert_storm, [])
    |> Keyword.get(:site_correlation_window_seconds, @default_time_window_seconds)
  end
end