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

  @behaviour Oban.Pro.Stage

  alias Oban.Pro.Utils
  alias Oban.Validation

  defstruct to: :return, limit: 64_000_000, safe_decode: false

  @impl Oban.Pro.Stage
  def init(_worker, true) do
    {:ok, %__MODULE__{}}
  end

  def init(_worker, opts) do
    with :ok <- validate(opts) do
      {:ok, struct(__MODULE__, opts)}
    end
  end

  @impl Oban.Pro.Stage
  def before_new(args, opts, _conf) do
    meta = %{"recorded" => true}
    opts = Keyword.update(opts, :meta, meta, &Map.merge(&1, meta))

    {:ok, args, opts}
  end

  @impl Oban.Pro.Stage
  def after_process(result, %{meta: %{"recorded" => true}} = job, conf) do
    with {:ok, return} <- result do
      encoded = Utils.encode64(return)

      cond do
        job.conf.engine != Oban.Pro.Engines.Smart ->
          {:error, "recording requires the Smart engine"}

        byte_size(encoded) > conf.limit ->
          {:error, "return is #{byte_size(encoded)} bytes, larger than the limit: #{conf.limit}"}

        true ->
          Process.put(:oban_recorded, %{to_string(conf.to) => encoded})

          :ok
      end
    end
  end

  def after_process(_result, _job, _conf), do: :ok

  @spec fetch_recorded(Oban.Job.t(), conf :: map()) :: {:ok, term()} | {:error, term()}
  def fetch_recorded(job, conf) do
    to = to_string(conf.to)

    case job.meta do
      %{"recorded" => true, ^to => value} ->
        {:ok, Utils.decode64(value, decode_opts(job.meta))}

      _ ->
        {:error, :missing}
    end
  end

  # Helpers

  defp validate(opts) do
    Validation.validate_schema(opts,
      to: {:or, [:atom, :string]},
      limit: :pos_integer,
      safe_decode: :boolean
    )
  end

  defp decode_opts(%{"safe_decode" => true}), do: [:safe]
  defp decode_opts(_meta), do: []
end
