1	import Base
2	import mylsm-lsm-store@0.3.2.0/mylsm.bend as MyLsmStore
3	import mylsm-lsm-store@0.3.2.0/src/Db.bend as DbTypes
4	import mylsm-lsm-store@0.3.2.0/src/Wal.bend as WalTypes
5	
6	# Session effects for ber-core. Op is a newtype over Db transitions and
7	# Handle over MyLSM Db; do-blocks desugar through Op.bind/Op.pure exactly
8	# like Sess. Pairs sit only in return and parameter positions (never as a
9	# Sess answer), so Op stays kind-polymorphic; computed pairs thread through
10	# helpers (match on parameters, never on computed values). Defs are ordered
11	# leaf-first: no def calls one written below it.
12	#
13	# Every op returns its answer with a journal of physical mutations
14	# (execution order). Pure run_op drops it; run_op_tracked exposes it and
15	# run_op_durable appends it through one WAL frame plus fsync. oput/odel
16	# journal exactly what they apply, so the WAL mirrors the memtable.
17	
18	# Journal rope: O(1) per-bind combination (no calls, just nesting); flattened
19	# once at flush with an explicit work stack (constant native stack) into
20	# chronological order. Reads build Empty spines that flatten to Nil.
21	type Journal is Data:
22	  JEmpty{}
23	  JOne{mut: WalTypes.Mut}
24	  JMore{prior: Journal, post: Journal}
25	
26	# Counts rope nodes structurally; the work loop below visits each once.
27	def rope_nodes(+journal: Journal) -> Nat:
28	  match journal:
29	    case JEmpty{}:
30	      1n
31	    case JOne{unlogged_mut}:
32	      1n
33	    case JMore{prior, post}:
34	      Nat.add(1n, Nat.add(rope_nodes(prior), rope_nodes(post)))
35	
36	# Flattens a rope oldest-first with an explicit work stack; the accumulator
37	# holds the reversed prefix. Fuel covers nodes plus the final empty check.
38	def flatten_go(fuel: Nat, +work: List<&2, Journal>, +acc: List<&2, WalTypes.Mut>) -> List<&2, WalTypes.Mut>:
39	  match fuel:
40	    case 0n:
41	      acc
42	    case 1n+fuel_left:
43	      match work:
44	        case Nil{}:
45	          List.reverse(&2, WalTypes.Mut, acc)
46	        case Con{head, tail}:
47	          match head:
48	            case JEmpty{}:
49	              flatten_go(fuel_left, tail, acc)
50	            case JOne{mut}:
51	              flatten_go(fuel_left, tail, Con{mut, acc})
52	            case JMore{prior, post}:
53	              flatten_go(fuel_left, Con{prior, Con{post, tail}}, acc)
54	
55	# Materializes a journal rope into its chronological mutation list.
56	def flatten_journal(+journal: Journal) -> List<&2, WalTypes.Mut>:
57	  flatten_go(Nat.add(rope_nodes(journal), 1n), Con{journal, Nil{}}, Nil{})
58	
59	# One session step chain: a Db transition answering with its journal.
60	type Op<a, -A: Kind(a)> is Kind(a <&> &1):
61	  Op{run: DbTypes.Db -> (DbTypes.Db & A) & Journal}
62	
63	# An opened store handle; affine, threaded through run_op.
64	type Handle is Type:
65	  Handle{db: DbTypes.Db}
66	
67	# Unwraps the inner transition for translators like bind and run_op.
68	def Op.inner(b, -B: Kind(b), op: Op<b, B>) -> DbTypes.Db -> (DbTypes.Db & B) & Journal:
69	  match op:
70	    case Op{run_fn}:
71	      run_fn
72	
73	# Lifts a pure value with an empty journal.
74	def Op.pure(a, -A: Kind(a), val: A) -> Op<a, A>:
75	  Op{st => ((st, val), JEmpty{})}
76	
77	# Concatenates a prior journal before a later one, finishing the second step.
78	def Op.seq2(a, -B: Kind(a), +prior: Journal, second: (DbTypes.Db & B) & Journal) -> (DbTypes.Db & B) & Journal:
79	  match second:
80	    case ((db_next, answer), journal):
81	      ((db_next, answer), JMore{prior, journal})
82	
83	# Runs the next op on the first answer, threading Db and journal.
84	def Op.seq(a, -A: Kind(a), -B: Kind(a), first: (DbTypes.Db & A) & Journal, fun: A -> Op<a, B>) -> (DbTypes.Db & B) & Journal:
85	  match first:
86	    case ((db_mid, answer), prior):
87	      Op.seq2(a, B, prior, Op.inner(a, B, fun(answer))(db_mid))
88	
89	# Sequences two ops, concatenating journals in execution order.
90	def Op.bind(a, -A: Kind(a), -B: Kind(a), op: Op<a, A>, fun: A -> Op<a, B>) -> Op<a, B>:
91	  match op:
92	    case Op{run1}:
93	      Op{st => Op.seq(a, A, B, run1(st), fun)}
94	
95	# Pairs a finished point write with its journaled mutation.
96	def pair_put(+key: String, +val: String, db_next: DbTypes.Db) -> (DbTypes.Db & Unit) & Journal:
97	  ((db_next, Unit{}), JOne{WalTypes.Put{key, val}})
98	
99	# Writes one key and journals the mutation.
100	def oput(+key: String, +val: String) -> Op<&2, Unit>:
101	  Op{st => pair_put(key, val, MyLsmStore.put_go(st, key, val))}
102	
103	# Pairs a finished point read with an empty journal.
104	def pair_get(finished: DbTypes.Db & Maybe<&2, String>) -> (DbTypes.Db & Maybe<&2, String>) & Journal:
105	  match finished:
106	    case (db_next, found_val):
107	      ((db_next, found_val), JEmpty{})
108	
109	# Reads one key; missing reads as None. Reads journal nothing.
110	def oget(+key: String) -> Op<&2, Maybe<&2, String>>:
111	  Op{st => pair_get(MyLsmStore.sget_go(st, key))}
112	
113	# Pairs a finished point delete with its journaled mutation.
114	def pair_del(+key: String, db_next: DbTypes.Db) -> (DbTypes.Db & Unit) & Journal:
115	  ((db_next, Unit{}), JOne{WalTypes.Del{key}})
116	
117	# Deletes one key and journals the mutation.
118	def odel(+key: String) -> Op<&2, Unit>:
119	  Op{st => pair_del(key, MyLsmStore.del_go(st, key))}
120	
121	# Opens a store handle over a directory label (pure; no IO touched).
122	def open_store(+dir: String) -> Handle:
123	  Handle{MyLsmStore.open(dir)}
124	
125	# Default store label for CLI use (pure; no IO touched).
126	def open_default_store() -> Handle:
127	  open_store("./.ber")
128	
129	# Wraps a reopened database; IO errors propagate as Fail.
130	def wrap_reopened(reopened: Result<&1, &1, U32 & String, DbTypes.Db>) -> Result<&1, &1, U32 & String, Handle>:
131	  match reopened:
132	    case Fail{error}:
133	      Fail{error}
134	    case Done{db}:
135	      Done{Handle{db}}
136	
137	# Opens a durable handle, replaying wal.log when present; missing file
138	# answers empty. The caller ensures the directory exists.
139	def open_durable(+dir: String) -> IO(Result<&1, &1, U32 & String, Handle>):
140	  do IO<Result<&1, &1, U32 & String, Handle>>:
141	    reopened : Result<&1, &1, U32 & String, DbTypes.Db> <- MyLsmStore.reopen(dir)
142	    return wrap_reopened(reopened)
143	
144	# Projects the materialized journal out of a tracked run (laws observe this).
145	def tracked_journal(a, -A: Kind(a), tracked: (Handle & A) & Journal) -> List<&2, WalTypes.Mut>:
146	  match tracked:
147	    case ((unused_handle, unused_answer), journal):
148	      flatten_journal(journal)
149	
150	# Rewraps a computed runner pair into Handle form, dropping the journal.
151	def rewrap_pair(a, -A: Kind(a), computed: (DbTypes.Db & A) & Journal) -> Handle & A:
152	  match computed:
153	    case ((db_next, answer), unused_journal):
154	      (Handle{db_next}, answer)
155	
156	# Executes a whole op chain against a handle (pure; journal dropped).
157	def run_op(a, -A: Kind(a), store: Handle, op: Op<a, A>) -> Handle & A:
158	  match store:
159	    case Handle{db}:
160	      rewrap_pair(a, A, Op.inner(a, A, op)(db))
161	
162	# Rewraps a tracked runner pair, keeping the journal.
163	def rewrap_tracked(a, -A: Kind(a), computed: (DbTypes.Db & A) & Journal) -> (Handle & A) & Journal:
164	  match computed:
165	    case ((db_next, answer), journal):
166	      ((Handle{db_next}, answer), journal)
167	
168	# Executes a whole op chain, exposing the journal for the WAL runner.
169	def run_op_tracked(a, -A: Kind(a), store: Handle, op: Op<a, A>) -> (Handle & A) & Journal:
170	  match store:
171	    case Handle{db}:
172	      rewrap_tracked(a, A, Op.inner(a, A, op)(db))
173	
174	# Wraps a written database with its answer; IO errors propagate as Fail.
175	def wrap_written(a, -A: Kind(a), written: Result<&1, &1, U32 & String, DbTypes.Db>, answer: A) -> Result<&1, &1, U32 & String, Handle & A>:
176	  match written:
177	    case Fail{error}:
178	      Fail{error}
179	    case Done{db_next}:
180	      Done{(Handle{db_next}, answer)}
181	
182	# Skips the WAL when the journal is empty (pure reads never fsync).
183	def flush_clean(a, -A: Kind(a), store_next: Handle, answer: A) -> IO(Result<&1, &1, U32 & String, Handle & A>):
184	  match store_next:
185	    case Handle{db}:
186	      IO.pure(Result<&1, &1, U32 & String, Handle & A>, Done{(Handle{db}, answer)})
187	
188	# Flushes a nonempty journal through one WAL frame plus fsync.
189	def flush_dirty(a, -A: Kind(a), store_next: Handle, answer: A, +head_mut: WalTypes.Mut, +tail_muts: List<&2, WalTypes.Mut>) -> IO(Result<&1, &1, U32 & String, Handle & A>):
190	  match store_next:
191	    case Handle{db}:
192	      do IO<Result<&1, &1, U32 & String, Handle & A>>:
193	        written : Result<&1, &1, U32 & String, DbTypes.Db> <- DbTypes.db_write(db, WalTypes.Batch{Con{head_mut, tail_muts}})
194	        return wrap_written(a, A, written, answer)
195	
196	# Dispatches on journal emptiness without matching computed values.
197	def flush_list(a, -A: Kind(a), store_next: Handle, answer: A, +mutations: List<&2, WalTypes.Mut>) -> IO(Result<&1, &1, U32 & String, Handle & A>):
198	  match mutations:
199	    case Nil{}:
200	      flush_clean(a, A, store_next, answer)
201	    case Con{head_mut, tail_muts}:
202	      flush_dirty(a, A, store_next, answer, head_mut, tail_muts)
203	
204	# Splits a tracked runner pair for the flush dispatch.
205	def flush_tracked(a, -A: Kind(a), tracked: (Handle & A) & Journal) -> IO(Result<&1, &1, U32 & String, Handle & A>):
206	  match tracked:
207	    case ((store_next, answer), journal):
208	      flush_list(a, A, store_next, answer, flatten_journal(journal))
209	
210	# Executes a whole op chain durably: pure run, then one WAL append plus
211	# fsync of exactly the journaled mutations.
212	def run_op_durable(a, -A: Kind(a), store: Handle, op: Op<a, A>) -> IO(Result<&1, &1, U32 & String, Handle & A>):
213	  flush_tracked(a, A, run_op_tracked(a, A, store, op))