phoenix_sync

defmodule Phoenix.Sync.ShapeRequestRegistry do
  @moduledoc false

  use GenServer

  alias Phoenix.Sync.PredefinedShape

  # Client API

  def start_link(opts \\ []) do
    GenServer.start_link(__MODULE__, opts, name: __MODULE__)
  end

  def register_shape(%PredefinedShape{} = shape) do
    GenServer.call(__MODULE__, {:register_shape, shape})
  end

  def unregister_shape(key) do
    GenServer.call(__MODULE__, {:unregister_shape, key})
  end

  def interrupt_matching(shape, shape_opts \\ [])

  def interrupt_matching(match_spec, _opts) when is_list(match_spec) do
    match_spec = Keyword.replace_lazy(match_spec, :params, &normalize_params_form/1)

    do_interrupt_matching(fn shape ->
      Enum.all?(match_spec, fn {k, v} -> Map.get(shape, k) == v end)
    end)
  end

  def interrupt_matching(matcher, _opts) when is_function(matcher, 1) do
    GenServer.call(__MODULE__, {:interrupt_matching, matcher})
  end

  def interrupt_matching(queryable, shape_opts)
      when is_atom(queryable) or is_struct(queryable, Ecto.Query) do
    match_shape =
      queryable
      |> PredefinedShape.new!(shape_opts)
      |> match_params()

    do_interrupt_matching(fn shape -> match_shape == shape end)
  rescue
    e -> {:error, e}
  end

  def interrupt_matching(table, opts) when is_binary(table) do
    interrupt_matching(Keyword.put(opts, :table, table))
  end

  defp do_interrupt_matching(fun) do
    GenServer.call(__MODULE__, {:interrupt_matching, fun})
  end

  defp normalize_params_form(params) when is_list(params) do
    params
    |> Enum.with_index(fn elem, index -> {to_string(index + 1), to_string(elem)} end)
    |> Map.new()
  end

  defp normalize_params_form(params) when is_map(params) do
    Map.new(params, fn {k, v} -> {to_string(k), to_string(v)} end)
  end

  def registered_requests do
    GenServer.call(__MODULE__, :registered_requests)
  end

  # Server implementation

  @impl GenServer
  def init(_opts) do
    {:ok, %{subscriptions: %{}, monitors: %{}}}
  end

  @impl GenServer
  def handle_call({:register_shape, shape}, {request_pid, _ref}, state) do
    %{
      monitors: monitors,
      subscriptions: subscriptions
    } = state

    monitor_ref = Process.monitor(request_pid)
    monitors = Map.put(monitors, monitor_ref, monitor_ref)

    shape_info = {shape, request_pid}
    subscriptions = Map.put(subscriptions, monitor_ref, shape_info)

    {:reply, {:ok, monitor_ref}, %{state | monitors: monitors, subscriptions: subscriptions}}
  end

  def handle_call({:unregister_shape, key}, _from, %{subscriptions: subscriptions} = state) do
    {:reply, :ok, %{state | subscriptions: Map.delete(subscriptions, key)}}
  end

  def handle_call({:interrupt_matching, matcher}, _from, state) do
    interrupted_count =
      state.subscriptions
      |> Enum.filter(fn {_id, {shape, _request_pid}} ->
        shape
        |> match_params()
        |> matcher.()
      end)
      |> Enum.reduce(0, fn {key, {_shape, request_pid}}, acc ->
        send(request_pid, {:interrupt_shape, key, :server_interrupt})

        acc + 1
      end)

    {:reply, {:ok, interrupted_count}, state}
  end

  def handle_call(:registered_requests, _from, %{subscriptions: subscriptions} = state) do
    {:reply, Map.to_list(subscriptions), state}
  end

  @impl GenServer
  def handle_info({:DOWN, monitor_ref, :process, _pid, _reason}, state) do
    case Map.pop(state.monitors, monitor_ref) do
      {nil, monitors} ->
        {:noreply, %{state | monitors: monitors}}

      {key, monitors} ->
        subscriptions = Map.delete(state.subscriptions, key)

        {:noreply, %{state | subscriptions: subscriptions, monitors: monitors}}
    end
  end

  defp match_params(%PredefinedShape{} = shape) do
    shape
    |> PredefinedShape.to_shape_params()
    |> Map.new()
    |> Map.drop([:replica])
  end
end