#!/usr/bin/env bash
# RA INTEGRATION — the REAL lib/host/ra.sx runner (host/ra--make-runner + host/ra--real-eval)
# driving next/'s live flow_store durable flow, end-to-end. (ra-spike.sh proved the path with
# inline Erlang; this proves the MODULE.) urgent→done, newsletter→suspended, resume→done.
set -uo pipefail
cd "$(git rev-parse --show-toplevel)"
SX_SERVER="${SX_SERVER:-hosts/ocaml/_build/default/bin/sx_server.exe}"
[ -x "$SX_SERVER" ] || SX_SERVER="/root/rose-ash/hosts/ocaml/_build/default/bin/sx_server.exe"
PASS=0; FAIL=0
TMPFILE=$(mktemp); trap "rm -f $TMPFILE" EXIT

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")
(load "lib/host/ra.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/flow/flow.erl\")) :name)")
(eval "(get (erlang-load-module (file-read \"next/flow/flow_spec.erl\")) :name)")
(eval "(get (erlang-load-module (file-read \"next/flow/flow_store.erl\")) :name)")
(eval "(get (erlang-load-module (file-read \"next/flow/flows/blog_publish_digest.erl\")) :name)")
(epoch 3)
;; gen_servers don't persist across separate erlang-eval-ast calls (flow README), so the injected
;; erl-eval boots the store + registers the flow inline on EVERY call (like the e2e). This proves
;; the MODULE's marshalling/dispatch/parse against the REAL flow; TRUE cross-call resume (suspend,
;; return, resume later) needs a PERSISTENT next/ kernel process — the RA-live deployment step.
(eval "(define ra/boot-eval (fn (src) (er-to-sx-deep (erlang-eval-ast (str \"flow_store:start_link(), FF = fun(_) -> [f1, f2, f3] end, flow_store:register_flow(bd, blog_publish_digest:build([{fetch_followers, FF}])), \" src)))))")
(eval "(define ra/R (host/ra--make-runner ra/boot-eval))")
(eval "(define ra/urgent {:verb \"create\" :actor \"alice\" :id \"u1\" :object-type \"article\" :category \"urgent\"})")
(eval "(define ra/news   {:verb \"create\" :actor \"alice\" :id \"n1\" :object-type \"article\" :category \"newsletter\"})")
;; ── urgent: the real MODULE runner → done ──
(epoch 10)
(eval "(get ((get ra/R :run) {:erl-flow \"bd\"} {:activity ra/urgent}) :status)")
;; ── newsletter: the real MODULE runner → suspended (durable wait) ──
(epoch 20)
(eval "(get ((get ra/R :run) {:erl-flow \"bd\"} {:activity ra/news}) :status)")
;; ── the suspended result carries a resume handle {:id :tag morning} ──
(epoch 21)
(eval "(str (get (get ((get ra/R :run) {:erl-flow \"bd\"} {:activity ra/news}) :resume) :tag))")
;; ── the done result carries the flow's effect-as-data (digest_sent) ──
(epoch 30)
(eval "(str (first (get ((get ra/R :run) {:erl-flow \"bd\"} {:activity ra/urgent}) :effects)))")
EPOCHS

OUT=$(timeout 360 "$SX_SERVER" < "$TMPFILE" 2>/dev/null)
grab() { echo "$OUT" | awk -v e="$1" '$0 ~ "^\\(ok " e " "{print;exit} $0 ~ "^\\(ok-len " e " "{getline;print;exit} $0 ~ "^\\(error " e " "{print;exit}'; }
ck() { local a; a=$(grab "$1"); if echo "$a" | grep -qF -- "$2"; then PASS=$((PASS+1)); echo "  ok  [$3]"; else FAIL=$((FAIL+1)); echo "  FAIL [$3] want '$2' got: $a"; fi; }

echo "── RA integration (the real host/ra.sx module) ──────"
ck 10 "done"        "urgent → real RA runner → done"
ck 20 "suspended"   "newsletter → real RA runner → suspended"
ck 21 "morning"     "suspended result carries a resume handle (:tag morning)"
ck 30 "digest_sent" "done result carries the flow's effect-as-data"
echo "─────────────────────────────────────────────────────"
echo "PASS=$PASS FAIL=$FAIL"