if Code.ensure_loaded?(Oban) do
  defmodule PromEx.Plugins.Oban do
    @moduledoc """
    This plugin captures metrics emitted by Oban. Specifically, it captures metrics from job events, producer events,
    and also from internal polling jobs to monitor queue sizes

    This plugin supports the following options:
    - `oban_supervisors`: This is an OPTIONAL option and it allows you to specify what Oban instances should have their events
      tracked. By default the only Oban instance that will have its events tracked is the default `Oban` instance. As a result, by
      default this option has a value of `[Oban]`. If you would like to track other named Oban instances, or perhaps your default
      and only Oban instance has a different name, you can pass in your own list of Oban instances (e.g. `[Oban, Oban.PrivateJobs]`).

    - `metric_prefix`: This option is OPTIONAL and is used to override the default metric prefix of
      `[otp_app, :prom_ex, :oban]`. If this changes you will also want to set `oban_metric_prefix`
      in your `dashboard_assigns` to the snakecase version of your prefix, the default
      `oban_metric_prefix` is `{otp_app}_prom_ex_oban`.

    - `poll_rate`: This option is OPTIONAL and is the rate at which poll metrics are refreshed (default is 5 seconds).

    - `duration_unit`: This is an OPTIONAL option and is a `Telemetry.Metrics.time_unit()`. It can be one of:
      `:second | :millisecond | :microsecond | :nanosecond`. It is `:millisecond` by default.

    This plugin exposes the following metric groups:
    - `:oban_init_event_metrics`
    - `:oban_job_event_metrics`
    - `:oban_producer_event_metrics`
    - `:oban_circuit_event_metrics`
    - `:oban_queue_poll_metrics`

    To use plugin in your application, add the following to your PromEx module:
    ```
    defmodule WebApp.PromEx do
      use PromEx, otp_app: :web_app

      @impl true
      def plugins do
        [
          ...
          {PromEx.Plugins.Oban, oban_supervisors: [Oban, Oban.AnotherSupervisor], poll_rate: 10_000}
        ]
      end

      @impl true
      def dashboards do
        [
          ...
          {:prom_ex, "oban.json"}
        ]
      end
    end
    ```
    """

    use PromEx.Plugin

    alias PromEx.Utils

    import Ecto.Query, only: [group_by: 3, select: 3]

    # Oban events
    @init_event [:oban, :supervisor, :init]
    @job_complete_event [:oban, :job, :stop]
    @job_exception_event [:oban, :job, :exception]
    @producer_complete_event [:oban, :producer, :stop]
    @producer_exception_event [:oban, :producer, :exception]
    @circuit_breaker_trip_event [:oban, :circuit, :trip]
    @circuit_breaker_open_event [:oban, :circuit, :open]

    # PromEx Oban proxy events
    @init_event_queue_limit_proxy [:prom_ex, :oban, :queue, :limit, :proxy]

    @impl true
    def event_metrics(opts) do
      otp_app = Keyword.fetch!(opts, :otp_app)
      metric_prefix = Keyword.get(opts, :metric_prefix, PromEx.metric_prefix(otp_app, :oban))
      duration_unit = Keyword.get(opts, :duration_unit, :millisecond)

      oban_supervisors = get_oban_supervisors(opts)
      keep_function_filter = keep_oban_instance_metrics(oban_supervisors)

      # Set up event proxies
      set_up_init_proxy_event(metric_prefix)

      [
        oban_supervisor_init_event_metrics(metric_prefix, keep_function_filter, duration_unit),
        oban_job_event_metrics(metric_prefix, keep_function_filter, duration_unit),
        oban_producer_event_metrics(metric_prefix, keep_function_filter, duration_unit),
        oban_circuit_breaker_event_metrics(metric_prefix, keep_function_filter)
      ]
    end

    @impl true
    def polling_metrics(opts) do
      otp_app = Keyword.fetch!(opts, :otp_app)
      metric_prefix = Keyword.get(opts, :metric_prefix, PromEx.metric_prefix(otp_app, :oban))
      poll_rate = Keyword.get(opts, :poll_rate, 5_000)

      oban_supervisors = get_oban_supervisors(opts)

      polling_opts = Keyword.get(opts, :opts, [])

      # Queue length details
      Polling.build(
        :oban_queue_poll_metrics,
        poll_rate,
        {__MODULE__, :execute_queue_metrics, [oban_supervisors]},
        [
          last_value(
            metric_prefix ++ [:queue, :length, :count],
            event_name: [:prom_ex, :plugin, :oban, :queue, :length, :count],
            description: "The total number jobs that are in the queue in the designated state",
            measurement: :count,
            tags: [:name, :queue, :state]
          )
        ],
        polling_opts
      )
    end

    @doc false
    def execute_queue_metrics(oban_supervisors) do
      oban_supervisors
      |> Enum.each(fn oban_supervisor ->
        oban_supervisor
        |> Oban.Registry.whereis()
        |> case do
          oban_pid when is_pid(oban_pid) ->
            config = Oban.Registry.config(oban_supervisor)
            handle_oban_queue_polling_metrics(oban_supervisor, config)

          _ ->
            :skip
        end
      end)
    end

    def oban_circuit_breaker_event_metrics(metric_prefix, keep_function_filter) do
      Event.build(
        :oban_circuit_breaker_event_metrics,
        [
          counter(
            metric_prefix ++ [:circuit, :trip, :total],
            event_name: @circuit_breaker_trip_event,
            description: "The number of circuit breaker events that have occurred",
            tag_values: &circuit_breaker_trip_tag_values/1,
            tags: [:name, :circuit_breaker],
            keep: keep_function_filter
          ),
          counter(
            metric_prefix ++ [:circuit, :open, :total],
            event_name: @circuit_breaker_open_event,
            description: "The number of circuit open events that have occurred.",
            tag_values: &circuit_breaker_trip_tag_values/1,
            tags: [:name, :circuit_breaker],
            keep: keep_function_filter
          )
        ]
      )
    end

    @doc false
    def handle_proxy_init_event(_event_name, _event_measurement, event_metadata, _config) do
      Enum.each(event_metadata.conf.queues, fn {queue, queue_opts} ->
        limit = Keyword.get(queue_opts, :limit, 0)

        metadata = %{
          queue: queue,
          name: event_metadata.conf.name
        }

        :telemetry.execute(@init_event_queue_limit_proxy, %{limit: limit}, metadata)
      end)
    end

    defp circuit_breaker_trip_tag_values(%{name: name, config: conf}) do
      %{
        name: normalize_module_name(conf.name),
        circuit_breaker: normalize_module_name(name)
      }
    end

    defp circuit_breaker_trip_tag_values(%{name: name, conf: conf}) do
      %{
        name: normalize_module_name(conf.name),
        circuit_breaker: normalize_module_name(name)
      }
    end

    defp oban_job_event_metrics(metric_prefix, keep_function_filter, duration_unit) do
      job_attempt_buckets = [1, 5, 10]
      job_duration_buckets = [10, 100, 500, 1_000, 5_000, 20_000]
      duration_unit_plural = Utils.make_plural_atom(duration_unit)

      Event.build(
        :oban_job_event_metrics,
        [
          distribution(
            metric_prefix ++ [:job, :processing, :duration, duration_unit_plural],
            event_name: @job_complete_event,
            measurement: :duration,
            description: "The amount of time it takes to processes an Oban job.",
            reporter_options: [
              buckets: job_duration_buckets
            ],
            tag_values: &job_complete_tag_values/1,
            tags: [:name, :queue, :state, :worker],
            unit: {:native, duration_unit},
            keep: keep_function_filter
          ),
          distribution(
            metric_prefix ++ [:job, :queue, :time, duration_unit_plural],
            event_name: @job_complete_event,
            measurement: :queue_time,
            description: "The amount of time that the Oban job was waiting in queue for processing.",
            reporter_options: [
              buckets: job_duration_buckets
            ],
            tag_values: &job_complete_tag_values/1,
            tags: [:name, :queue, :state, :worker],
            unit: {:native, duration_unit},
            keep: keep_function_filter
          ),
          distribution(
            metric_prefix ++ [:job, :complete, :attempts],
            event_name: @job_complete_event,
            measurement: fn _measurement, %{attempt: attempt} ->
              attempt
            end,
            description: "The number of times a job was attempted prior to completing.",
            reporter_options: [
              buckets: job_attempt_buckets
            ],
            tag_values: &job_complete_tag_values/1,
            tags: [:name, :queue, :state, :worker],
            keep: keep_function_filter
          ),
          distribution(
            metric_prefix ++ [:job, :exception, :duration, duration_unit_plural],
            event_name: @job_exception_event,
            measurement: :duration,
            description: "The amount of time it took to process a job the encountered an exception.",
            reporter_options: [
              buckets: job_duration_buckets
            ],
            tag_values: &job_exception_tag_values/1,
            tags: [:name, :queue, :state, :worker, :kind, :error],
            unit: {:native, duration_unit},
            keep: keep_function_filter
          ),
          distribution(
            metric_prefix ++ [:job, :exception, :queue, :time, duration_unit_plural],
            event_name: @job_exception_event,
            measurement: :queue_time,
            description:
              "The amount of time that the Oban job was waiting in queue for processing prior to resulting in an exception.",
            reporter_options: [
              buckets: job_duration_buckets
            ],
            tag_values: &job_exception_tag_values/1,
            tags: [:name, :queue, :state, :worker, :kind, :error],
            unit: {:native, duration_unit},
            keep: keep_function_filter
          ),
          distribution(
            metric_prefix ++ [:job, :exception, :attempts],
            event_name: @job_exception_event,
            measurement: fn _measurement, %{attempt: attempt} ->
              attempt
            end,
            description: "The number of times a job was attempted prior to throwing an exception.",
            reporter_options: [
              buckets: job_attempt_buckets
            ],
            tag_values: &job_exception_tag_values/1,
            tags: [:name, :queue, :state, :worker],
            keep: keep_function_filter
          )
        ]
      )
    end

    defp oban_producer_event_metrics(metric_prefix, keep_function_filter, duration_unit) do
      duration_unit_plural = Utils.make_plural_atom(duration_unit)

      Event.build(
        :oban_producer_event_metrics,
        [
          distribution(
            metric_prefix ++ [:producer, :duration, duration_unit_plural],
            event_name: @producer_complete_event,
            measurement: :duration,
            description: "How long it took to dispatch the job.",
            reporter_options: [
              buckets: [10, 100, 500, 1_000, 5_000, 10_000]
            ],
            unit: {:native, duration_unit},
            tag_values: &producer_tag_values/1,
            tags: [:queue, :name],
            keep: keep_function_filter
          ),
          distribution(
            metric_prefix ++ [:producer, :dispatched, :count],
            event_name: @producer_complete_event,
            measurement: fn _measurement, %{dispatched_count: count} ->
              count
            end,
            description: "The number of jobs that were dispatched.",
            reporter_options: [
              buckets: [5, 10, 50, 100]
            ],
            tag_values: &producer_tag_values/1,
            tags: [:queue, :name],
            keep: keep_function_filter
          ),
          distribution(
            metric_prefix ++ [:producer, :exception, :duration, duration_unit_plural],
            event_name: @producer_exception_event,
            measurement: :duration,
            description: "How long it took for the producer to raise an exception.",
            reporter_options: [
              buckets: [10, 100, 500, 1_000, 5_000, 10_000]
            ],
            unit: {:native, duration_unit},
            tag_values: &producer_tag_values/1,
            tags: [:queue, :name],
            keep: keep_function_filter
          )
        ]
      )
    end

    defp job_complete_tag_values(metadata) do
      config =
        case metadata do
          %{config: config} ->
            config

          %{conf: config} ->
            config
        end

      %{
        name: normalize_module_name(config.name),
        queue: metadata.job.queue,
        state: metadata.state,
        worker: metadata.worker
      }
    end

    defp job_exception_tag_values(metadata) do
      error =
        case metadata.error do
          %error_type{} -> normalize_module_name(error_type)
          _ -> "Undefined"
        end

      config =
        case metadata do
          %{config: config} ->
            config

          %{conf: config} ->
            config
        end

      %{
        name: normalize_module_name(config.name),
        queue: metadata.job.queue,
        state: metadata.state,
        worker: metadata.worker,
        kind: metadata.kind,
        error: error
      }
    end

    defp producer_tag_values(metadata) do
      %{
        queue: metadata.queue,
        name: normalize_module_name(metadata.conf.name)
      }
    end

    defp oban_supervisor_init_event_metrics(metric_prefix, keep_function_filter, duration_unit) do
      duration_unit_plural = Utils.make_plural_atom(duration_unit)

      Event.build(
        :oban_init_event_metrics,
        [
          last_value(
            metric_prefix ++ [:init, :status, :info],
            event_name: @init_event,
            description: "Information regarding the initialized oban supervisor.",
            measurement: fn _measurements -> 1 end,
            tags: [:name, :node, :plugins, :prefix, :queues, :repo],
            tag_values: &oban_init_tag_values/1,
            keep: keep_function_filter
          ),
          last_value(
            metric_prefix ++ [:init, :shutdown, :grace, :period, duration_unit_plural],
            event_name: @init_event,
            description: "The Oban supervisor's shutdown grace period value.",
            measurement: fn _measurements, %{conf: config} ->
              config.shutdown_grace_period
            end,
            tags: [:name],
            tag_values: &oban_init_tag_values/1,
            keep: keep_function_filter
          ),
          last_value(
            metric_prefix ++ [:init, :dispatch, :cooldown, duration_unit_plural],
            event_name: @init_event,
            description: "The Oban supervisor's dispatch cooldown value.",
            measurement: fn _measurements, %{conf: config} ->
              config.dispatch_cooldown
            end,
            tags: [:name],
            tag_values: &oban_init_tag_values/1,
            keep: keep_function_filter
          ),
          last_value(
            metric_prefix ++ [:init, :queue, :concurrency, :limit],
            event_name: @init_event_queue_limit_proxy,
            description: "The concurrency limits of each of the Oban queue.",
            measurement: :limit,
            tags: [:name, :queue],
            tag_values: &oban_init_queues_tag_values/1,
            keep: keep_function_filter
          )
        ]
      )
    end

    defp handle_oban_queue_polling_metrics(oban_supervisor, config) do
      query =
        Oban.Job
        |> group_by([j], [j.queue, j.state])
        |> select([j], {j.queue, j.state, count(j.id)})

      config
      |> Oban.Repo.all(query)
      |> include_zeros_for_missing_queue_states()
      |> Enum.each(fn {{queue, state}, count} ->
        measurements = %{count: count}
        metadata = %{name: normalize_module_name(oban_supervisor), queue: queue, state: state}

        :telemetry.execute([:prom_ex, :plugin, :oban, :queue, :length, :count], measurements, metadata)
      end)
    end

    defp include_zeros_for_missing_queue_states(query_result) do
      {_, opts} =
        Oban.config().plugins
        |> Enum.find({nil, [queues: Oban.config().queues]}, fn {plugin, _} ->
          plugin == Oban.Pro.Plugins.DynamicQueues
        end)

      all_queues =
        opts
        |> Keyword.get(:queues, [])
        |> Keyword.keys()

      all_states = Oban.Job.states()

      zeros = for queue <- all_queues, state <- all_states, into: %{}, do: {{to_string(queue), to_string(state)}, 0}
      counts = for {queue, state, count} <- query_result, into: %{}, do: {{queue, state}, count}

      Map.merge(zeros, counts)
    end

    defp get_oban_supervisors(opts) do
      opts
      |> Keyword.get(:oban_supervisors, [Oban])
      |> case do
        supervisors when is_list(supervisors) ->
          MapSet.new(supervisors)

        _ ->
          raise "Invalid :oban_supervisors option value."
      end
    end

    defp keep_oban_instance_metrics(oban_supervisors) do
      fn
        %{conf: %{name: name}} ->
          MapSet.member?(oban_supervisors, name)

        %{name: name} ->
          MapSet.member?(oban_supervisors, name)

        _ ->
          false
      end
    end

    defp oban_init_tag_values(%{conf: config}) do
      plugins_string_list =
        config.plugins
        |> Enum.map_join(", ", fn plugin ->
          normalize_module_name(plugin)
        end)

      queues_string_list =
        config.queues
        |> Enum.map_join(", ", fn {queue, _queue_opts} ->
          Atom.to_string(queue)
        end)

      %{
        name: normalize_module_name(config.name),
        node: config.node,
        plugins: plugins_string_list,
        prefix: config.prefix,
        queues: queues_string_list,
        repo: normalize_module_name(config.repo)
      }
    end

    defp oban_init_queues_tag_values(%{name: name, queue: queue}) do
      %{
        name: normalize_module_name(name),
        queue: queue
      }
    end

    defp set_up_init_proxy_event(prefix) do
      :telemetry.attach(
        [:prom_ex, :oban, :proxy] ++ prefix,
        @init_event,
        &__MODULE__.handle_proxy_init_event/4,
        %{}
      )
    end

    defp normalize_module_name(name) when is_atom(name) do
      name
      |> Atom.to_string()
      |> String.trim_leading("Elixir.")
    end

    defp normalize_module_name({name, _options}), do: normalize_module_name(name)

    defp normalize_module_name(name), do: name
  end
else
  defmodule PromEx.Plugins.Oban do
    @moduledoc false
    use PromEx.Plugin

    @impl true
    def event_metrics(_opts) do
      PromEx.Plugin.no_dep_raise(__MODULE__, "Oban")
    end
  end
end
