defmodule Beeline do
  @schema Beeline.Config.schema()
  @producer_schema Beeline.Producer.schema()

  @moduledoc """
  A tool for building in-order GenStage topologies for EventStoreDB

  Beeline provides a Broadway-like experience for building GenStage
  topologies for consuming streams of events from EventStoreDB in-order,
  usually one event at a time. Beeline aims to close over the supervision
  and basic configuration of the producer(s), as well as some of the
  run-of-the-mill procedure done in the `c:GenStage.init/1` callback of
  the consumer such as linking the producer process(es).

  ## The Beeline Topology

  Beeline creates a topology of GenStage, GenServer, and Supervisor processes.
  This topology looks like the following:

  ```text
  Supervisor
  ├── HealthChecker*
  └── StageSupervisor
      ├── Producer*
      └── Consumer
  ```

  Let's break these down from the bottom up:

  * "Consumer" - the GenStage consumer module which invokes
    `Beeline.start_link/2`, handles events, and increments stream
    positions.
  * "Producer*" - one or more GenStage producers which feed the consumer.
    These producers are declared with the `:producers` key and may either
    be `Kelvin.InOrderSubscription`, `Volley.InOrderSubscription`, or
    `Beeline.DummyProducer` producer modules.
  * "StageSupervisor" - a supervisor for the GenStage pipeline. This supervisor
    has a `:transient` restart strategy so that if the GenStage pipeline halts
    on an event it cannot handle, the `StageSupervisor` supervision tree is
    brought down but not the entire supervision tree. This behavior is
    desirable so that the health-checker process can continue reading the
    stream positions and so that an operator can perform any necessary
    manual intervention on the crashed supervision tree (for example,
    skipping the failure event).
  * "HealthChecker*" - a GenServer which periodically polls the stream positions
    of a producer. There is one health checker process per producer.
  * "Supervisor" - a top-level supervisor. This supervisor has a `:permanent`
    restart strategy.

  See the `start_link/2` documentation for a full configuration reference and
  examples.
  """

  defmacro __using__(_opts) do
    quote do
      use GenStage

      @impl GenStage
      def init(config) do
        producers =
          Enum.map(config.producers, fn {_key, producer} ->
            {producer.name,
             min_demand: producer.min_demand, max_demand: producer.max_demand}
          end)

        {:consumer, config.context, subscribe_to: producers}
      end

      defoverridable init: 1
    end
  end

  @doc """
  Starts a Beeline topology

  ## Options

  #{NimbleOptions.docs(@schema)}

  #### Producer options

  #{NimbleOptions.docs(@producer_schema)}

  ## Examples

      defmodule MyEventHandler do
        use Beeline

        def start_link(_opts) do
          Beeline.start_link(MyEventHandler,
            name: MyEventHandler,
            producers: [
              default: [
                name: MyEventHandler.EventListener,
                stream_name: "$ce-BoundedContext.AggregateName",
                connection: MyEventHandler.EventStoreDBConnection,
                adapter: :kelvin
              ]
            ]
          )
        end

        # .. GenStage callbacks

        @impl GenStage
        def handle_events([subscription_event], _from, state) do
          # .. handle the events one-by-one ..

          {:noreply, [], state}
        end
      end
  """
  @doc since: "0.1.0"
  @spec start_link(module :: module(), opts :: Keyword.t()) ::
          Supervisor.on_start()
  def start_link(module, opts) do
    case NimbleOptions.validate(opts, @schema) do
      {:error, reason} ->
        raise ArgumentError,
              "invalid configuration given to Beeline.start_link/2, " <>
                reason.message

      {:ok, opts} ->
        opts
        |> Keyword.put(:module, module)
        |> Beeline.Config.source()
        |> Beeline.Topology.start_link()
    end
  end

  @doc """
  Restarts the supervision tree of GenStages for the given Beeline topology

  This can be useful for manual intervention by a human operator in a remote
  console session, if the GenStage supervision tree crashes and exceeds the
  retry limits.

  ## Examples

      iex> Beeline.restart_stages(MyEventHandler)
      :ok
  """
  @spec restart_stages(module()) :: :ok | {:error, term()}
  def restart_stages(beeline) when is_atom(beeline) do
    beeline
    |> Module.concat(Topology)
    |> GenServer.call(:restart_stages)
  end

  @doc """
  Decodes the body of a subscription event

  This function performs JSON decoding if necessary and converts maps with
  string keys into maps keyed by atoms. This This can potentially lead to
  atom exhaustion, but the allowed atom count is quite high and usually this
  concern is only theoretical.

  ## Examples

      @impl GenStage
      def handle_events([subscription_event], _from, state) do
        event = Beeline.decode_event(subscription_event)
        # ..
  """
  defdelegate decode_event(subscription_event), to: Beeline.EventStoreDB

  @doc """
  Determines the stream position of the subscription event

  This function prefers link stream positions if available. This means that if
  the subscription from which the event is emitted is reading a projected
  stream such as a category stream, the returned stream position will reflect
  the position in the projected stream instead of the origin stream.

  ## Examples

      @impl GenStage
      def handle_events([subscription_event], _from, state) do
        # consume the events

        MyApp.Repo.transaction(fn ->
          # save some state

          producer = Beeline.producer(subscription_event)
          stream_position = Beeline.stream_position(subscription_event)
          MyApp.StreamPosition.persist(producer, stream_position)
        end)
      end
  """
  defdelegate stream_position(subscription_event), to: Beeline.EventStoreDB

  @doc """
  Determines which producer emitted the subscription event

  This can be useful in order to save stream positions when a consumer is
  subscribed to multiple producers. Should be used in tandem with
  `stream_position/1`.
  """
  defdelegate producer(subscription_event), to: Beeline.EventStoreDB

  @doc """
  Wraps an event in a subscription event packet

  This can be useful for building test events to pass through the dummy
  producer.
  """
  @spec as_subscription_event(map(), atom()) :: {atom(), map()}
  def as_subscription_event(event, producer) do
    {producer, %{event: %{data: Jason.encode!(event), event_number: 0}}}
  end

  @doc """
  Gives a set of events to a topology's dummy producer

  This function can be used to test running events through a topology.
  If there are multiple producers, one is picked at random.
  """
  def test_events(events, beeline) when is_atom(beeline) do
    beeline
    |> Module.concat(Topology)
    |> GenServer.call({:test_events, events})
  end
end