towerops/lib/towerops/workers/check_executor_worker.ex

defmodule Towerops.Workers.CheckExecutorWorker do
  @moduledoc """
  Unified worker for executing all check types.

  Dispatches to appropriate executor based on check_type:
  - snmp_sensor → SnmpSensorExecutor
  - snmp_interface → SnmpInterfaceExecutor
  - snmp_processor → SnmpProcessorExecutor
  - snmp_storage → SnmpStorageExecutor
  - http → HttpExecutor
  - tcp → TcpExecutor
  - dns → DnsExecutor
  - ping → PingExecutor (server-side fallback, skipped when agent assigned)

  Records results in check_results TimescaleDB hypertable and updates
  check state (OK/WARNING/CRITICAL/UNKNOWN).

  Self-schedules next execution based on check interval with distributed
  polling offsets to prevent thundering herd.
  """

  # `unique` keeps at most one pending job per check_id. Every insert path
  # (self-scheduling, discovery re-runs, JobHealthCheckWorker, the check form,
  # API and GraphQL) goes through this worker, so without it duplicate jobs
  # accumulate without bound. `:executing` is deliberately excluded: perform/1
  # self-schedules the next run while the current job is still executing, and
  # including it would make that reschedule conflict with itself.
  use Oban.Pro.Worker,
    queue: :check_executors,
    max_attempts: 3,
    unique: [
      keys: [:check_id],
      states: [:available, :scheduled, :executing, :retryable, :suspended],
      period: :infinity
    ]

  alias Towerops.Agents
  alias Towerops.Monitoring
  alias Towerops.Monitoring.Executors.DnsExecutor
  alias Towerops.Monitoring.Executors.HttpExecutor
  alias Towerops.Monitoring.Executors.PingExecutor
  alias Towerops.Monitoring.Executors.SnmpInterfaceExecutor
  alias Towerops.Monitoring.Executors.SnmpProcessorExecutor
  alias Towerops.Monitoring.Executors.SnmpSensorExecutor
  alias Towerops.Monitoring.Executors.SnmpStorageExecutor
  alias Towerops.Monitoring.Executors.SslExecutor
  alias Towerops.Monitoring.Executors.TcpExecutor
  alias Towerops.Snmp.Client
  alias Towerops.Workers.PollingOffset

  require Logger

  @impl Oban.Pro.Worker
  def process(%Oban.Job{args: %{"check_id" => check_id}}) do
    case Monitoring.get_check(check_id) do
      nil ->
        Logger.debug("Check #{check_id} deleted, skipping execution")
        :ok

      check ->
        _ = execute_or_skip(check)
        :ok
    end
  end

  defp execute_or_skip(%{enabled: false} = check) do
    Logger.debug("Check #{check.id} disabled, skipping execution")
  end

  defp execute_or_skip(check) do
    if should_skip_check?(check) do
      Logger.debug("Skipping check #{check.id} (#{check.check_type}) - device has agent or Phoenix SNMP disabled")
    else
      execute_and_record(check)
    end

    schedule_next_check(check)
  end

  # Unified skip logic for ALL check types.
  # Skip when the device has any effective agent assigned at any cascade level,
  # or when Phoenix SNMP is disabled for SNMP-type checks.
  defp should_skip_check?(check) do
    snmp_disabled_for_snmp_check?(check) or device_has_agent?(check)
  end

  defp snmp_disabled_for_snmp_check?(check) do
    check.check_type in ~w(snmp_sensor snmp_interface snmp_processor snmp_storage) and
      Client.phoenix_snmp_disabled()
  end

  defp device_has_agent?(check) do
    check.device_id != nil and Agents.device_has_effective_agent?(check.device_id)
  end

  defp execute_and_record(check) do
    Logger.debug("Executing check #{check.id} (#{check.check_type})")

    check
    |> dispatch_executor()
    |> record_result(check)
  end

  defp dispatch_executor(%{check_type: "snmp_sensor"} = check), do: SnmpSensorExecutor.execute(check)
  defp dispatch_executor(%{check_type: "snmp_interface"} = check), do: SnmpInterfaceExecutor.execute(check)
  defp dispatch_executor(%{check_type: "snmp_processor"} = check), do: SnmpProcessorExecutor.execute(check)
  defp dispatch_executor(%{check_type: "snmp_storage"} = check), do: SnmpStorageExecutor.execute(check)
  defp dispatch_executor(%{check_type: "http"} = check), do: execute_http_check(check)
  defp dispatch_executor(%{check_type: "tcp"} = check), do: execute_tcp_check(check)
  defp dispatch_executor(%{check_type: "dns"} = check), do: execute_dns_check(check)
  defp dispatch_executor(%{check_type: "ssl"} = check), do: execute_ssl_check(check)
  defp dispatch_executor(%{check_type: "ping"} = check), do: execute_ping_check(check)
  defp dispatch_executor(%{check_type: type}), do: {:error, "Unknown check type: #{type}"}

  defp record_result({:ok, %{value: value, status: status, output: output, response_time_ms: time}}, check) do
    Monitoring.create_check_result(%{
      check_id: check.id,
      organization_id: check.organization_id,
      value: value,
      status: status,
      output: output,
      response_time_ms: time,
      checked_at: DateTime.utc_now(),
      agent_token_id: check.agent_token_id
    })

    Monitoring.update_check_state(check, status, output)
    Logger.debug("Check #{check.id} completed: status=#{status}, value=#{value}, output=#{output}")
  end

  defp record_result({:error, reason}, check) do
    Logger.warning("Check #{check.id} failed: #{inspect(reason)}")

    Monitoring.create_check_result(%{
      check_id: check.id,
      organization_id: check.organization_id,
      status: 3,
      output: "Error: #{inspect(reason)}",
      checked_at: DateTime.utc_now(),
      agent_token_id: check.agent_token_id
    })

    Monitoring.update_check_state(check, 3, "Error: #{inspect(reason)}")
  end

  # Adapter for HTTP executor which returns different format
  defp execute_http_check(check) do
    case HttpExecutor.execute(check.config, check.timeout_ms) do
      {:ok, response_time, output} ->
        {:ok,
         %{
           value: response_time,
           status: 0,
           output: output,
           response_time_ms: response_time
         }}

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

  # Adapter for TCP executor which returns different format
  defp execute_tcp_check(check) do
    case TcpExecutor.execute(check.config, check.timeout_ms) do
      {:ok, response_time, output} ->
        {:ok,
         %{
           value: response_time,
           status: 0,
           output: output,
           response_time_ms: response_time
         }}

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

  # Adapter for DNS executor which returns different format
  defp execute_dns_check(check) do
    case DnsExecutor.execute(check.config, check.timeout_ms) do
      {:ok, response_time, output} ->
        {:ok,
         %{
           value: response_time,
           status: 0,
           output: output,
           response_time_ms: response_time
         }}

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

  # Adapter for SSL executor - returns status directly (OK/WARNING/CRITICAL)
  defp execute_ssl_check(check) do
    case SslExecutor.execute(check.config, check.timeout_ms) do
      {:ok, status, response_time, output} ->
        {:ok,
         %{
           value: response_time,
           status: status,
           output: output,
           response_time_ms: response_time
         }}

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

  # Adapter for Ping executor which returns different format
  defp execute_ping_check(check) do
    case PingExecutor.execute(check.config, check.timeout_ms) do
      {:ok, response_time, output} ->
        {:ok,
         %{
           value: response_time,
           status: 0,
           output: output,
           response_time_ms: response_time
         }}

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

  defp schedule_next_check(check) do
    # Calculate staggered offset based on check ID
    offset = PollingOffset.calculate_offset(check.id, check.interval_seconds)

    %{check_id: check.id}
    |> new(schedule_in: offset)
    |> Oban.insert()
  end
end