finitomata

defmodule Finitomata.Pool do
  @moduledoc """
  Fully asynchronous pool to manage many similar processes, like connections.

  The pool is to be started using `start_pool/1` directly or with `pool_spec/1` in the
    supervision tree.

  `initialize/2` is explicitly separated because usually this is to be done after some
    external service initialization. In a case of `AMQP` connection management, one
    would probably start the connection process and then a pool to manage channels.

  Once `initialize/2` has been called, the `run/3` function might be invoked to
    asynchronously execute the function passed as `actor` to `start_pool/1`.

  If the callbacks `on_result/2` and/or `on_error/2` are defined, they will be invoked
    respectively. Finally, the message to the calling process will be sent, unless
    the third argument in a call to `run/3` is `nil`.
  """
  @moduledoc since: "0.18.0"

  alias Finitomata.Pool.Actor

  @fsm """
  idle --> |init| ready
  idle --> |do| ready
  ready --> |do| ready
  ready --> |stop| done
  """

  @pool_size Application.compile_env(:finitomata, :pool_size, System.schedulers_online())
  @actor_prefix Application.compile_env(:finitomata, :pool_worker_prefix, "PoolWorker")

  use Finitomata, fsm: @fsm, auto_terminate: true

  @impl Finitomata
  def on_terminate(state) do
    require Logger
    Logger.info("[♻️] Terminating: " <> inspect(state))
  end

  @typedoc "The ID of the Pool"
  @type id :: Finitomata.id()

  @typedoc "The simple actor function in the pool"
  @type naive_actor :: (term() -> {:ok, term()} | {:error, any()})

  @typedoc "The actor function in the pool, receiving the state as a second argument"
  @type responsive_actor ::
          (term(), Finitomata.State.payload() -> {:ok, term()} | {:error, any()})

  @typedoc "The actor function in the pool"
  @type actor :: naive_actor() | responsive_actor()

  @typedoc "The simple handler of result/error in the pool"
  @type naive_handler :: (term() -> any())

  @typedoc "The actor function in the pool, receiving the state as a second argument"
  @type responsive_handler :: (term(), id() -> any())

  @typedoc "The handler function in the pool"
  @type handler :: naive_handler() | responsive_handler()

  defstate %{
    id: {StreamData, :atom, [:alias]},
    actor: {StreamData, :constant, [&Function.identity/1]},
    on_error: {Actor, :handler, [:error]},
    on_result: {Actor, :handler, [[]]},
    errors: [:term],
    payload: :term
  }

  @doc """
  Starts a pool of asynchronous workers wrapped by an _FSM_.
  """
  @spec start_pool([
          {:id, Finitomata.id()}
          | {:payload, :term}
          | {:count, pos_integer()}
          | {:actor, actor()}
          | {:on_error, handler()}
          | {:on_result, handler()}
        ]) ::
          GenServer.on_start()
  def start_pool(opts \\ []) do
    {id, opts} = Keyword.pop(opts, :id)
    {count, opts} = Keyword.pop(opts, :count, @pool_size)
    start_pool(id, count, opts)
  end

  @spec start_pool(
          id :: Finitomata.id(),
          count :: pos_integer(),
          [
            {:actor, actor()}
            | {:on_error, handler()}
            | {:on_result, handler()}
            | {:payload, :term}
          ]
          | %{
              required(:actor) => actor(),
              optional(:on_error) => handler(),
              optional(:on_result) => handler(),
              optional(:payload) => :term
            }
          | [{:implementation, module()} | {:payload, :term}]
          | %{required(:implementation) => module(), optional(:payload) => :term}
        ) ::
          GenServer.on_start()
  def start_pool(id, count, state) when is_list(state), do: start_pool(id, count, Map.new(state))

  def start_pool(id, count, %{implementation: impl} = state) do
    state =
      %{payload: Map.get(state, :payload), actor: &impl.actor/2}
      |> then(fn spec ->
        if function_exported?(impl, :on_result, 2),
          do: Map.put(spec, :on_result, &impl.on_result/2),
          else: spec
      end)
      |> then(fn spec ->
        if function_exported?(impl, :on_error, 2),
          do: Map.put(spec, :on_error, &impl.on_error/2),
          else: spec
      end)

    start_pool(id, count, state)
  end

  def start_pool(id, count, %{actor: actor} = state)
      when is_function(actor, 1) or is_function(actor, 2) do
    {:ok, state} = Estructura.coerce(__MODULE__, Map.put(state, :id, id))

    actor_launcher = fn ->
      Enum.each(
        1..count,
        &Infinitomata.start_fsm(id, "#{@actor_prefix}_#{&1}", Finitomata.Pool, state)
      )
    end

    id
    |> Infinitomata.start_link()
    |> case do
      {:ok, pid} ->
        actor_launcher.()
        {:ok, pid}

      {:error, {:already_started, pid}} ->
        actor_launcher.()
        {:ok, pid}

      {:error, reason} ->
        {:error, reason}
    end
  end

  @doc """
  Child spec for `Finitomata.Pool` to embed the process into a supervision tree
  """
  @spec pool_spec(keyword()) :: Supervisor.child_spec()
  def pool_spec(opts \\ []) do
    {child_opts, opts} = Keyword.pop(opts, :child_opts, %{})

    child_opts
    |> Map.new()
    |> Map.merge(%{
      id: Keyword.get(opts, :id, __MODULE__),
      start: {__MODULE__, :start_pool, [opts]}
    })
  end

  @doc """
  Initializes the started _FSM_s with the same payload, or with what `payload_fun/1`
    would return as many times as the number of workers.
  """
  @spec initialize(Finitomata.id(), (pos_integer() -> any()) | any()) :: :ok
  def initialize(id, payload_fun) when is_function(payload_fun, 1) do
    count = Infinitomata.count(id)

    Enum.each(
      1..count,
      &Infinitomata.transition(id, "#{@actor_prefix}_#{&1}", {:init, payload_fun.(&1)})
    )
  end

  def initialize(id, payload) do
    initialize(id, fn _ -> payload end)
  end

  @doc """
  The runner for the `actor` function with the specified payload.

  Basically, upon calling `run/3`, the following chain of calls would have happened:

  1. `actor.(payload, state)` (or `actor.(payload)` if the function of arity one had been given)
  2. `on_result(result, id)` / `on_error(result, id)` if callbacks are specified
  3. the message of the shape `{:transition, :success/:failure, self(), {payload, result, on_result/on_error}})` 
     will be sent to `pid` unless `nil` given as a third argument 
  """
  @spec run(Finitomata.id(), Finitomata.event_payload(), pid()) :: :ok
  def run(id, payload, pid \\ self()) do
    Infinitomata.transition(id, Infinitomata.random(id), {:do, {pid, payload}})
  end

  @doc false
  def on_transition(:idle, :init, payload, %{actor: _actor} = state) do
    {:ok, :ready, put_in(state, [:payload], payload)}
  end

  @doc false
  def on_transition(current, :do, {pid, payload}, %{actor: actor} = state)
      when current in ~w|idle ready|a do
    state =
      case invoke(actor, payload, state.payload) do
        {:ok, result} ->
          on_result = invoke(state.on_result, result, state.id)

          if not is_nil(pid),
            do: send(pid, {:transition, :success, self(), {payload, result, on_result}})

          state

        {:error, reason} ->
          on_error = invoke(state.on_error, reason, state.id)

          if not is_nil(pid),
            do: send(pid, {:transition, :failure, self(), {payload, reason, on_error}})

          update_in(state, [:errors], &[reason | &1])
      end

    {:ok, :ready, state}
  end

  defp invoke(fun, arg1, arg2) do
    case fun do
      nil -> nil
      f when is_function(f, 1) -> f.(arg1)
      f when is_function(f, 2) -> f.(arg1, arg2)
    end
  end
end