#!/usr/bin/env bash
# RA SPIKE — can an SX-side runner drive next/'s durable Erlang flow (blog_publish_digest)
# through flow_store, using OUR canonical activity (serialized from SX), incl. suspend→resume?
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")
(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/kernel/trigger_registry.erl\")) :name)")
(eval "(get (erlang-load-module (file-read \"next/kernel/flow_dispatch.erl\")) :name)")
(eval "(get (erlang-load-module (file-read \"next/kernel/pipeline.erl\")) :name)")
(epoch 3)
(eval "(get (erlang-load-module (file-read \"next/flow/flows/blog_publish_digest.erl\")) :name)")
(epoch 4)
;; the SX-side RA marshaller: our canonical activity dict -> the Erlang activity proplist SOURCE.
;; (id emitted as a bare atom for the spike; a real CID marshals to a binary.)
(eval "(define ra/erl-src (fn (a) (str \"[{type, \" (get a :verb) \"}, {actor, \" (get a :actor) \"}, {id, \" (get a :id) \"}, {object, [{type, \" (get a :object-type) \"}, {category, \" (get a :category) \"}]}]\")))")
;; boot: registry + store + the flow (injected 3-follower fetch) + an on-Article trigger.
(eval "(define ra/boot \"trigger_registry:start_link(), flow_store:start_link(), FF = fun(_) -> [f1, f2, f3] end, Flow = blog_publish_digest:build([{fetch_followers, FF}]), flow_store:register_flow(blog_digest, Flow), Guard = fun(A, _) -> case envelope:get_field(object, A) of {ok, O} -> envelope:get_field(type, O) =:= {ok, article}; _ -> false end end, trigger_registry:add(create, trigger_registry:mk_spec(<<116,99>>, blog_digest, Guard, any)), Cfg = [{trigger_registry, trigger_registry}], AS = [{actor_id, alice}],\")")
;; OUR canonical activities (host/blog--activity->erl shape) — urgent + newsletter.
(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: our serialized activity drives the flow → done, 3 emails (sync branch) ──
(epoch 10)
(eval "(get (erlang-eval-ast (str ra/boot \" pipeline:apply_triggers(\" (ra/erl-src ra/urgent) \", AS, Cfg), {ok, {done, {digest_sent, Emails, _}}} = flow_store:status(1), length(Emails) =:= 3\")) :name)")
;; ── newsletter: SUSPENDS on the morning timer (durable, off-request) ──
(epoch 20)
(eval "(get (erlang-eval-ast (str ra/boot \" pipeline:apply_triggers(\" (ra/erl-src ra/news) \", AS, Cfg), flow_store:status(1) =:= {ok, {suspended, morning}}\")) :name)")
;; ── newsletter: RESUME the suspended instance → completes, 3 emails (the async cycle) ──
(epoch 21)
(eval "(get (erlang-eval-ast (str ra/boot \" pipeline:apply_triggers(\" (ra/erl-src ra/news) \", AS, Cfg), {ok, {flow_done, {digest_sent, Emails, _}}} = flow_store:resume(1, morning_ts), length(Emails) =:= 3\")) :name)")
;; ── the marshaller output itself (prove the SX serializer produces the right Erlang source) ──
(epoch 30)
(eval "(ra/erl-src ra/urgent)")
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 spike ─────────────────────────────────────────"
ck 10 "true"       "urgent: SX activity → flow → done, 3 emails"
ck 20 "true"       "newsletter: SX activity → flow SUSPENDS (morning)"
ck 21 "true"       "newsletter: RESUME → completes, 3 emails (async cycle)"
ck 30 "category, urgent" "marshaller emits valid Erlang activity source"
echo "─────────────────────────────────────────────────────"
echo "PASS=$PASS FAIL=$FAIL"