phoenix_session_process

defmodule Phoenix.SessionProcess.ProcessSupervisor do
  require Logger

  # Automatically defines child_spec/1
  use DynamicSupervisor

  alias Phoenix.SessionProcess.{Telemetry, Error}

  @spec start_link(any()) :: :ignore | {:error, any()} | {:ok, pid()}
  def start_link(init_arg) do
    DynamicSupervisor.start_link(__MODULE__, init_arg, name: __MODULE__)
  end

  def start_child(worker, worker_arg) do
    worker_spec = {worker, worker_arg}
    Logger.debug("Start Child Worker: #{inspect(worker_spec)}")
    DynamicSupervisor.start_child(__MODULE__, worker_spec)
  end

  def terminate_child(pid) do
    Logger.debug("Terminating Child Worker: #{inspect(pid)}")
    DynamicSupervisor.terminate_child(__MODULE__, pid)
  end

  @impl true
  def init(_init_arg) do
    DynamicSupervisor.init(strategy: :one_for_one)
  end

  def start_session(session_id) do
    start_session_with_module(session_id, Phoenix.SessionProcess.Config.session_process())
  end

  def start_session(session_id, module) do
    start_session_with_module(session_id, module)
  end

  def start_session(session_id, module, arg) do
    start_session_with_module(session_id, module, arg)
  end

  @spec session_process_started?(binary()) :: boolean()
  def session_process_started?(session_id) do
    session_id
    |> session_process_pid()
    |> is_pid()
  end

  @spec terminate_session(binary()) :: :ok | {:error, :not_found}
  def terminate_session(session_id) do
    Logger.debug("End Session: #{inspect(session_id)}")
    start_time = System.monotonic_time()

    case session_process_pid(session_id) do
      nil ->
        Error.session_not_found(session_id)

      pid ->
        module = get_session_module(pid)
        result = terminate_child(pid)
        duration = System.monotonic_time() - start_time

        case result do
          :ok ->
            Telemetry.emit_session_stop(session_id, module, pid, duration: duration)
            :ok

          error ->
            Telemetry.emit_session_cleanup_error(session_id, module, error, duration: duration)
            error
        end
    end
  end

  def call_on_session(session_id, request, timeout \\ 15_000) do
    start_time = System.monotonic_time()

    case session_process_pid(session_id) do
      nil ->
        Error.session_not_found(session_id)

      pid ->
        module = get_session_module(pid)
        do_call_on_session(session_id, pid, module, request, timeout, start_time)
    end
  end

  def cast_on_session(session_id, request) do
    start_time = System.monotonic_time()

    case session_process_pid(session_id) do
      nil ->
        Error.session_not_found(session_id)

      pid ->
        module = get_session_module(pid)
        do_cast_on_session(session_id, pid, module, request, start_time)
    end
  end

  defp do_call_on_session(session_id, pid, module, request, timeout, start_time) do
    try do
      result = GenServer.call(pid, request, timeout)
      duration = System.monotonic_time() - start_time
      Telemetry.emit_session_call(session_id, module, pid, request, duration: duration)
      result
    catch
      :exit, {:timeout, _} ->
        duration = System.monotonic_time() - start_time

        Telemetry.emit_communication_error(session_id, module, :call, :timeout,
          duration: duration
        )

        Error.timeout(timeout)

      :exit, reason ->
        duration = System.monotonic_time() - start_time
        Telemetry.emit_communication_error(session_id, module, :call, reason, duration: duration)
        Error.call_failed(module, :call, {request}, reason)
    end
  end

  defp do_cast_on_session(session_id, pid, module, request, start_time) do
    try do
      result = GenServer.cast(pid, request)
      duration = System.monotonic_time() - start_time
      Telemetry.emit_session_cast(session_id, module, pid, request, duration: duration)
      result
    catch
      :exit, reason ->
        duration = System.monotonic_time() - start_time
        Telemetry.emit_communication_error(session_id, module, :cast, reason, duration: duration)
        Error.cast_failed(module, :cast, {request}, reason)
    end
  end

  @spec child_name(binary()) :: {:via, Registry, {Phoenix.SessionProcess.Registry, binary()}}
  def child_name(session_id) do
    {:via, Registry, {Phoenix.SessionProcess.Registry, session_id}}
  end

  @spec session_process_pid(binary()) :: nil | pid()
  def session_process_pid(session_id) do
    case Registry.whereis_name({Phoenix.SessionProcess.Registry, session_id}) do
      :undefined -> nil
      pid -> pid
    end
  end

  defp start_session_with_module(session_id, module, arg \\ nil) do
    start_time = System.monotonic_time()

    with :ok <- validate_session_id(session_id),
         :ok <- check_session_limits() do
      Logger.debug("Start Session: #{inspect(session_id)}")

      worker_args =
        if arg, do: [name: child_name(session_id), arg: arg], else: [name: child_name(session_id)]

      spec = {module, worker_args}

      case DynamicSupervisor.start_child(__MODULE__, spec) do
        {:ok, pid} = result ->
          Registry.register(Phoenix.SessionProcess.Registry, pid, module)
          Phoenix.SessionProcess.Cleanup.schedule_session_cleanup(session_id)
          duration = System.monotonic_time() - start_time
          Telemetry.emit_session_start(session_id, module, pid, duration: duration)
          result

        {:ok, pid, _info} = result ->
          Registry.register(Phoenix.SessionProcess.Registry, pid, module)
          Phoenix.SessionProcess.Cleanup.schedule_session_cleanup(session_id)
          duration = System.monotonic_time() - start_time
          Telemetry.emit_session_start(session_id, module, pid, duration: duration)
          result

        {:error, {:already_started, pid}} = result ->
          duration = System.monotonic_time() - start_time
          Telemetry.emit_session_start(session_id, module, pid, duration: duration)
          result

        {:error, reason} = error ->
          duration = System.monotonic_time() - start_time
          Telemetry.emit_session_start_error(session_id, module, reason, duration: duration)
          error
      end
    else
      {:error, :invalid_session_id} ->
        duration = System.monotonic_time() - start_time

        Telemetry.emit_session_start_error(session_id, module, :invalid_session_id,
          duration: duration
        )

        Error.invalid_session_id(session_id)

      {:error, :session_limit_reached} ->
        max_sessions = Phoenix.SessionProcess.Config.max_sessions()
        duration = System.monotonic_time() - start_time

        Telemetry.emit_session_start_error(
          session_id,
          module,
          {:session_limit_reached, max_sessions},
          duration: duration
        )

        Error.session_limit_reached(max_sessions)
    end
  end

  defp get_session_module(pid) do
    case Registry.lookup(Phoenix.SessionProcess.Registry, pid) do
      [{_, module}] -> module
      _ -> Phoenix.SessionProcess.Config.session_process()
    end
  end

  defp validate_session_id(session_id) do
    if Phoenix.SessionProcess.Config.valid_session_id?(session_id) do
      :ok
    else
      {:error, :invalid_session_id}
    end
  end

  defp check_session_limits() do
    max_sessions = Phoenix.SessionProcess.Config.max_sessions()
    current_sessions = Registry.count(Phoenix.SessionProcess.Registry)

    if current_sessions < max_sessions do
      :ok
    else
      {:error, :session_limit_reached}
    end
  end
end