extreme

defmodule Extreme.PersistentSubscription do
  use GenServer
  require Logger
  alias Extreme.Msg, as: ExMsg

  def start_link(connection_settings, subscriber, params) do
    GenServer.start_link(__MODULE__, {connection_settings, subscriber, params})
  end

  def init({connection_settings, subscriber, {subscription, stream, buffer_size}}) do
    state = %{
      connection_settings: connection_settings,
      subscriber: subscriber,
      subscription_ref: Process.monitor(subscriber),
      subscription_id: nil,
      connection: nil,
      params: %{subscription: subscription, stream: stream, buffer_size: buffer_size},
      status: :initialized,
    }
    GenServer.cast(self(), :connect)
    {:ok, state}
  end

  # confirm receipt of an event
  def ack(subscription, %{link: link}, correlation_id) when not is_nil(link) do
    GenServer.call(subscription, {:ack, link.event_id, correlation_id})
  end

  def ack(subscription, %{event: event}, correlation_id) when not is_nil(event) do
    GenServer.call(subscription, {:ack, event.event_id, correlation_id})
  end

  def ack(subscription, event_id, correlation_id) when is_binary(event_id) do
    GenServer.call(subscription, {:ack, event_id, correlation_id})
  end

  def nack(_, _, _, _, message \\ nil)
  def nack(subscription, %{link: link}, correlation_id, nack_action, message) when not is_nil(link) and is_atom(nack_action) do
    GenServer.call(subscription, {:nack, link.event_id, correlation_id, nack_action, message})
  end

  def nack(subscription, %{event: event}, correlation_id, nack_action, message) when is_atom(nack_action) do
    GenServer.call(subscription, {:nack, event.event_id, correlation_id, nack_action, message})
  end

  def handle_cast(:connect, %{connection_settings: connection_settings, params: params} = state) do
    # create a connection to the event store for this persistent subscription
    {:ok, connection} = Extreme.start_link(connection_settings)

    {:ok, %ExMsg.PersistentSubscriptionConfirmation{subscription_id: subscription_id}} = GenServer.call(connection, {:subscribe, self(), connect(params)})

    Logger.debug(fn -> "Successfully connected to persistent subscription id: #{inspect subscription_id}" end)

    {:noreply, %{state | connection: connection, subscription_id: subscription_id, status: :subscribed}}
  end

  def handle_cast({:ok, %ExMsg.PersistentSubscriptionStreamEventAppeared{event: event} = msg, correlation_id}, %{subscription_id: subscription_id, subscriber: subscriber} = state) do
    Logger.debug(fn -> "Persistent subscription #{inspect subscription_id} event appeared: #{inspect msg} correlation_id: #{inspect correlation_id}" end)
    send(subscriber, {:on_event, event, correlation_id})
    {:noreply, state}
  end

  def handle_cast({:ok, %ExMsg.SubscriptionDropped{ reason: :Unsubscribed}}, %{subscription_id: subscription_id} = state) do
    Logger.info(fn -> "Stopping persistent subscription #{inspect subscription_id} as subscriber has been unsubscribed" end)
    {:stop, {:shutdown, :subscriber_down}, state}
  end

  def handle_call({:ack, event_id, correlation_id}, _from, %{connection: connection, subscription_id: subscription_id} = state) do
    Logger.debug(fn -> "Persistent subscription #{inspect subscription_id} ack event id: #{inspect event_id} correlation_id: #{inspect correlation_id}" end)
    :ok = GenServer.call(connection, {:ack, ack_event(subscription_id, event_id), correlation_id})
    {:reply, :ok, state}
  end

  def handle_call({:nack, event_id, correlation_id, nack_action, message}, _from, %{connection: connection, subscription_id: subscription_id} = state) do
    Logger.debug(fn -> "Persistent subscription #{inspect subscription_id} nack event id: #{inspect event_id} correlation_id: #{inspect correlation_id} nack_action: #{inspect nack_action}" end)
    :ok = GenServer.call(connection, {:nack, nack_event(subscription_id, event_id, nack_action, message), correlation_id})
    {:reply, :ok, state}
  end

  # stop persistent subscription process when subscriber process is down
  def handle_info({:DOWN, ref, :process, _pid, reason}, %{subscription_ref: ref, subscription_id: subscription_id} = state) do
    Logger.info(fn -> "Stopping persistent subscription #{inspect subscription_id} as subscriber is down due to: #{inspect reason}" end)
    {:stop, {:shutdown, :subscriber_down}, state}
  end

  defp connect(params) do
    ExMsg.ConnectToPersistentSubscription.new(
      subscription_id: params.subscription,
      event_stream_id: params.stream,
      allowed_in_flight_messages: params.buffer_size
    )
  end

  defp nack_event(subscription_id, event_id, nack_action, message) do
    ExMsg.PersistentSubscriptionNakEvents.new(
      subscription_id: subscription_id,
      processed_event_ids: [event_id],
      message: message,
      action: nack_action
    )
  end

  defp ack_event(subscription_id, event_id) do
    ExMsg.PersistentSubscriptionAckEvents.new(
      subscription_id: subscription_id,
      processed_event_ids: [event_id]
    )
  end
end