defmodule Oban.Pro.Stages.Hooks do
  @moduledoc false

  @behaviour Oban.Pro.Stage

  alias Oban.Pro.Utils

  require Logger

  defstruct modules: []

  @impl Oban.Pro.Stage
  def init(worker, modules) do
    {:ok, struct!(__MODULE__, modules: [worker | modules])}
  end

  @impl Oban.Pro.Stage
  def before_new(args, opts, %{modules: modules}) do
    (modules ++ get_hooks())
    |> Enum.filter(&hook_exported?(&1, :before_new, 2))
    |> Enum.reduce_while({:ok, args, opts}, fn module, {:ok, args, opts} ->
      case module.before_new(args, opts) do
        {:ok, _, _} = ok -> {:cont, ok}
        other -> {:halt, other}
      end
    end)
  end

  @impl Oban.Pro.Stage
  def before_process(job, %{modules: modules}) do
    (modules ++ get_hooks())
    |> Enum.filter(&hook_exported?(&1, :before_process, 1))
    |> Enum.reduce_while({:ok, job}, fn module, {:ok, job} ->
      case module.before_process(job) do
        {:ok, _} = ok -> {:cont, ok}
        other -> {:halt, other}
      end
    end)
  end

  def attach_hook(module) do
    with :ok <- Utils.validate_opts([module], &validate_module/1) do
      modules =
        get_hooks()
        |> Enum.concat([module])
        |> Enum.uniq()

      :persistent_term.put(:oban_pro_hooks, modules)
    end
  end

  def detach_hook(module) do
    case get_hooks() do
      [_ | _] = modules ->
        :persistent_term.put(:oban_pro_hooks, modules -- [module])

      _ ->
        :ok
    end
  end

  def handle_event(_event, _timing, meta, _conf) do
    %{job: %{unsaved_error: unsaved_error}, state: state} = meta

    with {worker, job, opts} <- resolve_processing(meta.job) do
      hook_state = exec_to_hook_state(state)
      module_hooks = Keyword.get(opts, :hooks, [])
      global_hooks = get_hooks()

      job = %{job | unsaved_error: unsaved_error}
      result = Map.get(meta, :result, nil)

      ([worker | module_hooks] ++ global_hooks)
      |> Enum.uniq()
      |> Enum.each(fn module ->
        cond do
          hook_exported?(module, :after_process, 3) ->
            module.after_process(hook_state, job, result)

          hook_exported?(module, :after_process, 2) ->
            module.after_process(hook_state, job)

          true ->
            :ok
        end
      end)
    end
  catch
    kind, value ->
      Logger.error(fn ->
        "[Oban.Pro.Worker] hook error: " <> Exception.format(kind, value, __STACKTRACE__)
      end)
  after
    Process.delete(:oban_processing)
  end

  defp resolve_processing(job) do
    # The event may be triggered by a producer after the process was killed or experienced a
    # linked crash. In that event, the processing data isn't in the dictionary and we need to
    # resolve and apply before hooks to the job again.
    with nil <- Process.get(:oban_processing),
         {:ok, worker} <- Oban.Worker.from_string(job.worker),
         opts = worker.__opts__(),
         {:ok, job} <- Oban.Pro.Worker.before_process(worker, job, opts) do
      processing = {worker, job, opts}
      Process.put(:oban_processing, processing)
      processing
    end
  end

  defp get_hooks, do: :persistent_term.get(:oban_pro_hooks, [])

  defp exec_to_hook_state(:cancelled), do: :cancel
  defp exec_to_hook_state(:success), do: :complete
  defp exec_to_hook_state(:discard), do: :discard
  defp exec_to_hook_state(:failure), do: :error
  defp exec_to_hook_state(:snoozed), do: :snooze

  defp hook_exported?(mod, fun, arity) do
    Code.ensure_loaded?(mod) and function_exported?(mod, fun, arity)
  end

  # Validation

  defp validate_module(module) do
    cond do
      not Code.ensure_loaded?(module) ->
        {:error, "unable to load callback module #{inspect(module)}"}

      function_exported?(module, :before_new, 2) ->
        :ok

      function_exported?(module, :after_process, 2) ->
        :ok

      function_exported?(module, :after_process, 3) ->
        :ok

      function_exported?(module, :before_process, 1) ->
        :ok

      true ->
        {:error,
         "#{inspect(module)} doesn't implement after_process/3, before_new/2, before_process/1"}
    end
  end
end
