towerops/lib/towerops/alerts.ex

defmodule Towerops.Alerts do
  @moduledoc """
  The Alerts context.
  """

  import Ecto.Query

  alias Towerops.Alerts.Alert
  alias Towerops.Gaiia.ImpactAnalysis
  alias Towerops.Repo
  alias Towerops.Workers.AlertNotificationWorker
  alias ToweropsWeb.GraphQL.Subscriptions

  defp broadcast_alert_change(%Alert{organization_id: org_id}) when not is_nil(org_id) do
    Phoenix.PubSub.broadcast(
      Towerops.PubSub,
      "organization:#{org_id}:alerts",
      {:alert_changed, org_id}
    )
  end

  defp broadcast_alert_change(_alert), do: :ok

  @doc """
  Creates an alert for device status change.

  For device_down alerts, automatically computes and stores Gaiia impact analysis
  if the device belongs to an organization with a Gaiia integration.

  Automatically populates organization_id from the device for query performance.
  """
  @spec create_alert(map()) :: {:ok, Alert.t()} | {:error, Ecto.Changeset.t()}
  def create_alert(attrs) do
    attrs =
      attrs
      |> normalize_attrs_alert_type()
      |> enrich_with_organization_id()

    with {:ok, alert} <- Repo.insert(Alert.changeset(%Alert{}, attrs)) do
      maybe_compute_gaiia_impact(alert)
      broadcast_alert_change(alert)
      Subscriptions.publish_alert_event(alert, "created")
      {:ok, alert}
    end
  end

  # Populate organization_id from device for denormalized query performance
  defp enrich_with_organization_id(%{device_id: device_id} = attrs) when not is_nil(device_id) do
    case Towerops.Devices.get_device(device_id) do
      nil -> attrs
      device -> Map.put(attrs, :organization_id, device.organization_id)
    end
  end

  defp enrich_with_organization_id(attrs), do: attrs

  # Normalize alert_type in attrs map for create/update operations
  defp normalize_attrs_alert_type(%{alert_type: alert_type} = attrs) when is_atom(alert_type) do
    %{attrs | alert_type: to_string(alert_type)}
  end

  defp normalize_attrs_alert_type(attrs), do: attrs

  # Normalize alert_type to string for backward compatibility with tests
  defp normalize_alert_type(alert_type) when is_atom(alert_type), do: to_string(alert_type)
  defp normalize_alert_type(alert_type) when is_binary(alert_type), do: alert_type

  defp maybe_compute_gaiia_impact(%Alert{alert_type: "device_down", device_id: device_id} = alert)
       when not is_nil(device_id) do
    device = Towerops.Devices.get_device(device_id)

    if device do
      impact = ImpactAnalysis.analyze_device_impact(device.organization_id, device_id)

      if impact.total_subscribers > 0 do
        store_gaiia_impact(alert, ImpactAnalysis.to_json(impact))
      end
    end
  rescue
    # Only rescue specific expected errors, log unexpected ones
    error in [Ecto.QueryError, DBConnection.ConnectionError] ->
      require Logger

      Logger.warning(
        "Failed to compute GAIIA impact for alert #{alert.id} due to database error: #{inspect(error)}",
        alert_id: alert.id,
        device_id: device_id
      )

      :ok

    error ->
      require Logger

      Logger.error(
        "Unexpected error computing GAIIA impact for alert #{alert.id}: #{inspect(error)}",
        alert_id: alert.id,
        device_id: device_id,
        error: error
      )

      :ok
  end

  defp maybe_compute_gaiia_impact(_alert), do: :ok

  @doc """
  Returns the list of alerts for an device.
  """
  @spec list_devices_alerts(String.t(), integer()) :: [Alert.t()]
  def list_devices_alerts(device_id, limit \\ 100) do
    Repo.all(
      from(a in Alert,
        where: a.device_id == ^device_id,
        order_by: [desc: a.triggered_at],
        limit: ^limit,
        preload: [:device, :acknowledged_by]
      )
    )
  end

  @doc """
  Returns the list of active (unresolved) alerts for an organization.
  Only returns device_down alerts, as device_up alerts are informational.

  Uses denormalized organization_id for fast single-table query with composite index.
  """
  @spec list_organization_active_alerts(String.t()) :: [Alert.t()]
  def list_organization_active_alerts(organization_id) do
    Repo.all(
      from(a in Alert,
        where: a.organization_id == ^organization_id,
        where: a.alert_type == "device_down",
        where: is_nil(a.resolved_at),
        order_by: [desc: a.triggered_at],
        preload: [:acknowledged_by, device: [:site]]
      )
    )
  end

  @doc """
  Returns the count of active (unresolved) alerts for an organization.

  Uses denormalized organization_id for fast single-table query with composite index.
  """
  @spec count_active_alerts(String.t()) :: integer()
  def count_active_alerts(organization_id) do
    Repo.aggregate(
      from(a in Alert,
        where: a.organization_id == ^organization_id,
        where: a.alert_type == "device_down",
        where: is_nil(a.resolved_at)
      ),
      :count
    )
  end

  @doc "Returns the count of active alerts for devices at a specific site."
  def count_site_active_alerts(site_id) do
    Repo.aggregate(
      from(a in Alert,
        join: e in assoc(a, :device),
        where: e.site_id == ^site_id,
        where: a.alert_type == "device_down",
        where: is_nil(a.resolved_at)
      ),
      :count
    )
  end

  @doc "Returns active (unresolved) alerts for devices at a specific site."
  def list_site_active_alerts(site_id) do
    Repo.all(
      from(a in Alert,
        join: e in assoc(a, :device),
        where: e.site_id == ^site_id,
        where: a.alert_type == "device_down",
        where: is_nil(a.resolved_at),
        order_by: [desc: a.triggered_at],
        preload: [device: e]
      )
    )
  end

  @doc """
  Returns the list of all alerts for an organization.

  Accepts either an integer limit or a map of filters:
  - limit (integer): Max number of alerts to return
  - filters (map):
    - status: Filter by status ("active", "acknowledged", "resolved")
    - limit: Max number of alerts to return
  """
  @spec list_organization_alerts(String.t(), integer() | map()) :: [Alert.t()]
  def list_organization_alerts(organization_id, limit) when is_integer(limit) do
    list_organization_alerts(organization_id, %{"limit" => limit})
  end

  def list_organization_alerts(organization_id, filters) when is_map(filters) do
    limit = filters["limit"] || 100

    from(a in Alert,
      where: a.organization_id == ^organization_id,
      order_by: [desc: a.triggered_at],
      limit: ^limit,
      preload: [device: [:site], acknowledged_by: []]
    )
    |> filter_by_status(filters["status"])
    |> filter_by_device(filters["device_id"])
    |> Repo.all()
  end

  defp filter_by_status(query, "active"), do: where(query, [a], is_nil(a.resolved_at) and is_nil(a.acknowledged_at))

  defp filter_by_status(query, "acknowledged"),
    do: where(query, [a], is_nil(a.resolved_at) and not is_nil(a.acknowledged_at))

  defp filter_by_status(query, "resolved"), do: where(query, [a], not is_nil(a.resolved_at))

  defp filter_by_status(query, _), do: query

  defp filter_by_device(query, nil), do: query
  defp filter_by_device(query, device_id), do: where(query, [a], a.device_id == ^device_id)

  @doc """
  Returns alerts for multiple organizations (for GDPR data access).
  Accepts a `since` option to filter alerts created after a specific date.
  """
  def list_alerts_for_organizations(organization_ids, opts \\ []) when is_list(organization_ids) do
    since = Keyword.get(opts, :since)

    query =
      from(a in Alert,
        join: e in assoc(a, :device),
        join: s in assoc(e, :site),
        where: s.organization_id in ^organization_ids,
        order_by: [desc: a.triggered_at],
        preload: [device: {e, site: s}]
      )

    query =
      if since do
        where(query, [a], a.inserted_at >= ^since)
      else
        query
      end

    Repo.all(query)
  end

  @doc """
  Gets a single alert.
        where: s.organization_id == ^organization_id,
        order_by: [desc: a.triggered_at],
        limit: ^limit,
        preload: [device: {e, site: s}, acknowledged_by: []]
      )
    )
  end

  @doc \"""
  Gets a single alert.
  """
  def get_alert!(id) do
    Alert
    |> Repo.get!(id)
    |> Repo.preload([:device, :acknowledged_by])
  end

  @doc """
  Gets a single alert. Returns nil if not found.
  """
  def get_alert(id) do
    case Repo.get(Alert, id) do
      nil -> nil
      alert -> Repo.preload(alert, [:device, :acknowledged_by])
    end
  end

  @doc """
  Acknowledges an alert.
  """
  def acknowledge_alert(alert, user_id) do
    result =
      alert
      |> Alert.changeset(%{
        acknowledged_at: DateTime.truncate(DateTime.utc_now(), :second),
        acknowledged_by_id: user_id
      })
      |> Repo.update()

    case result do
      {:ok, updated_alert} ->
        AlertNotificationWorker.enqueue_acknowledge(updated_alert.id)
        Subscriptions.publish_alert_event(updated_alert, "acknowledged")
        {:ok, updated_alert}

      error ->
        error
    end
  end

  @doc """
  Resolves an alert.
  """
  def resolve_alert(alert) do
    result =
      alert
      |> Alert.changeset(%{resolved_at: DateTime.truncate(DateTime.utc_now(), :second)})
      |> Repo.update()

    case result do
      {:ok, updated_alert} ->
        AlertNotificationWorker.enqueue_resolve(updated_alert.id)
        broadcast_alert_change(updated_alert)
        Subscriptions.publish_alert_event(updated_alert, "resolved")
        {:ok, updated_alert}

      error ->
        error
    end
  end

  @doc """
  Resolves an alert without notifying PagerDuty (used when PagerDuty is the source).
  """
  def resolve_alert_silent(alert) do
    result =
      alert
      |> Alert.changeset(%{resolved_at: DateTime.truncate(DateTime.utc_now(), :second)})
      |> Repo.update()

    case result do
      {:ok, updated_alert} ->
        broadcast_alert_change(updated_alert)
        {:ok, updated_alert}

      error ->
        error
    end
  end

  @doc """
  Acknowledges an alert without notifying PagerDuty (used when PagerDuty is the source).
  """
  def acknowledge_alert_silent(alert) do
    alert
    |> Alert.changeset(%{acknowledged_at: DateTime.truncate(DateTime.utc_now(), :second)})
    |> Repo.update()
  end

  @doc """
  Stores Gaiia impact analysis data on an alert record.
  """
  def store_gaiia_impact(%Alert{} = alert, impact_data) when is_map(impact_data) do
    alert
    |> Alert.changeset(%{gaiia_impact: impact_data})
    |> Repo.update()
  end

  @doc """
  Checks if there's an active alert of the same type for the device.
  """
  def has_active_alert?(device_id, alert_type) do
    alert_type = normalize_alert_type(alert_type)

    Repo.exists?(
      from(a in Alert,
        where: a.device_id == ^device_id,
        where: a.alert_type == ^alert_type,
        where: is_nil(a.resolved_at)
      )
    )
  end

  @doc """
  Gets the latest active alert of a specific type for device.
  """
  def get_active_alert(device_id, alert_type) do
    alert_type = normalize_alert_type(alert_type)

    Repo.one(
      from(a in Alert,
        where: a.device_id == ^device_id,
        where: a.alert_type == ^alert_type,
        where: is_nil(a.resolved_at),
        order_by: [desc: a.triggered_at],
        limit: 1
      )
    )
  end

  @doc """
  Checks if there's an active alert for a check.
  """
  def has_active_check_alert?(check_id) do
    Repo.exists?(
      from(a in Alert,
        where: a.check_id == ^check_id,
        where: is_nil(a.resolved_at)
      )
    )
  end

  @doc """
  Resolves all active alerts for a check.
  """
  def resolve_check_alerts(check_id) do
    now = DateTime.truncate(DateTime.utc_now(), :second)

    Repo.update_all(from(a in Alert, where: a.check_id == ^check_id, where: is_nil(a.resolved_at)),
      set: [resolved_at: now]
    )
  end
end