;; lib/flow/tests/distributed.sx — Phase 4: distributed nodes via fed-sx (mocked).

(define flow-dist-pass 0)
(define flow-dist-fail 0)
(define flow-dist-fails (list))

(define
  flow-dist-test
  (fn
    (name actual expected)
    (if
      (= actual expected)
      (set! flow-dist-pass (+ flow-dist-pass 1))
      (begin
        (set! flow-dist-fail (+ flow-dist-fail 1))
        (append! flow-dist-fails {:name name :expected expected :actual actual})))))

(define flow-d (fn (src) (flow-run src)))

;; ── remote-node ─────────────────────────────────────────────────
(flow-dist-test
  "remote: a node executes on a peer"
  (flow-d
    "(flow-peer-register! (quote edge) (list (list (quote double) (lambda (x) (* x 2))))) (flow/start (remote-node (quote edge) (quote double)) 21)")
  42)
(flow-dist-test
  "remote: remote nodes compose in a sequence"
  (flow-d
    "(flow-peer-register! (quote edge) (list (list (quote inc) (lambda (x) (+ x 1))) (list (quote double) (lambda (x) (* x 2))))) (flow/start (sequence (remote-node (quote edge) (quote inc)) (remote-node (quote edge) (quote double))) 4)")
  10)
(flow-dist-test
  "remote: a remote node mixes with local nodes"
  (flow-d
    "(flow-peer-register! (quote edge) (list (list (quote double) (lambda (x) (* x 2))))) (flow/start (sequence (lambda (x) (+ x 5)) (remote-node (quote edge) (quote double)) (lambda (x) (- x 1))) 10)")
  29)
(flow-dist-test
  "remote: unreachable peer raises flow-remote-unreachable"
  (flow-d
    "(flow/start (try-catch (remote-node (quote ghost) (quote double)) (lambda (e) e)) 1)")
  "flow-remote-unreachable")
(flow-dist-test
  "remote: unknown function on a peer raises flow-remote-no-fn"
  (flow-d
    "(flow-peer-register! (quote edge) (list (list (quote double) (lambda (x) (* x 2))))) (flow/start (try-catch (remote-node (quote edge) (quote missing)) (lambda (e) e)) 1)")
  "flow-remote-no-fn")
(flow-dist-test
  "remote: a remote node can suspend the flow (peer returns control)"
  (flow-d
    "(flow-peer-register! (quote edge) (list (list (quote review) (lambda (x) x)))) (flow/start (sequence (remote-node (quote edge) (quote review)) (lambda (x) (suspend (quote human))) (lambda (v) (list (quote published) v))) 7)")
  (list "flow-suspended" 1 "human"))
(flow-dist-test
  "remote: a transient remote failure is recoverable with retry"
  (flow-d
    "(define hits 0) (flow-peer-register! (quote edge) (list (list (quote flaky) (lambda (x) (begin (set! hits (+ hits 1)) (if (< hits 2) (raise (quote down)) (* x 3))))))) (list (flow/start (retry 3 (remote-node (quote edge) (quote flaky))) 7) hits)")
  (list 21 2))

;; ── failover (retry on a different peer, fall through to local) ──
(flow-dist-test
  "failover: first reachable peer serves the request"
  (flow-d
    "(flow-peer-register! (quote p2) (list (list (quote f) (lambda (x) (+ x 100))))) (flow/start (remote-failover (list (quote p2) (quote down)) (quote f) (flow-const (quote local))) 5)")
  105)
(flow-dist-test
  "failover: skips an unreachable peer to the next one"
  (flow-d
    "(flow-peer-register! (quote p2) (list (list (quote f) (lambda (x) (+ x 100))))) (flow/start (remote-failover (list (quote down) (quote p2)) (quote f) (flow-const (quote local))) 5)")
  105)
(flow-dist-test
  "failover: skips a peer whose function raises"
  (flow-d
    "(flow-peer-register! (quote bad) (list (list (quote f) (lambda (x) (raise (quote boom)))))) (flow-peer-register! (quote good) (list (list (quote f) (lambda (x) (* x 10))))) (flow/start (remote-failover (list (quote bad) (quote good)) (quote f) (flow-const 0)) 4)")
  40)
(flow-dist-test
  "failover: all peers fail, the local fallback runs"
  (flow-d
    "(flow/start (remote-failover (list (quote down1) (quote down2)) (quote f) (lambda (x) (* x -1))) 9)")
  -9)
(flow-dist-test
  "failover: threads the input through to the chosen peer"
  (flow-d
    "(flow-peer-register! (quote p) (list (list (quote f) (lambda (x) (list (quote got) x))))) (flow/start (sequence (lambda (x) (+ x 1)) (remote-failover (list (quote p)) (quote f) (flow-const 0))) 41)")
  (list "got" 42))
(flow-dist-test
  "failover: composes inside a larger sequence"
  (flow-d
    "(flow-peer-register! (quote p) (list (list (quote f) (lambda (x) (* x 2))))) (flow/start (sequence (remote-failover (list (quote down) (quote p)) (quote f) (flow-const 1)) (lambda (x) (+ x 3))) 5)")
  13)

(define flow-dist-tests-run! (fn () {:total (+ flow-dist-pass flow-dist-fail) :passed flow-dist-pass :failed flow-dist-fail :fails flow-dist-fails}))