#!/usr/bin/env bash
# next/tests/discovery_fetch.sh — m2 Step 10c acceptance test.
#
# Two halves:
#   (a) http_server side: the new actor_doc Accept format negotiates
#       to a term_codec-encoded peer-actor-state proplist served
#       from `nx_kernel:actor_state/1`. Verified via http_server:route
#       in-process.
#   (b) discovery_fetch closure: builds the FetchFn that
#       peer_actors:lookup_or_fetch_srv/2 expects, GETs the actor
#       doc via httpc:request/4, decodes the body, returns the AS
#       proplist. Verified end-to-end against a background
#       `python3 -m http.server`-style stub that returns hand-crafted
#       term_codec bytes (so we exercise the wire, not just the
#       in-process route).

set -uo pipefail
cd "$(git rev-parse --show-toplevel)"

SX_SERVER="${SX_SERVER:-hosts/ocaml/_build/default/bin/sx_server.exe}"
if [ ! -x "$SX_SERVER" ]; then
  SX_SERVER="/root/rose-ash/hosts/ocaml/_build/default/bin/sx_server.exe"
fi
if [ ! -x "$SX_SERVER" ]; then
  echo "ERROR: sx_server.exe not found." >&2
  exit 1
fi

VERBOSE="${1:-}"
PASS=0; FAIL=0; ERRORS=""

# ── live stub server ─────────────────────────────────────────
# Python script that:
#   GET /actors/alice    -> 200 with term_codec-encoded AS
#                          (built in Python: matches term_codec
#                          netstring format spelled out in
#                          next/kernel/term_codec.erl).
#   GET /actors/missing  -> 404
PORT=$(python3 -c 'import socket;s=socket.socket();s.bind(("127.0.0.1",0));print(s.getsockname()[1]);s.close()')
SRVROOT=$(mktemp -d)
PYSRV="$SRVROOT/srv.py"
cat > "$PYSRV" <<'PY'
import sys, http.server, socketserver

PORT = int(sys.argv[1])

# term_codec encoding (mirror of next/kernel/term_codec.erl).
def enc_atom(s):
    b = s.encode()
    return f"a{len(b)}:".encode() + b
def enc_int(n):
    s = str(n).encode()
    return f"i{len(s)}:".encode() + s
def enc_bin(b):
    return f"b{len(b)}:".encode() + b
def enc_list(items):
    payload = b"".join(items)
    # term_codec uses ELEMENT COUNT (not byte length) for list/tuple
    # headers — see encode/1 in next/kernel/term_codec.erl.
    return f"l{len(items)}:".encode() + payload
def enc_tuple(items):
    payload = b"".join(items)
    return f"t{len(items)}:".encode() + payload
def enc_nil():
    return b"l0:"

# {public_keys, [[{id, k1}, {created, 0}, {value, <<1,2,3,4>>}]]}
KEY = enc_list([
    enc_tuple([enc_atom("id"), enc_atom("k1")]),
    enc_tuple([enc_atom("created"), enc_int(0)]),
    enc_tuple([enc_atom("value"), enc_bin(bytes([1,2,3,4]))]),
])
PROPLIST = enc_list([
    enc_tuple([enc_atom("public_keys"), enc_list([KEY])]),
])

class H(http.server.BaseHTTPRequestHandler):
    def do_GET(self):
        if self.path == "/actors/alice":
            self.send_response(200)
            self.send_header('content-type','application/vnd.fed-sx.actor-doc')
            self.send_header('content-length', str(len(PROPLIST)))
            self.end_headers()
            self.wfile.write(PROPLIST)
        else:
            self.send_response(404); self.end_headers(); self.wfile.write(b'not found')
    def log_message(self, fmt, *args): pass

with socketserver.TCPServer(("127.0.0.1", PORT), H) as srv:
    srv.serve_forever()
PY
python3 "$PYSRV" "$PORT" >/dev/null 2>&1 &
SRV_PID=$!
TMPFILE=$(mktemp)
trap "rm -rf $SRVROOT $TMPFILE; kill $SRV_PID 2>/dev/null || true" EXIT
for _ in 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15; do
  if curl -fsS "http://127.0.0.1:$PORT/actors/alice" >/dev/null 2>&1; then break; fi
  sleep 0.2
done

bytes_of() { python3 -c "import sys; print(','.join(str(b) for b in sys.argv[1].encode()))" "$1"; }
URL_BASE_BYTES=$(bytes_of "http://127.0.0.1:$PORT")

