batch_please

defmodule BatchPlease do
  @moduledoc ~S"""
  BatchPlease is a tool for collecting batches of items, and doing something
  with each batch when it reaches a certain size or age.

  It is built on top of GenServer, implemented as a behaviour,
  and invoked through `use BatchPlease`.

  It is useful to build specialized batch collectors on top of BatchPlease,
  in order to abstract more details from the end user.  Examples of this
  approach include `BatchPlease.MemoryBatcher` (which stores items in memory
  when batching) and `BatchPlease.FileBatcher` (which encodes items to
  string format and accumulates them in an on-disk file until ready for
  processing).

  Simple/trivial usage example:

      defmodule Summer do
        use BatchPlease, max_batch_size: 3

        def batch_init(_opts) do
          {:ok, %{sum: 0}}
        end

        def batch_add_item(batch, item) do
          {:ok, %{batch | sum: batch.sum + item}}
        end

        def batch_process(batch) do
          IO.puts("This batch added up to #{batch.sum}")
          :ok
        end
      end

      {:ok, pid} = GenServer.start_link(Summer)

      BatchPlease.add_item(pid, 1)
      BatchPlease.add_item(pid, 2)
      BatchPlease.add_item(pid, 3) # prints "This batch added up to 6"

      BatchPlease.add_item(pid, 4)
      BatchPlease.add_item(pid, 5)
      BatchPlease.add_item(pid, 6) # prints "This batch added up to 15"
  """

  import BatchPlease.DynamicResolvers



  #### Internal types

  @typedoc ~S"""
  A GenServer performing as a batch server.
  """
  @type batch_server :: pid

  @typedoc ~S"""
  A map representing the internal state of a batch server.  This map
  contains a `batch` key, representing the state of the current batch.
  """
  @type batch_server_state :: %{
    opts: opts,

    batch_init: ((opts) -> batch_return),
    batch_add_item: ((batch, item) -> batch_return),
    batch_pre_process: ((batch) -> batch_return),
    batch_process: ((batch) -> ok_or_error),
    batch_post_process: ((batch) -> ok_or_error),
    batch_terminate: ((batch) -> ok_or_error),

    max_batch_size: non_neg_integer | nil,
    max_batch_time_ms: non_neg_integer | nil,

    batch: batch,
    flush_count: non_neg_integer,
    current_item_count: non_neg_integer,
    total_item_count: non_neg_integer,

    last_flush_system_time: non_neg_integer | nil, # :erlang.system_time(:milli_seconds)
    last_flush_monotonic_time: integer | nil,      # :erlang.monotonic_time(:milli_seconds)
    last_item: item | nil,
  }

  @typedoc ~S"""
  A map representing the state of the current batch.
  Contents of this map are implementation-specific.
  """
  @type batch :: %{}

  @typedoc ~S"""
  Represents the return value of functions which generate a new batch state
  (`batch_init/1` and `batch_add_item/2`).
  """
  @type batch_return :: {:ok, batch} | {:error, String.t}

  @typedoc ~S"""
  Return value of functions which may fail, but do not return a new batch
  state (`batch_process/1` and `batch_terminate/1`).
  """
  @type ok_or_error :: :ok | {:error, String.t}

  @typedoc ~S"""
  Any item that can be added to a batch.
  """
  @type item :: any

  @typedoc ~S"""
  Configuration parameters for a batch server.
  """
  @type opts :: Keyword.t



  #### Behaviour stuff

  @doc ~S"""
  Creates a new batch state, given the configuration options in `opts`.
  This function is called not just once, but every time a new batch
  is created (i.e., at startup and after every flush).
  Defaults to creating an empty map, returning `{:ok, %{}}`.

  Returns the updated batch state or an error message.
  """
  @callback batch_init(opts) :: batch_return


  @doc ~S"""
  Adds an item to the batch represented in `batch`.

  Returns the updated batch state or an error message.
  """
  @callback batch_add_item(batch, item) :: batch_return


  @doc ~S"""
  Performs some pre-processing on the batch, before it is passed to
  `batch_process/1`.  Returns the updated batch state or an error message.

  This is an optional callback.
  """
  @callback batch_pre_process(batch) :: batch_return


  @doc ~S"""
  Processes the batch, whatever that entails.

  This function may operate synchronously or asynchronously,
  according to developer preference.  If it operates synchronously,
  calls to `BatchPlease.flush/1` will block until finished.

  Returns `:ok` on success, or `{:error, message}` otherwise.
  """
  @callback batch_process(batch) :: ok_or_error


  @doc ~S"""
  Performs some post-processing on the batch, after `batch_process/1`
  has completed successfully.  Does not return an updated batch, because
  this operation is immediately followed by `batch_init/1` to create a new
  batch.

  Returns `:ok` on success, or `{:error, message}` otherwise.

  This is an optional callback.
  """
  @callback batch_post_process(batch) :: ok_or_error


  @doc ~S"""
  Cleans up batch state before the batcher process is terminated.
  Defaults to no-op.  This is not guaranteed to be called at
  termination time -- for more information, see:
  https://hexdocs.pm/elixir/GenServer.html#c:terminate/2

  Returns `:ok` on success, or `{:error, message}` otherwise.

  This is an optional callback.
  """
  @callback batch_terminate(batch) :: ok_or_error

  @optional_callbacks [
    batch_terminate: 1,
    batch_pre_process: 1,
    batch_post_process: 1,
  ]



  #### Public functions

  @doc ~S"""
  Adds an item to a batch.
  Returns `:ok` on success, or `{:error, message}` otherwise.
  """
  @spec add_item(batch_server, item) :: :ok | {:error, String.t}
  def add_item(batch_server, item) do
    GenServer.call(batch_server, {:add_item, item})
  end

  @doc ~S"""
  Forces the processing and flushing of a batch.
  Returns `:ok` on success, or `{:error, message}` otherwise.
  """
  @spec flush(batch_server) :: :ok | {:error, String.t}
  def flush(batch_server) do
    GenServer.call(batch_server, {:flush})
  end


  @doc false
  def get_internal_state(batch_server) do
    GenServer.call(batch_server, {:get_internal_state})
  end


  #### GenServer implementation

  @doc false
  @spec init(Keyword.t, atom) :: batch_return
  def init(opts, module) do
    state = %{
      opts: opts,     ## invocation opts
      module: module, ## module containing implementation (if any)

      batch_init: opts[:batch_init],                 ## overrides module impl
      batch_add_item: opts[:batch_add_item],         ## overrides module impl
      batch_pre_process: opts[:batch_pre_process],   ## overrides module impl
      batch_post_process: opts[:batch_post_process], ## overrides module impl
      batch_process: opts[:batch_process],           ## overrides module impl
      batch_terminate: opts[:batch_terminate],       ## overrides module impl

      max_batch_size: opts[:max_batch_size],       ## nil == no limit

      ## TODO
      #max_batch_time_ms: opts[:max_batch_time_ms], ## nil == no limit

      batch: nil,            ## batch state
      flush_count: 0,        ## number of flushes since launch
      current_item_count: 0, ## number of items in current batch
      total_item_count: 0,   ## number of items added since launch

      last_flush_system_time: nil,    ## milliseconds since unix epoch
      last_flush_monotonic_time: nil, ## monotonic milliseconds
      last_item: nil,                 ## last item added
    }

    {:ok, batch} = do_batch_init(state, opts)

    {:ok, %{state | batch: batch}}
  end


  @doc false
  def handle_call({:add_item, item}, _from, state) do
    ## FIXME duplicated logic from :flush
    state = cond do
      state.max_batch_size && state.max_batch_size <= state.current_item_count ->
        :ok = process(state, state.batch)
        {:ok, new_batch} = do_batch_init(state, state.opts)
        %{state |
          batch: new_batch,
          last_flush_monotonic_time: :erlang.monotonic_time(:milli_seconds),
          last_flush_system_time: :erlang.system_time(:milli_seconds),
          flush_count: state.flush_count + 1,
          current_item_count: 0
        }
      :else ->
        state
    end

    case do_batch_add_item(state, state.batch, item) do
      {:ok, batch} ->
        {:reply, :ok, %{state |
          batch: batch,
          current_item_count: state.current_item_count + 1,
          total_item_count: state.total_item_count + 1,
          last_item: item,
        }}
      {:error, msg} ->
        {:reply, {:error, msg}, state}
    end
  end

  def handle_call({:flush}, _from, state) do
    case do_batch_process(state, state.batch) do
      :ok ->
        {:ok, new_batch} = do_batch_init(state, state.opts)
        {:reply, :ok, %{state |
          batch: new_batch,
          flush_count: state.flush_count + 1,
          last_flush_monotonic_time: :erlang.monotonic_time(:milli_seconds),
          last_flush_system_time: :erlang.system_time(:milli_seconds),
          current_item_count: 0,
        }}
      {:error, msg} ->
        {:reply, {:error, msg}, state}
    end
  end

  def handle_call({:get_internal_state}, _from, state) do
    {:reply, state, state}
  end

  @doc false
  def terminate(_reason, state) do
    do_batch_terminate(state, state.batch)
  end




  defp process(state, batch) do
    with {:ok, batch} <- do_batch_pre_process(state, batch),
         :ok <- do_batch_process(state, batch)
    do
      do_batch_post_process(state, batch)
    end
  end



  #### `use BatchPlease`

  @doc false
  defmacro __using__(opts) do
    quote do
      use GenServer
      def init(args), do: BatchPlease.init(unquote(opts) ++ args, __MODULE__)
      def handle_call(msg, from, state), do: BatchPlease.handle_call(msg, from, state)
      def terminate(reason, state), do: BatchPlease.terminate(reason, state)
      @behaviour BatchPlease
    end
  end
end