tribes-plugin-kobold/scripts/kobold_e2e_runner.exs

defmodule TribeOne.TribesPlugin.Kobold.E2ERunner do
  alias Parrhesia.API.Auth

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

  @origin "http://127.0.0.1:46201"
  @edge "http://127.0.0.1:46202"

  def main(_argv) do
    :ok = ensure_http_client_started()

    assert_ok(wait_http(@origin <> "/sign-in"), "origin node ready")
    assert_ok(wait_http(@edge <> "/sign-in"), "edge node ready")

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

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

    assert_plugin_loaded!(@origin, "origin", "tribe-one-kobold")
    assert_plugin_loaded!(@edge, "edge", "tribe-one-kobold")
    assert_plugin_loaded!(@origin, "origin", "tribe-one-trust")
    assert_plugin_loaded!(@edge, "edge", "tribe-one-trust")
    assert_kobold_schema_ready!(@origin, "origin")
    assert_kobold_schema_ready!(@edge, "edge")

    :ok = connect_cluster_pair(@origin, origin_info, @edge, edge_info)
    edge_pubkey = node_pubkey(edge_info)
    :ok = establish_unknown_tribe_trust(@origin, edge_pubkey)

    run_id = "kobold-docker-e2e-#{System.system_time(:millisecond)}"
    {:ok, _} = kobold(@origin, "kobold.reset", %{"run_id" => run_id})
    {:ok, _} = kobold(@edge, "kobold.reset", %{"run_id" => run_id})

    {:ok, %{"dataset" => public_dataset}} =
      kobold(@origin, "kobold.datasets.create", %{
        "run_id" => run_id,
        "name" => "Public seed catalog",
        "visibility" => "public"
      })

    {:ok, %{"dataset" => private_dataset}} =
      kobold(@origin, "kobold.datasets.create", %{
        "run_id" => run_id,
        "name" => "Private seed notes",
        "visibility" => "private"
      })

    :ok = create_seed_resource(@origin, public_dataset["id"])
    :ok = create_seed_resource(@origin, private_dataset["id"])

    {:ok, %{"record" => public_record}} =
      kobold(@origin, "kobold.records.upsert", %{
        "dataset_id" => public_dataset["id"],
        "resource_name" => "SeedVariety",
        "fields" => %{"name" => "Public tomato", "germination_days" => 8}
      })

    {:ok, %{"record" => private_record}} =
      kobold(@origin, "kobold.records.upsert", %{
        "dataset_id" => private_dataset["id"],
        "resource_name" => "SeedVariety",
        "fields" => %{"name" => "Private pepper", "germination_days" => 14}
      })

    assert_remote_cannot_read_or_write_private!(run_id, private_dataset["id"], edge_pubkey)
    :ok = grant_dataset_access(@origin, private_dataset["id"], edge_pubkey, "read")
    :ok = grant_dataset_access(@origin, private_dataset["id"], edge_pubkey, "write")

    {:ok, %{"record" => private_remote_record}} =
      kobold(@origin, "kobold.records.upsert", %{
        "dataset_id" => private_dataset["id"],
        "resource_name" => "SeedVariety",
        "remote_tribe_pubkey" => edge_pubkey,
        "fields" => %{"name" => "Remote-written private bean", "germination_days" => 10}
      })

    assert_remote_can_read_private!(
      run_id,
      private_dataset["id"],
      private_remote_record["record_id"],
      edge_pubkey
    )

    assert_origin_state!(run_id, public_record["record_id"], private_record["record_id"])

    assert_ok(
      wait_for_edge_public_private_state(
        run_id,
        public_dataset["id"],
        private_dataset["id"],
        public_record["record_id"],
        private_record["record_id"]
      ) == :ok,
      "edge receives public and private Kobold datasets through cluster sync",
      &dump_debug/0
    )

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

  defp create_seed_resource(base_url, dataset_id) do
    with {:ok, _resource} <-
           kobold(base_url, "kobold.resources.create", %{
             "dataset_id" => dataset_id,
             "name" => "SeedVariety",
             "fields" => %{
               "name" => %{"type" => "string", "required" => true},
               "germination_days" => %{"type" => "integer"}
             }
           }) do
      :ok
    end
  end

  defp assert_remote_cannot_read_or_write_private!(run_id, private_dataset_id, edge_pubkey) do
    {:ok, state} =
      kobold(@origin, "kobold.state", %{
        "run_id" => run_id,
        "remote_tribe_pubkey" => edge_pubkey
      })

    assert_ok(
      Enum.all?(Map.get(state, "datasets", []), &(&1["id"] != private_dataset_id)),
      "remote tribe cannot read private dataset without permission"
    )

    assert_ok(
      match?(
        {:error, %{body: %{"error" => error}}} when is_binary(error),
        kobold(@origin, "kobold.records.upsert", %{
          "dataset_id" => private_dataset_id,
          "resource_name" => "SeedVariety",
          "remote_tribe_pubkey" => edge_pubkey,
          "fields" => %{"name" => "Denied private bean", "germination_days" => 11}
        })
      ),
      "remote tribe cannot write private dataset without permission"
    )
  end

  defp assert_remote_can_read_private!(run_id, private_dataset_id, record_id, edge_pubkey) do
    {:ok, state} =
      kobold(@origin, "kobold.state", %{
        "run_id" => run_id,
        "remote_tribe_pubkey" => edge_pubkey
      })

    assert_ok(
      Enum.any?(Map.get(state, "datasets", []), &(&1["id"] == private_dataset_id)) and
        Enum.any?(Map.get(state, "records", []), &(&1["record_id"] == record_id)),
      "remote tribe can read explicitly shared private dataset"
    )
  end

  defp assert_origin_state!(run_id, public_record_id, private_record_id) do
    {:ok, state} = kobold(@origin, "kobold.state", %{"run_id" => run_id})
    records = Map.get(state, "records", [])

    assert_ok(
      Enum.any?(records, &(&1["record_id"] == public_record_id)),
      "origin has public record"
    )

    assert_ok(
      Enum.any?(records, &(&1["record_id"] == private_record_id)),
      "origin has private record"
    )
  end

  defp wait_for_edge_public_private_state(
         run_id,
         public_dataset_id,
         private_dataset_id,
         public_record_id,
         private_record_id
       ) do
    Enum.reduce_while(1..90, {:error, :timeout}, fn _attempt, last ->
      case kobold(@edge, "kobold.state", %{"run_id" => run_id}) do
        {:ok, %{"datasets" => datasets, "records" => records}} ->
          has_public_dataset? = Enum.any?(datasets, &(&1["id"] == public_dataset_id))
          has_private_dataset? = Enum.any?(datasets, &(&1["id"] == private_dataset_id))
          has_public_record? = Enum.any?(records, &(&1["record_id"] == public_record_id))
          has_private_record? = Enum.any?(records, &(&1["record_id"] == private_record_id))

          if has_public_dataset? and has_private_dataset? and has_public_record? and
               has_private_record? do
            {:halt, :ok}
          else
            Process.sleep(1_000)
            {:cont, {:error, %{datasets: datasets, records: records}}}
          end

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

  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 assert_kobold_schema_ready!(base_url, label) do
    {:ok, %{"schema" => %{"tables" => tables}}} = kobold(base_url, "kobold.schema", %{})

    Enum.each(
      ["datasets", "resource_definitions", "commits", "bookmarks", "record_projections"],
      fn table ->
        assert_ok(tables[table] == true, "#{label} kobold table #{table} ready")
      end
    )
  end

  defp establish_unknown_tribe_trust(base_url, edge_pubkey) do
    with {:ok, _relationship} <-
           trust(base_url, "trust.hello.receive", %{
             "from_tribe" => edge_pubkey,
             "name" => "Kobold edge",
             "api_url" => @edge <> "/api/tribes/v1",
             "relay_urls" => ["ws://kobold-edge:4000/nostr/relay"],
             "capabilities" => ["org.tribes.kobold.dataset@1"],
             "requested_capabilities" => ["org.tribes.kobold.dataset.read@1"]
           }) do
      :ok
    end
  end

  defp grant_dataset_access(base_url, dataset_id, subject_id, action) do
    with {:ok, _rule} <-
           kobold(base_url, "kobold.access.rules.create", %{
             "resource_type" => "kobold.dataset",
             "resource_id" => dataset_id,
             "action" => action,
             "subject_type" => "tribe",
             "subject_id" => subject_id,
             "effect" => "allow",
             "condition" => "none"
           }) do
      :ok
    end
  end

  defp kobold(base_url, method, params) do
    admin(base_url, "plugin.call", %{
      "plugin" => "tribe-one-kobold",
      "method" => method,
      "version" => "1",
      "params" => params
    })
  end

  defp trust(base_url, method, params) do
    admin(base_url, "plugin.call", %{
      "plugin" => "tribe-one-trust",
      "method" => method,
      "version" => "1",
      "params" => params
    })
  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) do
    admin(base_url, "cluster_nodes.upsert", %{
      "pubkey" => node_pubkey(node_info),
      "transport_address" => node_info["sync_url"],
      "scope" => "all",
      "status" => "active"
    })
  end

  defp node_pubkey(%{"node_pubkey" => pubkey}) when is_binary(pubkey) and pubkey != "", do: pubkey
  defp node_pubkey(%{"pubkey" => pubkey}) when is_binary(pubkey) and pubkey != "", do: pubkey

  defp admin(base_url, method, params) do
    url = base_url <> "/api/admin/management"
    body = JSON.encode!(%{"method" => method, "params" => params})

    headers = [
      {~c"content-type", ~c"application/json"},
      {~c"authorization", to_charlist(nip98_authorization("POST", url))}
    ]

    case :httpc.request(:post, {to_charlist(url), headers, ~c"application/json", body}, [],
           body_format: :binary
         ) do
      {:ok, {{_version, status, _reason}, _headers, response_body}} ->
        decoded = decode_json(response_body)

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

      {:error, reason} ->
        {:error, reason}
    end
  end

  defp admin_with_retry(base_url, method, params, opts) do
    attempts = Keyword.get(opts, :attempts, 10)
    delay_ms = Keyword.get(opts, :delay_ms, 500)

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

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

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

  defp http_get(url) do
    case :httpc.request(:get, {to_charlist(url), [{~c"x-forwarded-proto", ~c"https"}]}, [],
           body_format: :binary
         ) do
      {:ok, {{_version, status, _reason}, _headers, body}} -> {:ok, status, body}
      {:error, reason} -> {:error, reason}
    end
  end

  defp wait_http(url) do
    Enum.reduce_while(1..90, {:error, :timeout}, fn _attempt, _acc ->
      case http_get(url) do
        {:ok, status, _body} when status in 200..499 ->
          {:halt, :ok}

        other ->
          Process.sleep(1_000)
          {:cont, other}
      end
    end)
  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" => "kobold-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 decode_json(body) when is_binary(body) do
    case JSON.decode(body) do
      {:ok, decoded} -> decoded
      {:error, _reason} -> %{"raw" => body}
    end
  end

  defp ensure_http_client_started do
    case :inets.start() do
      :ok -> :ok
      {:error, {:already_started, :inets}} -> :ok
      {:error, reason} -> {:error, reason}
    end
  end

  defp assert_ok(:ok, message), do: assert_ok(true, message)
  defp assert_ok(true, message), do: IO.puts("✅ #{message}")
  defp assert_ok(other, message), do: assert_ok(other, message, nil)
  defp assert_ok(true, message, _debug), do: assert_ok(true, message)

  defp assert_ok(other, message, debug) do
    if debug do
      debug.()
    end

    raise "assertion failed: #{message}; got #{inspect(other, limit: :infinity)}"
  end

  defp dump_debug do
    IO.puts(:stderr, "==== kobold e2e debug dump ====")

    Enum.each([origin: @origin, edge: @edge], fn {name, base_url} ->
      IO.puts(:stderr, "-- #{name} #{base_url} --")

      IO.puts(
        :stderr,
        "node_info: #{inspect(admin(base_url, "node_info", %{}), limit: :infinity)}"
      )

      IO.puts(
        :stderr,
        "plugin_list: #{inspect(admin(base_url, "plugin_list", %{}), limit: :infinity)}"
      )

      IO.puts(
        :stderr,
        "kobold.schema: #{inspect(kobold(base_url, "kobold.schema", %{}), limit: :infinity)}"
      )
    end)
  end
end

TribeOne.TribesPlugin.Kobold.E2ERunner.main(System.argv())