cat > "$TMPFILE" <<'EPOCHS'
(epoch 1)
(load "lib/erlang/tokenizer.sx")
(load "lib/erlang/parser.sx")
(load "lib/erlang/parser-core.sx")
(load "lib/erlang/parser-expr.sx")
(load "lib/erlang/parser-module.sx")
(load "lib/erlang/transpile.sx")
(load "lib/erlang/runtime.sx")
(load "lib/erlang/vm/dispatcher.sx")
(epoch 2)
(eval "(er-load-gen-server!)")
(epoch 3)
(eval "(get (erlang-load-module (file-read \"next/kernel/envelope.erl\")) :name)")
(epoch 4)
(eval "(get (erlang-load-module (file-read \"next/kernel/log.erl\")) :name)")
(epoch 5)
(eval "(get (erlang-load-module (file-read \"next/kernel/pipeline.erl\")) :name)")
(epoch 6)
(eval "(get (erlang-load-module (file-read \"next/kernel/term_codec.erl\")) :name)")
(epoch 7)
(eval "(get (erlang-load-module (file-read \"next/kernel/outbox.erl\")) :name)")
(epoch 8)
(eval "(get (erlang-load-module (file-read \"next/kernel/peer_actors.erl\")) :name)")
(epoch 9)
(eval "(get (erlang-load-module (file-read \"next/kernel/nx_kernel.erl\")) :name)")
(epoch 10)
(eval "(get (erlang-load-module (file-read \"next/kernel/http_server.erl\")) :name)")
(epoch 11)
(eval "(get (erlang-load-module (file-read \"next/kernel/dispatch_http.erl\")) :name)")
(epoch 12)
(eval "(get (erlang-load-module (file-read \"next/kernel/discovery_fetch.erl\")) :name)")

;; (a) http_server side: actor_doc Accept negotiates to actor_doc
(epoch 20)
(eval "(get (erlang-eval-ast \"http_server:accept_format(<<97,112,112,108,105,99,97,116,105,111,110,47,118,110,100,46,102,101,100,45,115,120,46,97,99,116,111,114,45,100,111,99>>) =:= actor_doc\") :name)")

;; (a) actor_doc_response_for/3 with kernel + actor returns 200 +
;; term_codec body; decoded body has :public_keys. Inline SETUP
;; per epoch because separate (eval ...) calls share gen_server
;; state but not Erlang locals, and we need fresh kernel-aware
;; assertions even though the previous epoch's nx_kernel persists.
(epoch 21)
(eval "(get (erlang-eval-ast \"AK = <<1,2,3,4>>, AKS = [{key_id,k1},{algorithm,ed25519},{value,AK}], AAS = [{public_keys,[[{id,k1},{created,0},{value,AK}]]}], nx_kernel:start_link(alice, AKS, AAS), Cfg = [{kernel, nx_kernel}], R = http_server:actor_doc_response_for(<<97,108,105,99,101>>, actor_doc, Cfg), {ok, S} = envelope:get_field(status, R), S =:= 200\") :name)")

;; (a) body decodes to a proplist with :public_keys
(epoch 22)
(eval "(get (erlang-eval-ast \"AK = <<1,2,3,4>>, AKS = [{key_id,k1},{algorithm,ed25519},{value,AK}], AAS = [{public_keys,[[{id,k1},{created,0},{value,AK}]]}], nx_kernel:start_link(alice, AKS, AAS), R = http_server:actor_doc_response_for(<<97,108,105,99,101>>, actor_doc, [{kernel, nx_kernel}]), {ok, Body} = envelope:get_field(body, R), {ok, AS, _} = term_codec:decode(Body), case envelope:get_field(public_keys, AS) of {ok, [_|_]} -> true; _ -> false end\") :name)")

;; (a) unknown actor -> 404
(epoch 23)
(eval "(get (erlang-eval-ast \"AK = <<1,2,3,4>>, AKS = [{key_id,k1},{algorithm,ed25519},{value,AK}], AAS = [{public_keys,[[{id,k1},{created,0},{value,AK}]]}], nx_kernel:start_link(alice, AKS, AAS), R = http_server:actor_doc_response_for(<<109,105,115,115,105,110,103>>, actor_doc, [{kernel, nx_kernel}]), {ok, S} = envelope:get_field(status, R), S =:= 404\") :name)")

