towerops/lib/towerops_web/live/graph_live/polling.ex

defmodule ToweropsWeb.GraphLive.Polling do
  @moduledoc """
  SNMP polling functions for live-mode graph data collection.

  Extracted from `ToweropsWeb.GraphLive.Show` to keep the LiveView module
  focused on socket lifecycle, assigns management, and UI state transitions.
  """

  alias Towerops.Devices
  alias Towerops.Snmp

  require Logger

  @max_live_points 300

  @doc """
  Build SNMP client options from a device struct.
  """
  def build_snmp_client_opts(device) do
    snmp_config = Devices.get_snmp_config(device)

    [
      ip: device.ip_address,
      community: snmp_config.community,
      version: snmp_config.version,
      port: device.snmp_port,
      timeout: 3_000
    ]
  end

  @doc """
  Poll sensors for live mode, dispatching based on sensor type.
  """
  def poll_sensors(assigns, client_opts, timestamp_ms) do
    if assigns.sensor_type == "traffic" do
      poll_traffic(assigns, client_opts, timestamp_ms)
    else
      poll_regular_sensors(assigns, client_opts, timestamp_ms)
    end
  end

  @doc """
  Poll regular (non-traffic) sensors in parallel via Task.async_stream.
  """
  def poll_regular_sensors(assigns, client_opts, timestamp_ms) do
    sensors = get_sensors_for_live_mode(assigns)

    sensors
    |> Task.async_stream(&poll_sensor_to_data_point(&1, client_opts, timestamp_ms),
      max_concurrency: 10,
      timeout: 4_000,
      on_timeout: :kill_task
    )
    |> Enum.reduce([], fn
      {:ok, nil}, acc -> acc
      {:ok, point}, acc -> [point | acc]
      {:exit, _reason}, acc -> acc
    end)
  end

  @doc """
  Poll traffic stats (used from non-state-updating path).
  Requires two readings to calculate BPS; uses `previous_interface_stats` from assigns.
  """
  def poll_traffic(assigns, client_opts, timestamp_ms) do
    device = Snmp.get_device_with_associations(assigns.device_id)

    if is_nil(device) || Enum.empty?(device.interfaces) do
      []
    else
      current_stats = poll_interface_stats_parallel(device.interfaces, client_opts)

      if assigns.previous_interface_stats do
        calculate_traffic_bps_from_stats(
          assigns.previous_interface_stats,
          current_stats,
          timestamp_ms
        )
      else
        []
      end
    end
  end

  @doc """
  Poll traffic with socket state update.
  Returns `{data_points, updated_stats}` for the caller to assign back.
  """
  def poll_traffic_with_state(assigns, client_opts, timestamp_ms) do
    device = Snmp.get_device_with_associations(assigns.device_id)

    if is_nil(device) || Enum.empty?(device.interfaces) do
      {[], nil}
    else
      current_stats = poll_interface_stats_parallel(device.interfaces, client_opts)

      data_points =
        if assigns.previous_interface_stats do
          calculate_traffic_bps_from_stats(
            assigns.previous_interface_stats,
            current_stats,
            timestamp_ms
          )
        else
          []
        end

      {data_points, current_stats}
    end
  end

  @doc """
  Request live poll from agent via PubSub and wait for response.
  """
  def request_agent_live_poll(device_id, agent_token_id, assigns) do
    reply_topic = "live_poll_reply:#{:erlang.unique_integer([:positive])}"
    :ok = Phoenix.PubSub.subscribe(Towerops.PubSub, reply_topic)

    sensors = get_sensors_for_live_mode(assigns)
    sensor_oids = Enum.map(sensors, & &1.sensor_oid)

    _ =
      Phoenix.PubSub.broadcast(
        Towerops.PubSub,
        "agent:#{agent_token_id}:live_poll",
        {:live_poll_requested, device_id, sensor_oids, reply_topic}
      )

    receive do
      {:live_poll_result, oid_values} ->
        process_live_poll_oids(sensors, oid_values)
    after
      3_000 ->
        Logger.warning("Live poll timeout waiting for agent response")
        []
    end
  end

  @doc """
  Get sensors for the current live mode graph view.
  """
  def get_sensors_for_live_mode(%{sensor_id: sensor_id}) when not is_nil(sensor_id) do
    case Snmp.get_sensor(sensor_id) do
      nil -> []
      sensor -> [sensor]
    end
  end

  def get_sensors_for_live_mode(%{sensor_type: sensor_type, device_id: device_id}) do
    sensor_types = get_sensor_types_for_live_mode(sensor_type)
    device = Snmp.get_device_with_associations(device_id)

    if is_nil(sensor_types) || is_nil(device) do
      []
    else
      device.sensors
      |> Enum.filter(&(&1.sensor_type in sensor_types))
      |> Enum.sort_by(& &1.sensor_index)
    end
  end

  @doc """
  Update the rolling live data buffer with new points.
  """
  def update_buffer(buffer, current_count, new_points, max_points \\ @max_live_points)

  def update_buffer(buffer, current_count, new_points, max_points) do
    updated_buffer =
      Enum.reduce(new_points, buffer, fn point, acc ->
        sensor_buffer = Map.get(acc, point.sensor_id, [])

        new_buffer = [
          %{x: point.timestamp, y: point.value} | sensor_buffer
        ]

        trimmed_buffer =
          if current_count >= max_points do
            Enum.take(new_buffer, max_points)
          else
            new_buffer
          end

        Map.put(acc, point.sensor_id, trimmed_buffer)
      end)

    new_count = min(current_count + 1, max_points)
    {updated_buffer, new_count}
  end

  # ── Private helpers ──

  defp poll_sensor_to_data_point(sensor, client_opts, timestamp_ms) do
    case poll_single_sensor(sensor, client_opts) do
      {:ok, value} ->
        %{
          sensor_id: sensor.id,
          label: sensor.sensor_descr,
          value: Float.round(value, 1),
          timestamp: timestamp_ms
        }

      {:error, _reason} ->
        nil
    end
  end

  defp poll_single_sensor(sensor, client_opts) do
    case Snmp.Client.get(client_opts, sensor.sensor_oid) do
      {:ok, raw_value} when is_number(raw_value) ->
        value = raw_value / sensor.sensor_divisor
        {:ok, value}

      {:ok, _non_numeric} ->
        {:error, :non_numeric}

      {:error, reason} ->
        {:error, reason}
    end
  end

  defp poll_interface_stats_parallel(interfaces, client_opts) do
    interfaces
    |> Task.async_stream(&poll_single_interface_octets(&1, client_opts),
      max_concurrency: 10,
      timeout: 4_000,
      on_timeout: :kill_task
    )
    |> Enum.reduce([], fn
      {:ok, nil}, acc -> acc
      {:ok, stat}, acc -> [stat | acc]
      {:exit, _}, acc -> acc
    end)
  end

  defp poll_single_interface_octets(interface, client_opts) do
    hc_opts =
      case Keyword.get(client_opts, :version) do
        v when v in ["1", :v1] -> Keyword.put(client_opts, :version, "2c")
        _ -> client_opts
      end

    in_octets_oid = "1.3.6.1.2.1.31.1.1.1.6.#{interface.if_index}"
    out_octets_oid = "1.3.6.1.2.1.31.1.1.1.10.#{interface.if_index}"

    with {:ok, in_octets} <- Snmp.Client.get(hc_opts, in_octets_oid),
         {:ok, out_octets} <- Snmp.Client.get(hc_opts, out_octets_oid),
         true <- is_number(in_octets) and is_number(out_octets) do
      build_interface_stat(interface, in_octets, out_octets)
    else
      _ -> poll_single_interface_octets_32bit(interface, client_opts)
    end
  end

  defp poll_single_interface_octets_32bit(interface, client_opts) do
    in_octets_oid = "1.3.6.1.2.1.2.2.1.10.#{interface.if_index}"
    out_octets_oid = "1.3.6.1.2.1.2.2.1.16.#{interface.if_index}"

    with {:ok, in_octets} <- Snmp.Client.get(client_opts, in_octets_oid),
         {:ok, out_octets} <- Snmp.Client.get(client_opts, out_octets_oid),
         true <- is_number(in_octets) and is_number(out_octets) do
      build_interface_stat(interface, in_octets, out_octets)
    else
      _ -> nil
    end
  end

  defp build_interface_stat(interface, in_octets, out_octets) do
    %{
      interface_id: interface.id,
      interface_name: interface.if_name || interface.if_descr,
      if_in_octets: in_octets,
      if_out_octets: out_octets,
      polled_at: DateTime.utc_now()
    }
  end

  defp calculate_traffic_bps_from_stats(previous_stats, current_stats, timestamp_ms) do
    previous_map = Map.new(previous_stats, &{&1.interface_id, &1})

    bps_per_interface =
      Enum.flat_map(current_stats, fn current ->
        calculate_interface_bps_if_previous_exists(current, previous_map)
      end)

    total_in = Enum.reduce(bps_per_interface, 0.0, fn x, acc -> acc + x.in_bps end)
    total_out = Enum.reduce(bps_per_interface, 0.0, fn x, acc -> acc + x.out_bps end)

    [
      %{
        sensor_id: "traffic_in",
        label: "Inbound",
        value: Float.round(total_in, 2),
        timestamp: timestamp_ms
      },
      %{
        sensor_id: "traffic_out",
        label: "Outbound",
        value: Float.round(total_out, 2),
        timestamp: timestamp_ms
      }
    ]
  end

  defp calculate_interface_bps_if_previous_exists(current, previous_map) do
    case Map.get(previous_map, current.interface_id) do
      nil ->
        []

      previous ->
        time_diff = current.polled_at |> DateTime.diff(previous.polled_at, :second) |> max(1)
        in_bps = calculate_live_bps(current.if_in_octets, previous.if_in_octets, time_diff)
        out_bps = calculate_live_bps(current.if_out_octets, previous.if_out_octets, time_diff)
        [%{interface_id: current.interface_id, in_bps: in_bps, out_bps: out_bps}]
    end
  end

  defp calculate_live_bps(current_octets, previous_octets, time_diff) do
    if current_octets && previous_octets do
      Towerops.Capacity.calculate_bps(previous_octets, current_octets, time_diff)
    else
      0.0
    end
  end

  defp process_live_poll_oids(sensors, oid_values) do
    timestamp_ms = DateTime.to_unix(DateTime.utc_now(), :millisecond)

    sensors
    |> Enum.map(&build_live_poll_data_point(&1, oid_values, timestamp_ms))
    |> Enum.reject(&is_nil/1)
  end

  defp build_live_poll_data_point(sensor, oid_values, timestamp_ms) do
    with value when not is_nil(value) <- Map.get(oid_values, sensor.sensor_oid),
         parsed_value when not is_nil(parsed_value) <- parse_sensor_value(value) do
      final_value = parsed_value / sensor.sensor_divisor

      %{
        sensor_id: sensor.id,
        label: sensor.sensor_descr,
        value: Float.round(final_value, 1),
        timestamp: timestamp_ms
      }
    else
      _ -> nil
    end
  end

  defp parse_sensor_value(value) do
    case Towerops.Numeric.parse_float(value) do
      {:ok, f} -> f
      :error -> nil
    end
  end

  defp get_sensor_types_for_live_mode("processors"), do: ["cpu_load"]
  defp get_sensor_types_for_live_mode("memory"), do: ["memory_usage"]
  defp get_sensor_types_for_live_mode("storage"), do: ["disk_usage"]
  defp get_sensor_types_for_live_mode("temperature"), do: ["temperature", "cpu_temperature", "celsius"]
  defp get_sensor_types_for_live_mode("voltage"), do: ["voltage", "volts"]
  defp get_sensor_types_for_live_mode("traffic"), do: nil
  defp get_sensor_types_for_live_mode(sensor_type), do: [sensor_type]
end