rose-ash/lib/events/booking.sx

;; lib/events/booking.sx — transactional, capacity-safe booking on persist.
;;
;; Each bookable occurrence has an append-only stream of :booking / :cancel
;; events. The live roster is the stream FOLDED in order — a booking adds an
;; actor, a cancel removes them — so a cancellation frees a seat and capacity
;; reopens. Capacity safety is the contract: two bookers racing for the last
;; seat must NEVER both succeed. The guarantee is delegated to persist's
;; optimistic concurrency — `persist/append-expect` appends only if the
;; stream's last-seq still equals what the writer observed; otherwise it returns
;; a conflict value and the writer retries against the advanced roster. So the
;; capacity check + append are atomic at the persist boundary, no lock.
;;
;; A booking/cancel decision is made against an OBSERVED snapshot (folded
;; roster + last-seq): two concurrent bookers each see the same free seat, both
;; attempt, and append-expect lets exactly one win — the loser gets a conflict
;; it retries.

(define ev-booking-stream (fn (occ-key) (str "booking:" occ-key)))

(define
  ev-bk-member?
  (fn
    (x xs)
    (cond
      ((empty? xs) false)
      ((= x (first xs)) true)
      (else (ev-bk-member? x (rest xs))))))

(define
  ev-bk-index
  (fn
    (xs x i)
    (cond
      ((empty? xs) -1)
      ((= (first xs) x) i)
      (else (ev-bk-index (rest xs) x (+ i 1))))))

(define ev-bk-remove (fn (xs a) (filter (fn (x) (not (= x a))) xs)))
(define ev-bk-append (fn (xs a) (append xs (list a))))

;; Fold a booking stream into the live roster (join order, cancels removed).
(define
  ev-fold-roster
  (fn
    (events)
    (reduce
      (fn
        (acc e)
        (let
          ((typ (persist/event-type e))
            (actor (get (persist/event-data e) :actor)))
          (cond
            ((= typ :booking)
              (if (ev-bk-member? actor acc) acc (ev-bk-append acc actor)))
            ((= typ :cancel) (ev-bk-remove acc actor))
            (else acc))))
      (list)
      events)))

;; Live roster (actors currently holding a seat), oldest active first.
(define
  ev-booked-actors
  (fn
    (b occ-key)
    (ev-fold-roster (persist/read b (ev-booking-stream occ-key)))))

(define
  ev-actor-booked?
  (fn (b occ-key actor) (ev-bk-member? actor (ev-booked-actors b occ-key))))

;; Live seat count (folded roster size — not the physical event count).
(define
  ev-booking-count
  (fn (b occ-key) (len (ev-booked-actors b occ-key))))

;; 1-based seat number for an actor on the roster (0 if not booked).
(define
  ev-seat-of
  (fn
    (actors actor)
    (let
      ((i (ev-bk-index actors actor 0)))
      (if (< i 0) 0 (+ i 1)))))

;; One booking attempt decided against an OBSERVED snapshot: `observed-actors`
;; (the roster the booker saw) and `expected` (the last-seq it saw). Returns
;; :already / :full / :booked / :conflict. :conflict means a concurrent append
;; landed since the snapshot — the caller must re-observe and retry.
(define
  ev/book-with-observed
  (fn
    (b occ-key capacity actor observed-actors expected)
    (cond
      ((ev-bk-member? actor observed-actors) {:seat (ev-seat-of observed-actors actor) :actor actor :status :already})
      ((>= (len observed-actors) capacity) {:actor actor :capacity capacity :status :full})
      (else
        (let
          ((r (persist/append-expect b (ev-booking-stream occ-key) expected :booking 0 {:actor actor})))
          (if (persist/conflict? r) {:actual (persist/conflict-actual r) :actor actor :status :conflict} {:seat (+ (len observed-actors) 1) :actor actor :status :booked}))))))

;; Capacity-safe booking with retry. Observes a consistent (roster, last-seq)
;; snapshot, attempts, and retries on conflict (a concurrent booker won the
;; race) — bounded by capacity.
(define
  ev/book!
  (fn
    (b occ-key capacity actor)
    (let
      ((res (ev/book-with-observed b occ-key capacity actor (ev-booked-actors b occ-key) (persist/last-seq b (ev-booking-stream occ-key)))))
      (if
        (= (get res :status) :conflict)
        (ev/book! b occ-key capacity actor)
        res))))

;; One cancellation attempt against an observed snapshot. :not-booked when the
;; actor holds no seat; :conflict on a racing append (retry); else :cancelled.
(define
  ev/cancel-with-observed
  (fn
    (b occ-key actor observed-actors expected)
    (cond
      ((not (ev-bk-member? actor observed-actors)) {:actor actor :status :not-booked})
      (else
        (let
          ((r (persist/append-expect b (ev-booking-stream occ-key) expected :cancel 0 {:actor actor})))
          (if (persist/conflict? r) {:actual (persist/conflict-actual r) :actor actor :status :conflict} {:actor actor :status :cancelled}))))))

;; Cancel an actor's seat, freeing capacity. Retries on conflict.
(define
  ev/cancel!
  (fn
    (b occ-key actor)
    (let
      ((res (ev/cancel-with-observed b occ-key actor (ev-booked-actors b occ-key) (persist/last-seq b (ev-booking-stream occ-key)))))
      (if (= (get res :status) :conflict) (ev/cancel! b occ-key actor) res))))

;; The roster as a plain list of actors (oldest active first).
(define ev/roster (fn (b occ-key) (ev-booked-actors b occ-key)))

;; Seats remaining for an occurrence of the given capacity.
(define
  ev/seats-left
  (fn
    (b occ-key capacity)
    (max 0 (- capacity (ev-booking-count b occ-key)))))