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 usually invoked through `use BatchPlease`.

  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_flush(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"

  It is useful to build specialized batch collectors on top of BatchPlease,
  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).


  ## Behaviour

  The BatchPlease behaviour consists of several required callbacks:

  * `batch_init/1` takes a keyword list of options (from `use BatchPlease`
    and `GenServer.start_link/2-3`) and returns `{:error, msg}` or
    `{:ok, state}`, where `state` is a map representing the state of
    a single batch.  This function is called at the start of every batch.

  * `batch_add_item/2` adds an item to the batch state, returning
    `{:ok, state}` or `{:error, msg}`.

  * `batch_flush/1` performs an operation on the batch when it is ready
    for processing.  It returns `:ok` or `{:error, msg}`; returning an
    updated state is not necessary, because iff `batch_flush/1` returns
    successfully, a new batch is created with `batch_init/1`.

  As well as some optional callbacks:

  * `batch_pre_flush/1` can be used to perform pre-processing on a batch
    before `batch_flush/1` is called.

  * `batch_post_flush/1` can be used to perform post-processing on a batch
    after `batch_flush/1` returns successfully.

  * `batch_terminate/1` performs cleanup on a batch when the batch server
    is terminating.  Warning: it hooks into `GenServer.terminate/2`, which
    is not guaranteed to execute upon server shutdown!  See the GenServer
    docs for more information:
    https://hexdocs.pm/elixir/GenServer.html#c:terminate/2

  * `should_flush/1` is used to implement custom logic to determine when to
    flush a batch.


  ## Options

  BatchPlease supports several options to customize operation.

  They deal with flushing.  Each may be set to `nil` to disable
  that mode of behavior.

  * `max_batch_size: X` can be set to a positive integer to specify that
    flushing should occur when the batch hits size `X`.
    This check takes place before and after every `batch_add_item/2` call.

  * `max_time_since_last_flush: X` can be set to a positive integer to specify
    that a batch should flush if at least `X` milliseconds have passed since
    the last flush.
    This check takes place before and after every `batch_add_item/2` call.

  * `max_time_since_first_item: X` can be set to a positive integer to specify
    that a batch should flush if at least `X` milliseconds have passed
    since the time the first item was added to the current batch.
    This check takes place before and after every `batch_add_item/2` call.

  * `flush_interval: X` can be set to a positive integer to specify that
    flushing should take place every `X` milliseconds.  This option uses an
    internal timer, and is therefore independent from `batch_add_item/2`.

  """



  #### `use BatchPlease`

  @doc false
  defmacro __using__(opts) do
    quote do
      use GenServer

      @impl GenServer
      def init(args), do: BatchPlease.Impl.init(args ++ unquote(opts), __MODULE__)

      @impl GenServer
      def handle_call(msg, from, state), do: BatchPlease.Impl.handle_call(msg, from, state)

      @impl GenServer
      def handle_cast(msg, state), do: BatchPlease.Impl.handle_cast(msg, state)

      @impl GenServer
      def handle_info(msg, state), do: BatchPlease.Impl.handle_info(msg, state)

      @impl GenServer
      def terminate(reason, state), do: BatchPlease.Impl.terminate(reason, state)

      @behaviour BatchPlease
    end
  end


  #### 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 state :: %{
    opts: opts,
    module: atom,
    batch: batch,
    last_item: item | nil,
    flush_timer: pid,
    config: state_config,
    overrides: state_overrides,
    counts: state_counts,
    times: state_times,
  }

  @type state_config :: %{
    lazy_flush: boolean | nil,
    max_batch_size: non_neg_integer | nil,
    max_time_since_last_flush: non_neg_integer | nil,
    max_time_since_first_item: non_neg_integer | nil,
    flush_interval: non_neg_integer | nil,
  }

  @type state_overrides :: %{
    batch_init: ((opts) -> batch_return) | nil,
    batch_add_item: ((batch, item) -> batch_return) | nil,
    batch_pre_flush: ((batch) -> batch_return) | nil,
    batch_flush: ((batch) -> ok_or_error) | nil,
    batch_post_flush: ((batch) -> ok_or_error) | nil,
    batch_terminate: ((batch) -> ok_or_error) | nil,
    should_flush: ((state) -> boolean) | nil,
  }

  @type state_counts :: %{
    batch_items: non_neg_integer,
    total_items: non_neg_integer,
    flushes: non_neg_integer,
  }

  @type state_times :: %{
    first_item_of_batch: integer | nil,
    last_flush: integer | nil,
  }

  @type state_return :: {:ok, state} | {:error, String.t}

  @type reply_return :: {ok_or_error, state}

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

  @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_flush/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 :: [option]

  @type option ::
    {:max_batch_size, non_neg_integer | nil} |
    {:max_time_since_last_flush, non_neg_integer | nil} |
    {:max_time_since_first_item, non_neg_integer | nil} |
    {:flush_interval, non_neg_integer | nil}



  #### Behaviour

  @doc ~S"""
  Creates a new batch state, given the configuration options in `opts`,
  which come from the options given in `use BatchPlease` and
  `GenServer.start_link/2-3`.

  This function is called every time a new batch is created (i.e., at startup
  and after every flush).

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


  @doc ~S"""
  Adds an item to the 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_flush/1`.  Returns the updated batch state or an error message.

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


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

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


  @doc ~S"""
  Performs some post-processing on the batch after `batch_flush/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_flush(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


  @doc ~S"""
  Given the current module state, returns whether the current batch
  should be processed now.  Does not prevent the evaluation of other
  flushing options (e.g., `max_batch_size`, etc).

  This is an optional callback.
  """
  @callback should_flush(state) :: boolean

  @optional_callbacks [
    batch_terminate: 1,
    batch_pre_flush: 1,
    batch_post_flush: 1,
    should_flush: 1,
  ]



  #### Public API

  @doc ~S"""
  Adds an item to a batch asynchronously.

  Set `opts[:error_pid]` to receive a message of the form `{:error, msg}`
  in case this operation fails.  Alternately, use `sync_add_item/2` for
  a synchronous version of this function.

  Returns `:ok`.
  """
  @spec add_item(batch_server, item, Keyword.t) :: :ok
  def add_item(batch_server, item, opts \\ []) do
    GenServer.cast(batch_server, {:add_item, item, opts[:error_pid]})
  end


  @doc ~S"""
  Adds an item to a batch synchronously.
  Causes a flush if appropriate, also synchronously.

  Returns `:ok` or `{:error, msg}`.
  """
  @spec sync_add_item(batch_server, item) :: :ok | {:error, String.t}
  def sync_add_item(batch_server, item) do
    GenServer.call(batch_server, {:add_item, item})
  end


  @doc ~S"""
  Forces the processing and flushing of a batch asynchronously.

  Set `opts[:error_pid]` to receive a message of the form `{:error, msg}`
  in case this operation fails.  Alternately, use `sync_flush/1` for
  a synchronous version of this function.

  Returns `:ok`.
  """
  @spec flush(batch_server, Keyword.t) :: :ok | {:error, String.t}
  def flush(batch_server, opts \\ []) do
    GenServer.cast(batch_server, {:flush, opts[:error_pid]})
  end


  @doc ~S"""
  Forces the processing and flushing of a batch synchronously.

  Returns `:ok` or `{:error, msg}`.
  """
  @spec sync_flush(batch_server) :: :ok | {:error, String.t}
  def sync_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
end