;; (b) discovery_fetch:actor_doc_url builds <base>/actors/alice
(epoch 30)
(eval "(get (erlang-eval-ast \"U = discovery_fetch:actor_doc_url(<<__URL_BASE__>>, alice), U =:= <<__URL_BASE__,47,97,99,116,111,114,115,47,97,108,105,99,101>>\") :name)")

;; (b) discovery_fetch:fetch live -> {ok, AS} with :public_keys
(epoch 31)
(eval "(get (erlang-eval-ast \"R = discovery_fetch:fetch(<<__URL_BASE__,47,97,99,116,111,114,115,47,97,108,105,99,101>>, []), case R of {ok, AS} -> case envelope:get_field(public_keys, AS) of {ok, [_|_]} -> true; _ -> false end; _ -> false end\") :name)")

;; (b) closure produced by make_fetch_fn dispatches ok
(epoch 32)
(eval "(get (erlang-eval-ast \"Fn = discovery_fetch:make_fetch_fn([{peer_url, [{alice, <<__URL_BASE__>>}]}]), case Fn(alice) of {ok, AS} -> case envelope:get_field(public_keys, AS) of {ok, [_|_]} -> true; _ -> false end; _ -> false end\") :name)")

;; (b) closure on missing peer -> {error, no_peer_url}
(epoch 33)
(eval "(get (erlang-eval-ast \"Fn = discovery_fetch:make_fetch_fn([{peer_url, []}]), case Fn(alice) of {error, no_peer_url} -> true; _ -> false end\") :name)")

;; (b) closure GETs 404 path -> {error, {status, 404}}
(epoch 34)
(eval "(get (erlang-eval-ast \"R = discovery_fetch:fetch(<<__URL_BASE__,47,97,99,116,111,114,115,47,109,105,115,115,105,110,103>>, []), case R of {error, {status, 404}} -> true; _ -> false end\") :name)")

;; (b) lookup_or_fetch on cache miss writes the result back
(epoch 35)
(eval "(get (erlang-eval-ast \"Fn = discovery_fetch:make_fetch_fn([{peer_url, [{alice, <<__URL_BASE__>>}]}]), {R, NewState} = case peer_actors:lookup_or_fetch(alice, Fn, peer_actors:new()) of {ok, _AS, S} -> {ok, S}; {error, R0, S} -> {error, S} end, R =:= ok andalso peer_actors:peers(NewState) =:= [alice]\") :name)")
EPOCHS

sed -i "s|__URL_BASE__|${URL_BASE_BYTES}|g" "$TMPFILE"

OUTPUT=$(timeout 360 "$SX_SERVER" < "$TMPFILE" 2>/dev/null)

check() {
  local epoch="$1" desc="$2" expected="$3"
  local actual
  actual=$(echo "$OUTPUT" | awk -v e="$epoch" '
    $0 ~ "^\\(ok-len " e " " { getline; print; exit }
    $0 ~ "^\\(ok " e " "     { print; exit }
    $0 ~ "^\\(error " e " "  { print; exit }
  ')
  [ -z "$actual" ] && actual="<no output for epoch $epoch>"
  if echo "$actual" | grep -qF -- "$expected"; then
    PASS=$((PASS+1))
    [ "$VERBOSE" = "-v" ] && echo "  ok $desc"
  else
    FAIL=$((FAIL+1))
    ERRORS+="  FAIL [$desc] (epoch $epoch) expected: $expected | actual: $actual
"
  fi
}

check 12  "discovery_fetch loaded"                       "discovery_fetch"
check 20  "actor_doc Accept negotiates"                  "true"
check 21  "actor_doc /3 with kernel -> 200"              "true"
check 22  "body decodes to proplist w/ :public_keys"     "true"
check 23  "unknown actor -> 404"                         "true"
check 30  "actor_doc_url builds /actors/X"               "true"
check 31  "fetch live -> {ok, AS}"                       "true"
check 32  "closure -> {ok, AS}"                          "true"
check 33  "closure on missing peer -> no_peer_url"       "true"
check 34  "closure on 404 -> {status, 404}"              "true"
check 35  "lookup_or_fetch caches result"                "true"

TOTAL=$((PASS+FAIL))
if [ $FAIL -eq 0 ]; then
  echo "ok $PASS/$TOTAL next/tests/discovery_fetch.sh passed"
else
  echo "FAIL $PASS/$TOTAL passed, $FAIL failed:"
  echo "$ERRORS"
fi
[ $FAIL -eq 0 ]