#!/usr/bin/env bash
# next/tests/smoke_federate.sh — m2 Step 12 acceptance test.
#
# Spins up TWO sx_server instances on distinct ephemeral ports,
# wires each as a federation instance (one actor per instance,
# peer-AS pre-populated for inbound signature verification, peer
# URL pre-populated so dispatch_http knows where to send outbound
# activities), then drives the live HTTP federation flow:
#
#   1. Both listeners up + serving their welcome route.
#   2. Each instance serves its own actor-doc (kernel-aware route,
#      proves the Blockers #4 fix landed end-to-end).
#   3. alice@A signs a Follow envelope targeting bob@B and POSTs it
#      to B's /actors/bob/inbox over real HTTP. B's auto-accept
#      fires (pipeline validates the sig against the pre-populated
#      peer-AS, kernel appends to inbox, accept Activity gets
#      published into bob's outbox + delivery_worker for alice).
#   4. bob's outbox tip advances by at least 1 (the Accept).
#
# Step 8b-timer is still gated on Blockers #3 (send_after), so the
# delivery_worker queue is drained synchronously rather than via the
# retry loop — the test inspects worker state directly.

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=""

PORT_A=$(python3 -c 'import socket;s=socket.socket();s.bind(("127.0.0.1",0));print(s.getsockname()[1]);s.close()')
PORT_B=$(python3 -c 'import socket;s=socket.socket();s.bind(("127.0.0.1",0));print(s.getsockname()[1]);s.close()')

EF_A=$(mktemp); EF_B=$(mktemp)
LOG_A=$(mktemp); LOG_B=$(mktemp)
FIFO_A=$(mktemp -u); FIFO_B=$(mktemp -u)
ENV_FILE=$(mktemp)
mkfifo "$FIFO_A"; mkfifo "$FIFO_B"

cleanup() {
  for pid in ${SXA:-} ${SXB:-} ${HA:-} ${HB:-}; do
    kill -KILL "$pid" 2>/dev/null || true
    wait "$pid" 2>/dev/null || true
  done
  rm -f "$EF_A" "$EF_B" "$LOG_A" "$LOG_B" "$FIFO_A" "$FIFO_B" "$ENV_FILE"
}
trap cleanup EXIT

# Per-instance boot script. Each instance:
#   - registers its actor with its KEY
#   - registers a delivery_worker for the PEER actor
#   - populates Cfg with auto-accept + peer-AS for sig verification
#   - http_server:start(PORT, Cfg)
write_boot() {
  local out="$1" port="$2" actor="$3" actor_kb="$4" peer="$5" peer_kb="$6"
  cat > "$out" <<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!)")
(eval "(get (erlang-load-module (file-read \"next/kernel/envelope.erl\")) :name)")
(eval "(get (erlang-load-module (file-read \"next/kernel/log.erl\")) :name)")
(eval "(get (erlang-load-module (file-read \"next/kernel/pipeline.erl\")) :name)")
(eval "(get (erlang-load-module (file-read \"next/kernel/term_codec.erl\")) :name)")
(eval "(get (erlang-load-module (file-read \"next/kernel/outbox.erl\")) :name)")
(eval "(get (erlang-load-module (file-read \"next/kernel/follower_graph.erl\")) :name)")
(eval "(get (erlang-load-module (file-read \"next/kernel/delivery.erl\")) :name)")
(eval "(get (erlang-load-module (file-read \"next/kernel/backfill.erl\")) :name)")
(eval "(get (erlang-load-module (file-read \"next/kernel/delivery_worker.erl\")) :name)")
(eval "(get (erlang-load-module (file-read \"next/kernel/nx_kernel.erl\")) :name)")
(eval "(get (erlang-load-module (file-read \"next/kernel/http_server.erl\")) :name)")
(epoch 3)
(eval "(erlang-eval-ast \"AK = <<${actor_kb},${actor_kb},${actor_kb},${actor_kb}>>, AKS = [{key_id,k1},{algorithm,ed25519},{value,AK}], AAS = [{public_keys,[[{id,k1},{created,0},{value,AK}]]}], BK = <<${peer_kb},${peer_kb},${peer_kb},${peer_kb}>>, BAS = [{public_keys,[[{id,k1},{created,0},{value,BK}]]}], nx_kernel:start_link(${actor}, AKS, AAS), delivery_worker:start_link(${peer}), Cfg = [{kernel, nx_kernel}, {auto_accept_follows, true}, {backfill_enabled, false}, {peer_as, [{${peer}, BAS}]}], http_server:start(${port}, Cfg)\")")
EPOCHS
}

# alice@A: key bytes 1; expects bob with key bytes 2
write_boot "$EF_A" "$PORT_A" "alice" "1" "bob"   "2"
# bob@B: key bytes 2; expects alice with key bytes 1
write_boot "$EF_B" "$PORT_B" "bob"   "2" "alice" "1"

# Boot both instances.
( cat "$EF_A"; sleep 900 ) > "$FIFO_A" &
HA=$!
"$SX_SERVER" < "$FIFO_A" > "$LOG_A" 2>&1 &
SXA=$!
rm -f "$FIFO_A"

( cat "$EF_B"; sleep 900 ) > "$FIFO_B" &
HB=$!
"$SX_SERVER" < "$FIFO_B" > "$LOG_B" 2>&1 &
SXB=$!
rm -f "$FIFO_B"

