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 events:
;;
;;   :booking  free booking — actor immediately holds a confirmed seat
;;   :hold     provisional hold — seat reserved while payment is pending
;;   :confirm  a held seat becomes confirmed (payment succeeded)
;;   :release  a held seat is abandoned (payment failed/expired) — seat freed
;;   :cancel   a held or confirmed seat is given up — seat freed
;;
;; The live state is the stream FOLDED in order into per-actor seat states
;; (:held / :confirmed); an actor in ANY state occupies a seat, so both held and
;; confirmed seats count toward capacity — a pending payment cannot be
;; oversold. A freed seat (release/cancel) reopens capacity.
;;
;; Capacity safety is the contract: two writers racing for the last seat must
;; NEVER both succeed. Seat-ACQUIRING writes (:booking, :hold) go through
;; persist's optimistic concurrency — `persist/append-expect` appends only if
;; the stream's last-seq still equals what the writer observed; else it returns
;; a conflict the writer retries. Seat-FREEING writes (:cancel, :release) and
;; the state transition (:confirm) never oversell, so they append directly.

(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-append (fn (xs a) (append xs (list a))))
(define ev-bk-remove (fn (xs a) (filter (fn (x) (not (= x a))) xs)))

;; ---- per-actor state association list: ((actor state) ...) in join order ----

(define
  ev-state-has?
  (fn
    (states actor)
    (cond
      ((empty? states) false)
      ((= (first (first states)) actor) true)
      (else (ev-state-has? (rest states) actor)))))

(define
  ev-state-get
  (fn
    (states actor)
    (cond
      ((empty? states) :none)
      ((= (first (first states)) actor) (first (rest (first states))))
      (else (ev-state-get (rest states) actor)))))

(define
  ev-state-del
  (fn (states actor) (filter (fn (p) (not (= (first p) actor))) states)))

(define
  ev-state-set
  (fn
    (states actor st)
    (if
      (ev-state-has? states actor)
      (map (fn (p) (if (= (first p) actor) (list actor st) p)) states)
      (append states (list (list actor st))))))

;; Fold the booking stream into per-actor seat states (join order preserved).
(define
  ev-fold-states
  (fn
    (events)
    (reduce
      (fn
        (acc e)
        (let
          ((typ (persist/event-type e))
            (actor (get (persist/event-data e) :actor)))
          (cond
            ((= typ :booking) (ev-state-set acc actor :confirmed))
            ((= typ :hold) (ev-state-set acc actor :held))
            ((= typ :confirm)
              (if
                (ev-state-has? acc actor)
                (ev-state-set acc actor :confirmed)
                acc))
            ((= typ :cancel) (ev-state-del acc actor))
            ((= typ :release) (ev-state-del acc actor))
            (else acc))))
      (list)
      events)))

(define
  ev-states-of
  (fn
    (b occ-key)
    (ev-fold-states (persist/read b (ev-booking-stream occ-key)))))

;; Live roster (actors holding a seat — held or confirmed), oldest active first.
(define
  ev-booked-actors
  (fn (b occ-key) (map (fn (p) (first p)) (ev-states-of b 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 — both held and confirmed seats).
(define
  ev-booking-count
  (fn (b occ-key) (len (ev-booked-actors b occ-key))))

;; Seat state for an actor: :held / :confirmed / :none.
(define
  ev/seat-state
  (fn (b occ-key actor) (ev-state-get (ev-states-of b occ-key) actor)))

;; 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)))))

;; ---- seat-acquiring writes (capacity-guarded via append-expect) ----

;; One seat-acquiring attempt of `kind` (:booking or :hold) against an OBSERVED
;; snapshot (roster the writer saw + the last-seq). Returns :already / :full /
;; :conflict, or a success dict tagged with `ok-status`. :conflict means a
;; concurrent append landed since the snapshot — the caller must re-observe.
(define
  ev-acquire-with-observed
  (fn
    (b occ-key capacity actor observed-actors expected kind ok-status)
    (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 kind 0 {:actor actor})))
          (if (persist/conflict? r) {:actual (persist/conflict-actual r) :actor actor :status :conflict} {:seat (+ (len observed-actors) 1) :actor actor :status ok-status}))))))

(define
  ev-acquire!
  (fn
    (b occ-key capacity actor kind ok-status)
    (let
      ((res (ev-acquire-with-observed b occ-key capacity actor (ev-booked-actors b occ-key) (persist/last-seq b (ev-booking-stream occ-key)) kind ok-status)))
      (if
        (= (get res :status) :conflict)
        (ev-acquire! b occ-key capacity actor kind ok-status)
        res))))

;; Capacity-safe confirmed booking (retrying on conflict).
(define
  ev/book!
  (fn
    (b occ-key capacity actor)
    (ev-acquire! b occ-key capacity actor :booking :booked)))

;; Capacity-safe provisional hold (retrying on conflict). The seat is reserved
;; (counts toward capacity) until confirmed or released.
(define
  ev/hold!
  (fn
    (b occ-key capacity actor)
    (ev-acquire! b occ-key capacity actor :hold :held)))

;; Test seam: one attempt against a caller-supplied snapshot (book or hold).
(define
  ev/book-with-observed
  (fn
    (b occ-key capacity actor observed-actors expected)
    (ev-acquire-with-observed
      b
      occ-key
      capacity
      actor
      observed-actors
      expected
      :booking :booked)))

