defmodule Oban.Pro.Workflow.Cascade do
  @moduledoc false

  use Oban.Pro.Worker, recorded: [limit: 256_000]

  alias Oban.Pro.Workflow

  args_schema do
    field :mod, :string
    field :fun, :string
    field :arg, :term, default: :none
  end

  @impl Oban.Pro.Worker
  def process(%{args: %{arg: arg, mod: mod, fun: fun}, conf: conf, meta: meta}) do
    mod =
      mod
      |> String.split(".")
      |> Module.safe_concat()

    fun = String.to_existing_atom(fun)

    full = Map.merge(all_context(conf, meta), all_recorded(conf, meta))
    args = if arg == :none, do: [full], else: [arg, full]

    case apply(mod, fun, args) do
      {:ok, _} = result -> result
      {:error, _} = error -> error
      {:snooze, _} = snooze -> snooze
      {:cancel, _} = cancel -> cancel
      other -> {:ok, other}
    end
  end

  defp all_context(conf, meta) do
    sup = Map.get(meta, "sup_workflow_id")
    wid = Map.get(meta, "workflow_id")
    ancestors = Map.get(meta, "ancestor_ids", [])

    (Enum.reverse(ancestors) ++ [sup, wid])
    |> Enum.filter(& &1)
    |> Enum.uniq()
    |> Enum.reduce(%{}, fn id, acc ->
      Map.merge(acc, Workflow.get_context(conf, id) || %{})
    end)
  end

  defp all_recorded(_conf, %{"deps" => []}), do: %{}

  defp all_recorded(conf, %{"deps" => deps, "workflow_id" => wid} = meta) do
    atomize? = Map.get(meta, "atom_keys", true)

    grouper = fn
      [sup, name], acc -> Map.update(acc, sup, [name], &[name | &1])
      name, acc -> Map.update(acc, wid, [name], &[name | &1])
    end

    deps
    |> Enum.reject(&match?("grafter_" <> _, &1))
    |> Enum.reduce(%{}, grouper)
    |> Enum.reduce(%{}, fn {dep_wid, names}, acc ->
      filter = if names == ["*"], do: [], else: [names: names]
      filter = if dep_wid != wid, do: Keyword.put(filter, :with_subs, true), else: filter

      recorded =
        conf
        |> Workflow.all_recorded(dep_wid, filter)
        |> then(fn map -> if atomize?, do: atomize_map(map), else: map end)

      Map.merge(acc, recorded)
    end)
  end

  defp atomize_map(map) when is_map(map) and not is_struct(map) do
    Map.new(map, fn {key, val} -> {atomize_key(key), val} end)
  end

  defp atomize_map(val), do: val

  defp atomize_key(key) when is_binary(key) do
    String.to_existing_atom(key)
  rescue
    ArgumentError -> key
  end

  defp atomize_key(key), do: key
end
