drinkup

defmodule Drinkup.Tap.Event do
  @moduledoc """
  Event handling and dispatch for Tap events.

  Parses incoming JSON events from Tap and dispatches them to the configured
  consumer via Task.Supervisor. After successful processing, sends an ack
  message back to the socket.
  """

  require Logger
  alias Drinkup.Tap.{Event, Options}

  @type t() :: Event.Record.t() | Event.Identity.t()

  @doc """
  Parse a JSON map into an event struct.

  Returns the appropriate event struct based on the "type" field.
  """
  @spec from(map()) :: t() | nil
  def from(%{"type" => "record"} = payload), do: Event.Record.from(payload)
  def from(%{"type" => "identity"} = payload), do: Event.Identity.from(payload)
  def from(_payload), do: nil

  @doc """
  Dispatch an event to the consumer via Task.Supervisor.

  Spawns a task that:
  1. Processes the event via the consumer's handle_event/1 callback
  2. Sends an ack to Tap if acks are enabled and the consumer returns :ok, {:ok, _}, or nil
  3. Does not ack if the consumer returns an error-like value or raises an exception

  Consumer return value semantics (when acks are enabled):
  - `:ok` or `{:ok, any()}` or `nil` -> Success, send ack
  - `{:error, _}` or any error-like tuple -> Failure, don't ack (Tap will retry)
  - Exception raised -> Failure, don't ack (Tap will retry)

  If `disable_acks: true` is set in options, no acks are sent regardless of
  consumer return value.
  """
  @spec dispatch(t(), Options.t(), pid(), :gun.stream_ref()) :: :ok
  def dispatch(
        event,
        %Options{consumer: consumer, name: name, disable_acks: disable_acks},
        conn,
        stream
      ) do
    supervisor_name = {:via, Registry, {Drinkup.Registry, {name, TapTasks}}}
    event_id = get_event_id(event)

    {:ok, _pid} =
      Task.Supervisor.start_child(supervisor_name, fn ->
        try do
          result = consumer.handle_event(event)

          unless disable_acks do
            case result do
              :ok ->
                send_ack(conn, stream, event_id)

              {:ok, _} ->
                send_ack(conn, stream, event_id)

              nil ->
                send_ack(conn, stream, event_id)

              :error ->
                Logger.error("Consumer returned error for event #{event_id}, not acking.")

              {:error, reason} ->
                Logger.error(
                  "Consumer returned error for event #{event_id}, not acking: #{inspect(reason)}"
                )

              _ ->
                Logger.warning(
                  "Consumer returned unexpected value for event #{event_id}, acking anyway: #{inspect(result)}"
                )

                send_ack(conn, stream, event_id)
            end
          end
        rescue
          e ->
            Logger.error(
              "Error in Tap event handler (event #{event_id}), not acking: #{Exception.format(:error, e, __STACKTRACE__)}"
            )
        end
      end)

    :ok
  end

  @spec send_ack(pid(), :gun.stream_ref(), integer()) :: :ok
  defp send_ack(conn, stream, event_id) do
    ack_message = Jason.encode!(%{type: "ack", id: event_id})

    :ok = :gun.ws_send(conn, stream, {:text, ack_message})
    Logger.debug("[Drinkup.Tap] Acked event #{event_id}")
  end

  @spec get_event_id(t()) :: integer()
  defp get_event_id(%Event.Record{id: id}), do: id
  defp get_event_id(%Event.Identity{id: id}), do: id
end