(define
  ev/hold-with-observed
  (fn
    (b occ-key capacity actor observed-actors expected)
    (ev-acquire-with-observed
      b
      occ-key
      capacity
      actor
      observed-actors
      expected
      :hold :held)))

;; ---- state transitions / seat-freeing writes (no oversell, append direct) ----

;; Confirm a held seat (payment succeeded). :confirmed on success,
;; :already-confirmed if it was confirmed, :not-held otherwise.
(define
  ev/confirm!
  (fn
    (b occ-key actor)
    (let
      ((st (ev/seat-state b occ-key actor)))
      (cond
        ((= st :held)
          (begin
            (persist/append
              b
              (ev-booking-stream occ-key)
              :confirm 0
              {:actor actor})
            {:actor actor :status :confirmed}))
        ((= st :confirmed) {:actor actor :status :already-confirmed})
        (else {:actor actor :status :not-held})))))

;; Release a held seat (payment failed/expired), freeing it. Only valid for a
;; held seat — confirmed bookings are given up via ev/cancel!.
(define
  ev/release!
  (fn
    (b occ-key actor)
    (let
      ((st (ev/seat-state b occ-key actor)))
      (if
        (= st :held)
        (begin
          (persist/append
            b
            (ev-booking-stream occ-key)
            :release 0
            {:actor actor})
          {:actor actor :status :released})
        {:actor actor :status :not-held}))))

;; Cancel a held or confirmed seat, freeing it. :cancelled or :not-booked.
(define
  ev/cancel!
  (fn
    (b occ-key actor)
    (if
      (ev-bk-member? actor (ev-booked-actors b occ-key))
      (begin
        (persist/append
          b
          (ev-booking-stream occ-key)
          :cancel 0
          {:actor actor})
        {:actor actor :status :cancelled})
      {:actor actor :status :not-booked})))

;; 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)))))

;; ---- waitlist ----
;; When an occurrence is full, actors join a FIFO waitlist (:waitlist /
;; :unwaitlist events on the same stream). Taking a seat (:booking / :hold)
;; removes an actor from the queue, so the waitlist fold is independent of the
;; seat fold. Cancelling/releasing a seat can auto-promote the head of the
;; queue (a :booking appended for them).

(define
  ev-fold-waiting
  (fn
    (events)
    (reduce
      (fn
        (acc e)
        (let
          ((typ (persist/event-type e))
            (actor (get (persist/event-data e) :actor)))
          (cond
            ((= typ :waitlist) (if (ev-bk-member? actor acc) acc (ev-bk-append acc actor)))
            ((= typ :unwaitlist) (ev-bk-remove acc actor))
            ((= typ :booking) (ev-bk-remove acc actor))
            ((= typ :hold) (ev-bk-remove acc actor))
            (else acc))))
      (list)
      events)))

;; The current waitlist queue (FIFO, oldest first).
(define
  ev/waitlist
  (fn (b occ-key) (ev-fold-waiting (persist/read b (ev-booking-stream occ-key)))))

;; 1-based queue position for an actor (0 if not waiting).
(define
  ev/waitlist-position
  (fn (b occ-key actor) (ev-seat-of (ev/waitlist b occ-key) actor)))

;; Book if a seat is free, else join the waitlist. Idempotent: already seated →
;; :already; already queued → :already-waiting.
(define
  ev/waitlist!
  (fn
    (b occ-key capacity actor)
    (let
      ((seats (ev-booked-actors b occ-key))
        (waiting (ev/waitlist b occ-key)))
      (cond
        ((ev-bk-member? actor seats)
          {:status :already :seat (ev-seat-of seats actor) :actor actor})
        ((ev-bk-member? actor waiting)
          {:status :already-waiting :position (ev-seat-of waiting actor) :actor actor})
        (else
          (let
            ((r (ev/book! b occ-key capacity actor)))
            (if
              (= (get r :status) :booked)
              r
              (begin
                (persist/append b (ev-booking-stream occ-key) :waitlist 0 {:actor actor})
                {:status :waitlisted
                  :position (+ (len waiting) 1)
                  :actor actor}))))))))

;; Leave the waitlist. :left or :not-waiting.
(define
  ev/leave-waitlist!
  (fn
    (b occ-key actor)
    (if
      (ev-bk-member? actor (ev/waitlist b occ-key))
      (begin
        (persist/append b (ev-booking-stream occ-key) :unwaitlist 0 {:actor actor})
        {:status :left :actor actor})
      {:status :not-waiting :actor actor})))

;; Cancel a seat and, if that frees capacity, auto-promote the head of the
;; waitlist (a confirmed booking). Returns the cancel result plus :promoted
;; (the actor promoted, or nil).
(define
  ev/cancel-promote!
  (fn
    (b occ-key capacity actor)
    (let
      ((c (ev/cancel! b occ-key actor)))
      (if
        (= (get c :status) :cancelled)
        (let
          ((waiting (ev/waitlist b occ-key))
            (seats (ev-booked-actors b occ-key)))
          (if
            (and (not (empty? waiting)) (< (len seats) capacity))
            (let
              ((promoted (first waiting)))
              (begin
                (persist/append b (ev-booking-stream occ-key) :booking 0 {:actor promoted})
                {:status :cancelled :actor actor :promoted promoted}))
            {:status :cancelled :actor actor :promoted nil}))
        c))))