defmodule Polyn.PullConsumer do
  @moduledoc """
  Use `Polyn.PullConsumer` to connect and process messages from an existing [NATS Consumer](https://docs.nats.io/nats-concepts/jetstream/consumers)
  that was setup with [Polyn CLI](https://github.com/SpiffInc/polyn-cli). This module is a
  wrapper around `Jetstream.PullConsumer` that does schema validation with the received messages.
  A key difference that Polyn adds is that the `:consumer_name` will be taken care of for you
  by using the passed `type` and configured `:source_root`. You can pass `:source` to `start_link/3`
  to get a more specific `:consumer_name`. This type of Consumer is meant for simple use cases that
  don't involve concurrency or batching.

  ## Example

      defmodule MyApp.PullConsumer do
        use Polyn.PullConsumer

        def start_link(arg) do
          Polyn.PullConsumer.start_link(__MODULE__, arg,
            connection_name: :gnat,
            type: "user.created.v1")
        end

        @impl true
        def init(_arg) do
          {:ok, nil}
        end

        @impl true
        def handle_message(message, state) do
          # Do some processing with the message.
          {:ack, state}
        end
      end
  """

  use Jetstream.PullConsumer
  alias Polyn.Serializers.JSON

  @doc """
  Invoked when the server is started. `start_link/3` or `start/3` will block until it returns.

  `init_arg` is the argument term (second argument) passed to `start_link/3`.

  See `c:Connection.init/1` for more details.
  """
  @callback init(init_arg :: term) ::
              {:ok, state :: term()}
              | :ignore
              | {:stop, reason :: any}

  @doc """
  Invoked to synchronously process a message pulled by the consumer.
  Depending on the value it returns, the acknowledgement is or is not sent.
  Polyn will deserialize the message body into a `Polyn.Event` struct and use
  that as the first argument, followed by the original message, follwed by the state.

  ## ACK actions

  See `c:Jetstream.PullConsumer.handle_message/2` for available options

  ## Example

      def handle_message(event, _message, state) do
        IO.inspect(event)
        {:ack, state}
      end

  """
  @callback handle_message(
              event :: Polyn.Event.t(),
              message :: Jetstream.message(),
              state :: term()
            ) ::
              {ack_action, new_state}
            when ack_action: :ack | :nack | :term | :noreply, new_state: term()

  @typedoc """
  Options for starting a Polyn.PullConsumer

  * `:type` - Required. The event type to consume
  * `:connection_name` - Required. The Gnat connection identifier
  * `:source` - Optional. More specific name for the consumer to add to the `:source_root`
  * All other options will be assumed to be GenServer options
  """
  @type option ::
          {:type, binary()}
          | {:source, binary()}
          | {:connection_name, Gnat.t()}
          | GenServer.option()

  @doc false
  defmacro __using__(opts) do
    quote location: :keep, bind_quoted: [opts: opts] do
      @behaviour Polyn.PullConsumer

      @spec child_spec(arg :: GenServer.options()) :: Supervisor.child_spec()
      def child_spec(arg) do
        default = %{
          id: __MODULE__,
          start: {__MODULE__, :start_link, [arg]}
        }

        Supervisor.child_spec(default, unquote(Macro.escape(opts)))
      end

      defoverridable child_spec: 1
    end
  end

  @doc """
  Starts a pull consumer linked to the current process with the given function.

  This is often used to start the pull consumer as part of a supervision tree.

  Once the server is started, the `c:init/1` function of the given `module` is called with
  `init_arg` as its argument to initialize the server. To ensure a synchronized start-up procedure,
  this function does not return until `c:init/1` has returned.

  See `GenServer.start_link/3` for more details.

  ## Example

      {:ok, consumer} =
        Polyn.PullConsumer.start_link(ExamplePullConsumer, %{initial_arg: "foo"},
          connection_name: :gnat,
          type: "user.updated.v1",
          stream: "TEST_STREAM",
        )
  """
  @spec start_link(module(), init_arg :: term(), options :: [option()]) ::
          GenServer.on_start()
  def start_link(module, init_arg, options \\ []) when is_atom(module) and is_list(options) do
    Jetstream.PullConsumer.start_link(
      __MODULE__,
      {initial_state(module, options), init_arg},
      options
    )
  end

  @doc """
  Starts a `Jetstream.PullConsumer` process without links (outside of a supervision tree).

  See `start_link/3` for more information.
  """
  @spec start(module(), init_arg :: term(), options :: [option()]) ::
          GenServer.on_start()
  def start(module, init_arg, options \\ []) when is_atom(module) and is_list(options) do
    Jetstream.PullConsumer.start(__MODULE__, {initial_state(module, options), init_arg}, options)
  end

  @doc """
  Closes the pull consumer and stops underlying process.

  ## Example

      {:ok, consumer} =
        PullConsumer.start_link(ExamplePullConsumer, %{initial_arg: "foo"},
          connection_name: :gnat,
          type: "user.updated.v1",
          stream: "TEST_STREAM",
        )

      :ok = PullConsumer.close(consumer)

  """
  @spec close(consumer :: Jetstream.PullConsumer.consumer()) :: :ok
  def close(consumer) do
    Jetstream.PullConsumer.close(consumer)
  end

  @impl Jetstream.PullConsumer
  def init({%{module: module} = internal_state, init_arg}) do
    case module.init(init_arg) do
      {:ok, state} ->
        # Keep the `module` in the internal state so we can know
        # what functions to call
        internal_state = %{internal_state | state: state}

        {:ok, internal_state, connection_options(internal_state)}

      other ->
        other
    end
  end

  @impl Jetstream.PullConsumer
  def handle_message(message, %{module: module, state: state} = internal_state) do
    conn = Map.fetch!(internal_state, :connection_name)

    case JSON.deserialize(message.body, conn, store_name: internal_state.store_name) do
      {:ok, event} ->
        {response, state} = module.handle_message(event, message, state)

        {response, %{internal_state | state: state}}

      {:error, error} ->
        # If a validation error happens we want to tell NATS to stop sending the message
        # and that it won't be processed (ACKTERM) and will prevent us from raising the
        # same error over and over.
        Jetstream.ack_term(message)

        raise Polyn.ValidationException, error
    end
  end

  defp initial_state(module, opts) do
    %{
      module: module,
      state: nil,
      store_name: store_name(opts),
      connection_name: Keyword.fetch!(opts, :connection_name),
      type: Keyword.fetch!(opts, :type),
      source: Keyword.get(opts, :source)
    }
  end

  defp connection_options(%{
         type: type,
         source: source,
         connection_name: connection_name
       }) do
    consumer_name = Polyn.Naming.consumer_name(type, source)
    stream = Polyn.Naming.lookup_stream_name!(connection_name, type)
    [connection_name: connection_name, stream_name: stream, consumer_name: consumer_name]
  end

  defp store_name(opts) do
    Keyword.get(opts, :store_name, Polyn.SchemaStore.store_name())
  end
end