defmodule Oban.Pro.Workers.Chain do
  @moduledoc """
  A dedicated worker for linking jobs to run in strict sequential order.

  This worker is deprecated in favor of `Oban.Pro.Worker`, which is composable and allows better
  parallelism.
  """

  @moduledoc deprecated: "Use Oban.Pro.Worker instead"

  import Ecto.Query

  alias Oban.{Job, Repo, Validation}
  alias Oban.Pro.Utils

  @default_meta %{
    on_discarded: "ignore",
    on_cancelled: "ignore",
    chain_key: nil,
    hold_snooze: 60,
    wait_retry: 10,
    wait_sleep: 1000,
    wait_snooze: 5
  }

  @opts_keys ~w(by hold_snooze on_cancelled on_discarded wait_retry wait_sleep wait_snooze)a

  @doc false
  defmacro __using__(opts) do
    {chain_opts, other_opts} = Keyword.split(opts, @opts_keys)

    quote do
      Validation.validate!(unquote(chain_opts), &Oban.Pro.Workers.Chain.validate/1)

      use Oban.Pro.Worker, unquote(other_opts)

      @base_meta unquote(chain_opts)
                 |> Keyword.delete(:by)
                 |> Map.new()

      @impl Oban.Worker
      def new(args, opts) when is_map(args) and is_list(opts) do
        meta =
          opts
          |> Keyword.get(:meta, %{})
          |> Map.merge(@base_meta)
          |> Map.put_new(:chain_by, Keyword.fetch!(unquote(chain_opts), :by))
          |> Map.update!(:chain_by, &Utils.normalize_by/1)

        chain_key = Oban.Pro.Workers.Chain.to_chain_key(__MODULE__, args, meta)
        chain_id = Oban.Pro.Stages.Chain.to_chain_id(meta.chain_by, __MODULE__, args, opts)

        # Include the `chain_id` for compatibility with chains in v1.5
        meta =
          meta
          |> Map.put(:chain_key, chain_key)
          |> Map.put(:chain_id, chain_id)

        super(args, Keyword.put(opts, :meta, meta))
      end

      @impl Oban.Worker
      def perform(%Job{} = job) do
        opts = __opts__()

        with {:ok, job} <- Oban.Pro.Worker.before_process(__MODULE__, job, opts) do
          job
          |> Oban.Pro.Workers.Chain.maybe_process(__MODULE__)
          |> Oban.Pro.Worker.after_process(__MODULE__, job, opts)
        end
      end
    end
  end

  @doc false
  def validate(opts) do
    Validation.validate_schema(opts,
      by: {:custom, &Oban.Pro.Validation.validate_by/1},
      hold_snooze: :non_neg_integer,
      on_cancelled: {:custom, &validate_on_action(:on_cancelled, &1)},
      on_discarded: {:custom, &validate_on_action(:on_discarded, &1)},
      wait_retry: :non_neg_integer,
      wait_sleep: :timeout,
      wait_snooze: :non_neg_integer
    )
  end

  defp validate_on_action(_key, :halt), do: :ok
  defp validate_on_action(_key, :hold), do: :ok
  defp validate_on_action(_key, :ignore), do: :ok

  defp validate_on_action(key, action) do
    {:error, "expected #{key} to be :halt, :hold, or :ignore, got: #{inspect(action)}"}
  end

  @doc false
  def to_chain_key(worker, args, meta) do
    meta
    |> Map.fetch!(:chain_by)
    |> List.wrap()
    |> Enum.reduce(%{}, fn
      :worker, acc -> Map.put(acc, :worker, worker)
      [:args, keys], acc -> Map.put(acc, :args, take_keys(args, keys))
      [:meta, keys], acc -> Map.put(acc, :meta, take_keys(meta, keys))
    end)
    |> :erlang.phash2()
  end

  defp take_keys(map, keys) do
    keys
    |> Enum.sort()
    |> Enum.map(fn key -> map[key] || map[to_string(key)] end)
  end

  @doc false
  def maybe_process(job, worker, wait_count \\ 0)

  # It's running in :inline testing mode and the job doesn't have an id.
  def maybe_process(%Job{id: nil} = job, worker, _wait_count) do
    worker.process(job)
  end

  def maybe_process(%Job{conf: conf, id: id, meta: meta} = job, worker, wait_count) do
    meta = meta_with_defaults(meta)

    query =
      Job
      |> select([j], {j.state, j.scheduled_at})
      |> where([j], fragment("? @> ?", j.meta, ^%{chain_key: meta.chain_key}))
      |> where([j], j.state != "completed")
      |> where([j], j.id < ^id)
      |> order_by(desc: :id)
      |> limit(1)

    case Repo.one(conf, query) do
      {"executing", _at} ->
        if wait_count >= meta.wait_retry do
          {:snooze, meta.wait_snooze}
        else
          Process.sleep(meta.wait_sleep)

          maybe_process(job, worker, wait_count + 1)
        end

      {state, _at} when state in ~w(cancelled discarded) ->
        key = if state == "cancelled", do: :on_cancelled, else: :on_discarded

        case Map.fetch!(meta, key) do
          "halt" -> {:cancel, "chain halted by upstream job"}
          "hold" -> {:snooze, meta.hold_snooze}
          "ignore" -> worker.process(job)
        end

      {_state, scheduled_at} ->
        seconds =
          scheduled_at
          |> DateTime.diff(DateTime.utc_now())
          |> max(meta.wait_snooze)

        {:snooze, seconds}

      nil ->
        worker.process(job)
    end
  end

  defp meta_with_defaults(meta) do
    Map.new(@default_meta, fn {key, val} -> {key, meta[to_string(key)] || val} end)
  end
end
