phoenix_gen_api

defmodule PhoenixGenApi.StreamCall do
  @moduledoc """
  A GenServer that manages a streaming function call.

  This process is responsible for executing a function that returns a stream of
  results, and sending those results back to the client in chunks.
  """

  use GenServer, restart: :temporary

  alias PhoenixGenApi.Structs.{FunConfig, Request, Response}
  alias PhoenixGenApi.Executor

  require Logger

  def start_link(args = %{fun_config: %FunConfig{}, request: %Request{}, receiver: nil}) do
    receiver = self()
    start_link(Map.put(args, :receiver, receiver))
  end

  def start_link(args = %{fun_config: %FunConfig{}, request: %Request{}, receiver: receiver})
      when is_pid(receiver) do
    case GenServer.start_link(__MODULE__, args) do
      {:ok, pid} ->
        Process.put({:phoenix_gen_api, :stream_call_pid, args.request.request_id}, pid)
        {:ok, pid}

      {:error, reason} ->
        {:error, reason}
    end
  end

  def stop(pid) when is_pid(pid) do
    GenServer.cast(pid, :stop)
  end

  def stop(request_id) when is_binary(request_id) do
    case Process.get({:phoenix_gen_api, :stream_call_pid, request_id}) do
      nil ->
        Logger.warning(
          "PhoenixGenApi.StreamCall, stop, not found stream for request_id: #{inspect(request_id)}"
        )

        {:error, :not_found}

      pid when is_pid(pid) ->
        Logger.debug("PhoenixGenApi.StreamCall, stop, stream call pid: #{inspect(pid)}")
        stop(pid)
    end
  end

  ### GenServer Callbacks

  @impl true
  def init(args) do
    {:ok, args, {:continue, :start_stream}}
  end

  @impl true
  def handle_continue(:start_stream, state) do
    result = Executor.sync_call(state.request, state.fun_config)

    if Response.error?(result) do
      send(state.receiver, {:stream_response, result})
      {:stop, :normal, state}
    else
      send(
        state.receiver,
        {:stream_response, Response.stream_response(state.request.request_id, result)}
      )

      {:noreply, state}
    end
  end

  @impl true
  def handle_cast(:stop, state) do
    send_completion(state)
    {:stop, :normal, state}
  end

  @impl true
  def handle_info({:result, result}, state) do
    send(
      state.receiver,
      {:stream_response, Response.stream_response(state.request.request_id, result)}
    )

    {:noreply, state}
  end

  def handle_info({:last_result, result}, state) do
    send(
      state.receiver,
      {:stream_response, Response.stream_response(state.request.request_id, result, false)}
    )

    {:stop, :normal, state}
  end

  def handle_info({:error, error}, state) do
    error_message =
      if Application.get_env(:phoenix_gen_api, :detail_error, false) do
        "Internal Server Error: #{inspect(error)}"
      else
        "Internal Server Error"
      end

    send(
      state.receiver,
      {:stream_response, Response.error_response(state.request.request_id, error_message)}
    )

    {:stop, :normal, state}
  end

  def handle_info(:complete, state) do
    send_completion(state)
    {:stop, :normal, state}
  end

  def handle_info(unknown, state) do
    Logger.warning("PhoenixGenApi.StreamCall, received unknown message: #{inspect(unknown)}")
    {:noreply, state}
  end

  @impl true
  def terminate(reason, state) do
    Logger.debug(
      "PhoenixGenApi.StreamCall, terminating for request_id: #{state.request.request_id}, reason: #{inspect(reason)}"
    )

    :ok
  end

  defp send_completion(state) do
    send(
      state.receiver,
      {:stream_response, Response.stream_end_response(state.request.request_id)}
    )
  end
end