#!/bin/sh
set -eu

ROOT=$(CDPATH= cd -- "$(dirname -- "$0")/.." && pwd)
MODE=${1:-compose}
COMPOSE_PROJECT_NAME=${COMPOSE_PROJECT_NAME:-who_need_help}

case "$MODE" in
  compose)
    TARGET=$(docker compose -p "$COMPOSE_PROJECT_NAME" ps -q web | head -n 1)
    if [ -z "$TARGET" ]; then
      echo "No running Compose web replica was found." >&2
      exit 1
    fi
    RUN="docker exec $TARGET"
    ;;
  kind)
    PATH="$ROOT/.tools/bin:$PATH"
    export PATH
    TARGET=$(
      kubectl --context kind-who-need-help --namespace who-need-help get pods \
        -l app.kubernetes.io/component=web \
        --field-selector=status.phase=Running \
        --no-headers \
        -o custom-columns='NAME:.metadata.name,READY:.status.containerStatuses[0].ready,DELETING:.metadata.deletionTimestamp' |
        awk '$2 == "true" && $3 == "<none>" {print $1; exit}'
    )
    if [ -z "$TARGET" ]; then
      echo "No running kind web replica was found." >&2
      exit 1
    fi
    RUN="kubectl --context kind-who-need-help --namespace who-need-help exec $TARGET --"
    ;;
  *)
    echo "Usage: $0 [compose|kind]" >&2
    exit 1
    ;;
esac

# The subscriber runs on the selected web node. The broadcast is executed by a
# different connected BEAM node. Success proves distributed PubSub fan-out.
$RUN /app/bin/who_need_help rpc '
  live_peers =
    Enum.filter(Node.list(), fn peer ->
      case :rpc.call(peer, Process, :whereis, [WhoNeedHelp.PubSub]) do
        pid when is_pid(pid) -> true
        _ -> false
      end
    end)

  case live_peers do
    [] ->
      exit({:no_live_cluster_peers, Node.list()})

    peers ->
      topic = "cluster:verify:" <> Integer.to_string(System.unique_integer([:positive]))
      :ok = Phoenix.PubSub.subscribe(WhoNeedHelp.PubSub, topic)

      result =
        Enum.reduce_while(peers, nil, fn peer, _result ->
          case :rpc.call(peer, Phoenix.PubSub, :broadcast, [
                 WhoNeedHelp.PubSub,
                 topic,
                 {:cross_replica_probe, peer}
               ]) do
            :ok ->
              receive do
                {:cross_replica_probe, ^peer} ->
                  {:halt,
                   %{
                     status: :ok,
                     subscriber: node(),
                     broadcaster: peer,
                     peers: Node.list()
                   }}
              after
                5_000 -> {:cont, nil}
              end

            _error ->
              {:cont, nil}
          end
        end)

      result =
        result ||
          %{status: :timeout, subscriber: node(), attempted_peers: peers, peers: Node.list()}

      IO.inspect(result)
      if result.status != :ok, do: exit({:cross_replica_pubsub_failed, result})
  end
'