#!/usr/bin/env bash
# hosts/ocaml/test/persist_durable_test.sh
# Acceptance test for the host durable-storage adapter (Sx_persist_store).
#
# Exercises `persist/durable-backend` (REAL `perform`, not the mock) under the
# WORKTREE-built sx_server.exe, and asserts:
#   1. durable: writes land on disk and read back (the silent-data-loss repro
#      from plans/persist-on-sx.md now returns correct values).
#   2. last-seq is monotonic across truncate (compaction never reassigns a seq).
#   3. kv ops round-trip and delete.
#   4. recovery: a REAL process restart (write, exit, fresh process, replay)
#      recovers state from disk.
#
# Run from repo root or anywhere; locates the worktree binary relative to itself.
set -uo pipefail

HERE="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
ROOT="$(cd "$HERE/../../.." && pwd)"        # repo/worktree root
cd "$ROOT"

SX="hosts/ocaml/_build/default/bin/sx_server.exe"
if [ ! -x "$SX" ]; then
  echo "ERROR: worktree binary not found at $SX — build it first:" >&2
  echo "  (cd hosts/ocaml && dune build bin/sx_server.exe)" >&2
  exit 1
fi

DATADIR="$(mktemp -d)"
trap 'rm -rf "$DATADIR"' EXIT

PASS=0
FAIL=0
check() { # check <label> <got> <expected>
  if [ "$2" = "$3" ]; then
    PASS=$((PASS + 1)); printf '  ok   %-40s => %s\n' "$1" "$2"
  else
    FAIL=$((FAIL + 1)); printf '  FAIL %-40s got [%s] want [%s]\n' "$1" "$2" "$3"
  fi
}

PRELUDE='(epoch 1)
(load "spec/stdlib.sx")
(load "lib/r7rs.sx")
(load "lib/persist/event.sx")
(load "lib/persist/backend.sx")
(load "lib/persist/log.sx")
(load "lib/persist/kv.sx")
(load "lib/persist/durable.sx")
(load "lib/persist/blob.sx")
(epoch 2)'

# run_eval <sx-expr-string>: prints the final (ok-len 2 ...) payload line.
run_eval() {
  local expr="$1"
  printf '%s\n(eval %s)\n' "$PRELUDE" "$expr" \
    | SX_PERSIST_DIR="$DATADIR" timeout 60 "$SX" 2>/dev/null \
    | awk '/^\(ok-len 2 / {getline; print; exit}'
}

# escape an SX program into a single-line double-quoted SX string literal for
# (eval "..."). The REPL reads one command per physical line, so newlines in the
# program are collapsed to spaces.
q() { printf '"%s"' "$(printf '%s' "$1" | tr '\n' ' ' | sed 's/\\/\\\\/g; s/"/\\"/g')"; }

echo "== durable: append/read/last-seq round-trip on disk =="
GOT=$(run_eval "$(q '(let ((b (persist/durable-backend)))
  (begin
    (persist/append b "s" "x" 0 {:v 1})
    (persist/append b "s" "x" 0 {:v 2})
    (list (persist/event-seq (persist/append b "s" "x" 0 {:v 3}))
          (persist/count b "s")
          (len (persist/read b "s")))))')")
check "append/count/read" "$GOT" "(3 3 3)"

echo "== last-seq monotonic across truncate =="
GOT=$(run_eval "$(q '(let ((b (persist/durable-backend)))
  (begin
    (persist/append b "t" "x" 0 {})
    (persist/append b "t" "x" 0 {})
    (persist/append b "t" "x" 0 {})
    (persist/truncate b "t" 2)
    (list (persist/last-seq b "t") (persist/count b "t"))))')")
check "last-seq survives truncate" "$GOT" "(3 1)"

echo "== streams set survives compaction =="
GOT=$(run_eval "$(q '(let ((b (persist/durable-backend)))
  (sort ((get b "streams"))))')")
check "streams" "$GOT" '("s" "t")'

echo "== kv round-trip + delete =="
GOT=$(run_eval "$(q '(let ((b (persist/durable-backend)))
  (begin
    (persist/kv-put b "k" {:a 1 :b "two"})
    (persist/kv-put b "gone" 9)
    (persist/kv-delete b "gone")
    (list (get (persist/kv-get b "k") :b)
          (persist/kv-has? b "k")
          (persist/kv-has? b "gone"))))')")
check "kv get/has/delete" "$GOT" '("two" true false)'

echo "== recovery: state survives a REAL process restart =="
# write in process A then let it exit; the next run is a brand-new process.
run_eval "$(q '(let ((b (persist/durable-backend)))
  (begin
    (persist/append b "r" "ev" 0 {:n 1})
    (persist/append b "r" "ev" 0 {:n 2})
    (persist/kv-put b "survive" "yes")
    (persist/count b "r")))')" >/dev/null
# fresh process, same SX_PERSIST_DIR — must replay from disk.
GOT=$(run_eval "$(q '(let ((b (persist/durable-backend)))
  (list (persist/count b "r")
        (persist/last-seq b "r")
        (get (get (nth (persist/read b "r") 1) :data) :n)
        (persist/kv-get b "survive")))')")
check "recovered after restart" "$GOT" '(2 2 2 "yes")'

echo "== blob: content-addressed put/get/has? round-trip =="
GOT=$(run_eval "$(q '(let ((bs (persist/blob-store-backend)))
  (let ((r (persist/blob-store bs "hello world" "text/plain")))
    (list (persist/blob-size r)
          (persist/blob-mime r)
          (persist/blob-fetch bs r)
          (persist/blob-exists? bs r))))')")
check "blob size/mime/fetch/exists" "$GOT" '(11 "text/plain" "hello world" true)'

echo "== blob: put is content-addressed (idempotent cid) =="
GOT=$(run_eval "$(q '(let ((bs (persist/blob-store-backend)))
  (equal? (persist/blob-cid (persist/blob-store bs "same bytes" "x"))
          (persist/blob-cid (persist/blob-store bs "same bytes" "x"))))')")
check "same bytes -> same cid" "$GOT" "true"

echo "== blob: bytes + ref-in-kv survive a REAL restart =="
# process A: store a blob, keep only its ref in the durable kv.
run_eval "$(q '(let ((b (persist/durable-backend)) (bs (persist/blob-store-backend)))
  (begin (persist/kv-put b "logo" (persist/blob-store bs "PNGDATA" "image/png")) nil))')" >/dev/null
# fresh process: read the ref from kv, fetch the bytes from the blob store.
GOT=$(run_eval "$(q '(let ((b (persist/durable-backend)) (bs (persist/blob-store-backend)))
  (let ((r (persist/kv-get b "logo")))
    (list (persist/blob-fetch bs r) (persist/blob-exists? bs r) (persist/blob-mime r))))')")
check "blob recovered via ref after restart" "$GOT" '("PNGDATA" true "image/png")'

echo
echo "durable adapter: $PASS passed, $FAIL failed"
[ "$FAIL" -eq 0 ]