persist: materialized views — stay current on write, O(1) read + 11 tests

view.sx: persist/view bundles stream + fold + snapshot name; view-attach
subscribes it to a hub so each publish refreshes the snapshot incrementally,
making view-peek an O(1) current read. view-value always folds the tail so it
is never stale. The consumer read-model abstraction (feed indices, audit
rollups, search counters). 122/122.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

lib/persist/conformance.sh

@@ -13,7 +13,7 @@ if [ ! -x "$SX_SERVER" ]; then
   exit 1
 fi
 
-SUITES=(event log kv project subscribe concurrency snapshot compaction durable blob recovery)
+SUITES=(event log kv project subscribe concurrency snapshot compaction durable blob view recovery)
 
 OUT_JSON="lib/persist/scoreboard.json"
 OUT_MD="lib/persist/scoreboard.md"
@@ -37,6 +37,7 @@ run_suite() {
 (load "lib/persist/compaction.sx")
 (load "lib/persist/durable.sx")
 (load "lib/persist/blob.sx")
+(load "lib/persist/view.sx")
 (load "lib/persist/subscribe.sx")
 (load "lib/persist/api.sx")
 (epoch 2)

lib/persist/scoreboard.json

@@ -10,9 +10,10 @@
     "compaction": {"pass": 11, "fail": 0},
     "durable": {"pass": 15, "fail": 0},
     "blob": {"pass": 14, "fail": 0},
+    "view": {"pass": 11, "fail": 0},
     "recovery": {"pass": 6, "fail": 0}
   },
-  "total_pass": 111,
+  "total_pass": 122,
   "total_fail": 0,
-  "total": 111
+  "total": 122
 }

lib/persist/scoreboard.md

@@ -14,5 +14,6 @@ _Generated by `lib/persist/conformance.sh`_
 | compaction | 11 | 0 | 11 |
 | durable | 15 | 0 | 15 |
 | blob | 14 | 0 | 14 |
+| view | 11 | 0 | 11 |
 | recovery | 6 | 0 | 6 |
-| **Total** | **111** | **0** | **111** |
+| **Total** | **122** | **0** | **122** |

lib/persist/tests/view.sx

@@ -0,0 +1,105 @@
+; Extension — materialized views: stay current on write, read O(1) via peek.
+
+(define vw-count (fn (acc e) (+ acc 1)))
+(define vw (persist/view "order-count" "orders" vw-count 0))
+
+(persist-test "view-name" (persist/view-name vw) "order-count")
+(persist-test "view-stream" (persist/view-stream vw) "orders")
+(persist-test
+  "view-value folds the stream"
+  (let
+    ((b (persist/open)))
+    (begin
+      (persist/append b "orders" "x" 0 {})
+      (persist/append b "orders" "x" 0 {})
+      (persist/view-value b vw)))
+  2)
+(persist-test
+  "view-refresh persists a snapshot that peek then reads"
+  (let
+    ((b (persist/open)))
+    (begin
+      (persist/append b "orders" "x" 0 {})
+      (persist/view-refresh b vw)
+      (persist/view-peek b vw)))
+  1)
+(persist-test
+  "peek lags an un-refreshed tail"
+  (let
+    ((b (persist/open)))
+    (begin
+      (persist/append b "orders" "x" 0 {})
+      (persist/view-refresh b vw)
+      (persist/append b "orders" "x" 0 {})
+      (persist/view-peek b vw)))
+  1)
+(persist-test
+  "view-value sees the whole stream even after a stale snapshot"
+  (let
+    ((b (persist/open)))
+    (begin
+      (persist/append b "orders" "x" 0 {})
+      (persist/view-refresh b vw)
+      (persist/append b "orders" "x" 0 {})
+      (persist/view-value b vw)))
+  2)
+(persist-test
+  "attached view stays current on publish — peek needs no manual refresh"
+  (let
+    ((b (persist/open)))
+    (let
+      ((h (persist/view-attach (persist/hub b) vw)))
+      (begin
+        (persist/publish h "orders" "x" 0 {})
+        (persist/publish h "orders" "x" 0 {})
+        (persist/publish h "orders" "x" 0 {})
+        (persist/view-peek b vw))))
+  3)
+(persist-test
+  "attached view advances the snapshot seq incrementally"
+  (let
+    ((b (persist/open)))
+    (let
+      ((h (persist/view-attach (persist/hub b) vw)))
+      (begin
+        (persist/publish h "orders" "x" 0 {})
+        (persist/publish h "orders" "x" 0 {})
+        (persist/project-seq
+          (persist/snapshot-load b "order-count" 0)))))
+  2)
+(persist-test
+  "attach only reacts to its own stream"
+  (let
+    ((b (persist/open)))
+    (let
+      ((h (persist/view-attach (persist/hub b) vw)))
+      (begin
+        (persist/publish h "other" "x" 0 {})
+        (persist/view-peek b vw))))
+  0)
+(persist-test
+  "materialized view works on the durable backend"
+  (let
+    ((db (persist/mock-durable (persist/mem-backend))))
+    (let
+      ((h (persist/view-attach (persist/hub db) vw)))
+      (begin
+        (persist/publish h "orders" "x" 0 {})
+        (persist/publish h "orders" "x" 0 {})
+        (persist/view-peek db vw))))
+  2)
+(persist-test
+  "view sum over event data"
+  (let
+    ((b (persist/open))
+      (sumv
+        (persist/view
+          "rev"
+          "sales"
+          (fn (acc e) (+ acc (get (persist/event-data e) :amt)))
+          0)))
+    (begin
+      (persist/append b "sales" "sale" 0 {:amt 10})
+      (persist/append b "sales" "sale" 1 {:amt 25})
+      (persist/view-value b sumv)))
+  35)

lib/persist/view.sx

@@ -0,0 +1,49 @@
+; persist/view — a materialized view: the consumer-facing read model. It bundles
+; a stream, a fold (step + seed) and a snapshot name. Attached to a hub it
+; refreshes incrementally on every publish, so the materialized value stays
+; current on write and reads are O(1) snapshot loads (persist/view-peek) instead
+; of a full fold. This is what feed indices, mod audit rollups, search counters,
+; etc. sit on. Requires: lib/persist/snapshot.sx, lib/persist/subscribe.sx.
+
+(define persist/view (fn (name stream step seed) {:name name :step step :stream stream :seed seed}))
+(define persist/view-name (fn (v) (get v :name)))
+(define persist/view-stream (fn (v) (get v :stream)))
+
+; bring the view's snapshot up to date with the log tail; returns the state
+(define
+  persist/view-refresh
+  (fn
+    (b v)
+    (persist/checkpoint
+      b
+      (get v :stream)
+      (get v :name)
+      (get v :step)
+      (get v :seed))))
+
+; current materialized value — refreshes first, so never stale
+(define
+  persist/view-value
+  (fn (b v) (persist/project-value (persist/view-refresh b v))))
+
+; O(1) read of the last persisted snapshot value WITHOUT folding the tail. Equal
+; to view-value when the view is attached (kept current on every publish);
+; otherwise may lag the log by the un-refreshed tail.
+(define
+  persist/view-peek
+  (fn
+    (b v)
+    (persist/project-value
+      (persist/snapshot-load b (get v :name) (get v :seed)))))
+
+; attach to a hub: refresh the view on every publish to its stream
+(define
+  persist/view-attach
+  (fn
+    (h v)
+    (begin
+      (persist/subscribe
+        h
+        (persist/view-stream v)
+        (fn (bk s e) (persist/view-refresh bk v)))
+      h)))