tribes-plugin-supertest/scripts/supertest_e2e_runner.exs

defmodule Supertest.E2ERunner do
  alias Parrhesia.API.Auth

  @admin_pubkey "985826ac4ec99f6304785ebfaa303ec42973653a682cfa0a06caf9fec662eaa7"
  @admin_privkey Base.decode16!(
                   "7ee088208557dcc06405ccf6547115b62a597c5d107ee6044bb430e71a68655c",
                   case: :lower
                 )

  @origin "http://127.0.0.1:46201"
  @peer "http://127.0.0.1:46202"
  @origin_vm "http://127.0.0.1:48101"
  @peer_vm "http://127.0.0.1:48102"

  def main(_argv) do
    {:ok, _} = Application.ensure_all_started(:req)

    assert_ok(wait_victoria_metrics(@origin_vm), "origin VictoriaMetrics ready")
    assert_ok(wait_victoria_metrics(@peer_vm), "peer VictoriaMetrics ready")
    assert_ok(wait_http(@origin <> "/sign-in"), "origin node ready")
    assert_ok(wait_http(@peer <> "/sign-in"), "peer node ready")

    {:ok, origin_info} =
      admin_with_retry(@origin, "node_info", %{}, attempts: 90, delay_ms: 1_000)

    {:ok, peer_info} = admin_with_retry(@peer, "node_info", %{}, attempts: 90, delay_ms: 1_000)

    assert_plugin_loaded!(@origin, "origin", "tribes_ui")
    assert_plugin_loaded!(@peer, "peer", "tribes_ui")
    assert_plugin_loaded!(@origin, "origin", "supertest")
    assert_plugin_loaded!(@peer, "peer", "supertest")

    {:ok, %{"plugin" => "supertest", "ok" => true}} = supertest_get("supertest-origin", "/health")
    {:ok, %{"plugin" => "supertest", "ok" => true}} = supertest_get("supertest-peer", "/health")

    metrics_probe_value = 100_000 + System.unique_integer([:positive, :monotonic])
    emit_supertest_metrics_probe!(metrics_probe_value)

    {:ok, %{"schema" => schema}} = supertest_get("supertest-origin", "/schema")
    assert_ok(schema["tables"]["cases"] == true, "supertest cases table exists")
    assert_ok(schema["tables"]["events"] == true, "supertest events table exists")

    :ok = connect_cluster_pair(@origin, origin_info, @peer, peer_info)

    {:ok, %{"ok" => true}} =
      admin(@origin, "plugin.call", %{
        "plugin" => "supertest",
        "method" => "supertest.echo",
        "version" => "1",
        "params" => %{"phase" => "docker-e2e"}
      })

    run_id = "supertest-e2e-#{System.unique_integer([:positive, :monotonic])}"
    reset_supertest_state!(run_id)

    {:ok, %{"case" => %{"id" => case_id}}} =
      supertest_post("supertest-origin", "/cases", %{
        "run_id" => run_id,
        "name" => "origin-created",
        "counter" => 1,
        "status" => "active",
        "payload" => %{"source" => "origin"}
      })

    {:ok, %{"event" => %{"id" => event_id}}} =
      supertest_post("supertest-origin", "/events", %{
        "case_id" => case_id,
        "run_id" => run_id,
        "sequence" => 1,
        "body" => "origin event",
        "payload" => %{"source" => "origin"}
      })

    assert_ok(
      wait_for_supertest_state("supertest-peer", run_id, fn state ->
        has_case?(state, case_id, 1) and has_event?(state, event_id)
      end) == :ok,
      "supertest resources sync origin -> peer"
    )

    assert_ok(run_cluster_pubsub_probe(run_id) == :ok, "supertest cluster PubSub probe")

    assert_ok(
      wait_for_supertest_metric_rollup(@origin, metrics_probe_value) == :ok,
      "supertest plugin metric appears in rollup"
    )

    IO.puts("supertest e2e assertions passed")
  end

  defp emit_supertest_metrics_probe!(value) do
    {:ok, %{"ok" => true, "value" => ^value}} =
      supertest_post("supertest-origin", "/metrics/probe", %{"value" => value})

    :ok
  end

  defp reset_supertest_state!(run_id) do
    {:ok, _} = supertest_post("supertest-origin", "/reset", %{"run_id" => run_id})
    {:ok, _} = supertest_post("supertest-peer", "/reset", %{"run_id" => run_id})
    :ok
  end

  defp run_cluster_pubsub_probe(run_id) do
    pubsub_run_id = run_id <> "-pubsub"

    {:ok, _} =
      supertest_post("supertest-origin", "/cluster-pubsub/reset", %{"run_id" => pubsub_run_id})

    {:ok, _} =
      supertest_post("supertest-peer", "/cluster-pubsub/reset", %{"run_id" => pubsub_run_id})

    {:ok, %{"message" => %{"id" => message_id}}} =
      supertest_post("supertest-origin", "/cluster-pubsub/broadcast", %{
        "run_id" => pubsub_run_id,
        "payload" => %{"source" => "origin"}
      })

    wait_for_cluster_pubsub_message(pubsub_run_id, message_id)
  end

  defp wait_for_cluster_pubsub_message(run_id, message_id) do
    Enum.reduce_while(1..90, {:error, :timeout}, fn _attempt, _acc ->
      states = [
        supertest_get("supertest-origin", "/cluster-pubsub/received?run_id=#{run_id}"),
        supertest_get("supertest-peer", "/cluster-pubsub/received?run_id=#{run_id}")
      ]

      if Enum.all?(states, &cluster_pubsub_message_seen?(&1, message_id)) do
        {:halt, :ok}
      else
        Process.sleep(1_000)
        {:cont, {:error, states}}
      end
    end)
  end

  defp cluster_pubsub_message_seen?({:ok, %{"messages" => messages}}, message_id)
       when is_list(messages) do
    Enum.any?(messages, fn
      %{"id" => ^message_id} -> true
      _other -> false
    end)
  end

  defp cluster_pubsub_message_seen?(_state, _message_id), do: false

  defp wait_for_supertest_state(service, run_id, predicate) do
    Enum.reduce_while(1..90, {:error, :timeout}, fn _attempt, _acc ->
      case supertest_get(service, "/state?run_id=#{run_id}") do
        {:ok, state} ->
          if predicate.(state) do
            {:halt, :ok}
          else
            Process.sleep(1_000)
            {:cont, {:error, state}}
          end

        other ->
          Process.sleep(1_000)
          {:cont, other}
      end
    end)
  end

  defp wait_for_supertest_metric_rollup(base_url, expected_value) do
    Enum.reduce_while(1..150, {:error, :timeout}, fn _attempt, _acc ->
      case admin(base_url, "metrics_rollups.list", %{"limit" => 120}) do
        {:ok, %{"rollups" => rollups}} ->
          if Enum.any?(rollups, &supertest_metric_rollup?(&1, expected_value)) do
            {:halt, :ok}
          else
            Process.sleep(1_000)
            {:cont, {:error, rollups}}
          end

        other ->
          Process.sleep(1_000)
          {:cont, other}
      end
    end)
  end

  defp supertest_metric_rollup?(%{"plugins" => plugins}, expected_value) when is_map(plugins) do
    plugins
    |> get_in(["supertest", "probe_value", "last"])
    |> numeric_equal?(expected_value)
  end

  defp supertest_metric_rollup?(_rollup, _expected_value), do: false

  defp numeric_equal?(value, expected) when is_integer(value) or is_float(value) do
    value == expected
  end

  defp numeric_equal?(value, expected) when is_binary(value) do
    case Float.parse(value) do
      {parsed, ""} -> parsed == expected
      _other -> false
    end
  end

  defp numeric_equal?(_value, _expected), do: false

  defp has_case?(%{"cases" => cases}, case_id, counter) when is_list(cases) do
    Enum.any?(cases, fn
      %{"id" => ^case_id, "counter" => ^counter} -> true
      _other -> false
    end)
  end

  defp has_case?(_state, _case_id, _counter), do: false

  defp has_event?(%{"events" => events}, event_id) when is_list(events) do
    Enum.any?(events, fn
      %{"id" => ^event_id} -> true
      _other -> false
    end)
  end

  defp has_event?(_state, _event_id), do: false

  defp assert_plugin_loaded!(base_url, label, plugin_name) do
    {:ok, plugins} = admin_with_retry(base_url, "plugin_list", %{}, attempts: 30, delay_ms: 1_000)

    loaded? =
      Enum.any?(plugins["plugins"], fn
        %{"name" => ^plugin_name, "status" => "loaded"} -> true
        _other -> false
      end)

    assert_ok(loaded?, "#{label} #{plugin_name} plugin loaded")
  end

  defp connect_cluster_pair(left_base_url, left_node_info, right_base_url, right_node_info) do
    with {:ok, _} <- upsert_cluster_node(left_base_url, right_node_info),
         {:ok, _} <- upsert_cluster_node(right_base_url, left_node_info) do
      :ok
    else
      {:error, reason} ->
        assert_ok(false, "unable to connect cluster pair: #{inspect(reason)}")
    end
  end

  defp upsert_cluster_node(base_url, node_info) when is_map(node_info) do
    admin(base_url, "cluster_nodes.upsert", %{
      "pubkey" => node_info["node_pubkey"],
      "transport_address" => node_info["sync_url"],
      "scope" => "all",
      "status" => "active"
    })
  end

  defp supertest_get(service, path), do: supertest_api(service, "GET", path, nil)

  defp supertest_post(service, path, payload), do: supertest_api(service, "POST", path, payload)

  defp supertest_api(service, method, path, payload) do
    url = "http://127.0.0.1:4000/plugins-api/supertest" <> path

    args = [
      "compose",
      "-f",
      compose_file(),
      "exec",
      "-T",
      service,
      "curl",
      "-sS",
      "-f",
      "-X",
      method
    ]

    args =
      if is_nil(payload) do
        args ++ [url]
      else
        args ++ ["-H", "content-type: application/json", "-d", JSON.encode!(payload), url]
      end

    case System.cmd("docker", args, env: compose_env(), stderr_to_stdout: true) do
      {output, 0} ->
        {:ok, JSON.decode!(output)}

      {output, status} ->
        {:error, %{status: status, output: output}}
    end
  end

  defp admin(base_url, method, params) do
    url = base_url <> "/api/admin/management"

    response =
      Req.post!(url,
        headers: [
          {"content-type", "application/json"},
          {"authorization", nip98_authorization("POST", url)}
        ],
        json: %{"method" => method, "params" => params},
        receive_timeout: 15_000
      )

    body = response.body

    if response.status == 200 and body["ok"] == true do
      {:ok, body["result"]}
    else
      {:error, %{status: response.status, body: body}}
    end
  end

  defp admin_with_retry(base_url, method, params, opts) when is_list(opts) do
    attempts = Keyword.get(opts, :attempts, 10)
    delay_ms = Keyword.get(opts, :delay_ms, 500)
    retry_statuses = Keyword.get(opts, :retry_statuses, [403, 500, 502, 503, 504])

    Enum.reduce_while(1..attempts, {:error, :retry_exhausted}, fn attempt, _acc ->
      case admin_safe(base_url, method, params) do
        {:ok, _result} = ok ->
          {:halt, ok}

        {:error, %{status: status}} = error ->
          if attempt < attempts and status in retry_statuses do
            Process.sleep(delay_ms)
            {:cont, error}
          else
            {:halt, error}
          end

        {:error, %{exception: _reason}} = error when attempt < attempts ->
          Process.sleep(delay_ms)
          {:cont, error}

        other ->
          {:halt, other}
      end
    end)
  end

  defp admin_safe(base_url, method, params) do
    admin(base_url, method, params)
  rescue
    error ->
      {:error, %{exception: Exception.message(error)}}
  end

  defp nip98_authorization(method, url) do
    base = %{
      "pubkey" => @admin_pubkey,
      "created_at" => System.system_time(:second),
      "kind" => 27_235,
      "tags" => [["method", method], ["u", url]],
      "content" => "supertest-e2e-#{System.unique_integer([:positive, :monotonic])}"
    }

    event_id = Auth.compute_event_id(base)
    {:ok, sig} = Tribes.Keyring.sign_event(Base.decode16!(event_id, case: :lower), @admin_privkey)
    event = Map.merge(base, %{"id" => event_id, "sig" => sig})

    "Nostr " <> Base.encode64(JSON.encode!(event))
  end

  defp wait_http(url) do
    Enum.reduce_while(1..60, {:error, :timeout}, fn _, _acc ->
      case Req.get(url, receive_timeout: 2_000, retry: false) do
        {:ok, %{status: status}} when status in 200..499 ->
          {:halt, :ok}

        _ ->
          Process.sleep(1000)
          {:cont, {:error, :retry}}
      end
    end)
  end

  defp wait_victoria_metrics(base_url) do
    Enum.reduce_while(1..60, {:error, :timeout}, fn _, _acc ->
      case Req.get(base_url <> "/prometheus/api/v1/query",
             params: [query: "vector(1)"],
             receive_timeout: 2_000,
             retry: false
           ) do
        {:ok,
         %{
           status: 200,
           body: %{
             "status" => "success",
             "data" => %{"result" => [%{"value" => [_timestamp, "1"]}]}
           }
         }} ->
          {:halt, :ok}

        other ->
          Process.sleep(1000)
          {:cont, other}
      end
    end)
  end

  defp compose_file, do: System.get_env("E2E_COMPOSE_FILE", "test/e2e/compose.supertest.yaml")

  defp compose_env do
    case System.get_env("COMPOSE_PROJECT_NAME") do
      nil -> []
      value -> [{"COMPOSE_PROJECT_NAME", value}]
    end
  end

  defp assert_ok(true, _message), do: :ok
  defp assert_ok(:ok, _message), do: :ok

  defp assert_ok(false, message) do
    IO.puts(:stderr, "assertion failed: #{message}")
    System.halt(1)
  end

  defp assert_ok(other, message) do
    IO.puts(:stderr, "assertion failed: #{message} (got: #{inspect(other)})")
    System.halt(1)
  end
end

Supertest.E2ERunner.main(System.argv())