wait_bound() {
  local port="$1" started="$2"
  while [ $(($(date +%s) - started)) -lt 400 ]; do
    if (exec 3<>/dev/tcp/127.0.0.1/$port) 2>/dev/null; then
      exec 3<&-; exec 3>&-
      return 0
    fi
    sleep 1
  done
  return 1
}

START=$(date +%s)
if ! wait_bound "$PORT_A" "$START"; then
  echo "FAIL: instance A never bound on port $PORT_A"
  echo "--- log A tail ---"; tail -20 "$LOG_A"
  exit 1
fi
if ! wait_bound "$PORT_B" "$START"; then
  echo "FAIL: instance B never bound on port $PORT_B"
  echo "--- log B tail ---"; tail -20 "$LOG_B"
  exit 1
fi

[ "$VERBOSE" = "-v" ] && echo "  ok both instances up after $(($(date +%s) - START))s (A=$PORT_A B=$PORT_B)"

# ── helpers ───────────────────────────────────────────────────
check_text() {
  local desc="$1" url="$2" needle="$3"
  local resp
  resp=$(curl -s --max-time 15 "$url" 2>/dev/null || echo "<curl-failed>")
  if echo "$resp" | grep -qF -- "$needle"; then
    PASS=$((PASS+1)); [ "$VERBOSE" = "-v" ] && echo "  ok $desc"
  else
    FAIL=$((FAIL+1))
    ERRORS+="  FAIL [$desc] expected '$needle' in resp: $(echo "$resp" | head -c 120)
"
  fi
}

check_status() {
  local desc="$1" method="$2" url="$3" body_file="$4" expected="$5"
  local args=(-s -o /tmp/sfederate_body -w "%{http_code}" -X "$method" --max-time 15)
  if [ "$method" = "POST" ]; then
    args+=(-H "Content-Type: application/vnd.fed-sx.activity" --data-binary "@$body_file")
  fi
  args+=("$url")
  local code
  code=$(curl "${args[@]}" 2>/dev/null || echo "000")
  if [ "$code" = "$expected" ]; then
    PASS=$((PASS+1)); [ "$VERBOSE" = "-v" ] && echo "  ok $desc ($code)"
  else
    FAIL=$((FAIL+1))
    local body=$(cat /tmp/sfederate_body 2>/dev/null | head -c 120)
    ERRORS+="  FAIL [$desc] expected $expected got $code body: $body
"
  fi
}

# ── 1. Welcome on both instances ─────────────────────────────
check_text "A serves welcome /" "http://127.0.0.1:$PORT_A/" "fed-sx kernel m1"
check_text "B serves welcome /" "http://127.0.0.1:$PORT_B/" "fed-sx kernel m1"

# ── 2. Each instance serves its own actor's outbox (kernel-aware) ─
check_text "A: alice outbox tip" "http://127.0.0.1:$PORT_A/actors/alice/outbox" "tip: 0"
check_text "B: bob outbox tip"   "http://127.0.0.1:$PORT_B/actors/bob/outbox"   "tip: 0"

# ── 3. Build a signed Follow envelope (alice -> bob) ─────────
# Run a separate sx_server subprocess to construct + sign + encode.
cat > /tmp/build_follow.sx <<'BUILD'
(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")
(epoch 2)
(eval "(get (erlang-load-module (file-read \"next/kernel/envelope.erl\")) :name)")
(eval "(get (erlang-load-module (file-read \"next/kernel/outbox.erl\")) :name)")
(eval "(get (erlang-load-module (file-read \"next/kernel/term_codec.erl\")) :name)")
(epoch 10)
(eval "(let ((b (erlang-eval-ast \"AK = <<1,1,1,1>>, AKS = [{key_id,k1},{algorithm,ed25519},{value,AK}], Env = outbox:construct(follow, alice, 1, bob), Signed = outbox:sign(Env, AKS), term_codec:encode(Signed)\"))) (file-write \"__ENV_FILE__\" (list->string (map integer->char (get b :bytes)))))")
BUILD
sed -i "s|__ENV_FILE__|${ENV_FILE}|g" /tmp/build_follow.sx
timeout 240 "$SX_SERVER" < /tmp/build_follow.sx > /dev/null 2>&1
rm -f /tmp/build_follow.sx

if [ ! -s "$ENV_FILE" ]; then
  echo "FAIL: signed Follow envelope was not built (empty file)"
  exit 1
fi

# ── 4. POST the signed Follow into B's inbox ────────────────
check_status "alice -> bob Follow accepted" POST \
  "http://127.0.0.1:$PORT_B/actors/bob/inbox" "$ENV_FILE" "202"

# Give B's auto-accept a moment to publish the Accept into the
# outbox. The publish is synchronous from the route handler's
# point of view, but the gen_server reply to nx_kernel may queue
# behind our outbox tip read.
sleep 1

# ── 5. bob's outbox tip should now show >= 1 (the Accept) ────
check_text "B: bob outbox tip after Accept" \
  "http://127.0.0.1:$PORT_B/actors/bob/outbox" "tip: 1"

TOTAL=$((PASS+FAIL))
if [ $FAIL -eq 0 ]; then
  echo "ok $PASS/$TOTAL next/tests/smoke_federate.sh passed (A=$PORT_A B=$PORT_B)"
else
  echo "FAIL $PASS/$TOTAL passed, $FAIL failed:"
  echo "$ERRORS"
  if [ "$VERBOSE" = "-v" ]; then
    echo "--- log A tail ---"; tail -25 "$LOG_A"
    echo "--- log B tail ---"; tail -25 "$LOG_B"
  fi
fi
[ $FAIL -eq 0 ]