tribes-plugin-supertest/lib/supertest/cluster_pubsub_probe.ex

defmodule TribeOne.TribesPlugin.Supertest.ClusterPubSubProbe do
  @moduledoc false

  use GenServer

  @topic "supertest.cluster_pubsub_probe"

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

  def broadcast(run_id, payload) when is_binary(run_id) and run_id != "" and is_map(payload) do
    GenServer.call(__MODULE__, {:broadcast, run_id, payload})
  catch
    :exit, _reason -> {:error, :cluster_pubsub_probe_unavailable}
  end

  def received(run_id) when is_binary(run_id) and run_id != "" do
    GenServer.call(__MODULE__, {:received, run_id})
  catch
    :exit, _reason -> {:error, :cluster_pubsub_probe_unavailable}
  end

  def reset(run_id) when is_binary(run_id) and run_id != "" do
    GenServer.call(__MODULE__, {:reset, run_id})
  catch
    :exit, _reason -> {:error, :cluster_pubsub_probe_unavailable}
  end

  @impl true
  def init(opts) do
    pubsub = Keyword.fetch!(opts, :pubsub)
    :ok = Phoenix.PubSub.subscribe(pubsub, @topic)

    {:ok, %{pubsub: pubsub, received: %{}}}
  end

  @impl true
  def handle_call({:broadcast, run_id, payload}, _from, state) do
    message = %{
      id: Ash.UUID.generate(),
      run_id: run_id,
      payload: payload,
      origin_node: Atom.to_string(node()),
      sent_at_usec: System.system_time(:microsecond)
    }

    result =
      Phoenix.PubSub.broadcast(
        state.pubsub,
        @topic,
        {:supertest_cluster_pubsub_probe, message}
      )

    {:reply, result_to_reply(result, message), state}
  end

  def handle_call({:received, run_id}, _from, state) do
    messages =
      state.received
      |> Map.get(run_id, [])
      |> Enum.reverse()

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

  def handle_call({:reset, run_id}, _from, state) do
    {:reply, :ok, %{state | received: Map.delete(state.received, run_id)}}
  end

  @impl true
  def handle_info({:supertest_cluster_pubsub_probe, %{run_id: run_id} = message}, state)
      when is_binary(run_id) do
    received = Map.update(state.received, run_id, [message], &[message | &1])
    {:noreply, %{state | received: received}}
  end

  def handle_info(_message, state), do: {:noreply, state}

  defp result_to_reply(:ok, message), do: {:ok, message}
  defp result_to_reply({:error, reason}, _message), do: {:error, reason}
end