1	# Retry-safe commands: deduplicate by command id, replay recorded results, and commit state, record, and outbox in one SQLite transaction. Source: https://github.com/paymog/bend-kit/tree/main/commands
2	import Base
3	import bend-kit-bytes@0.3.2.0/bytes.bend as Bytes
4	import bend-kit-sqlite@0.1.0.1/sqlite.bend as Sqlite
5	
6	# ---- the pure processor
7	
8	type Outcome<-R: Type> is Type:
9	  Accepted{result: R}
10	  Refused{result: R}
11	
12	# What the store keeps for one command id: its payload, its outcome, and when it was recorded.
13	type Record<-P: Type, -R: Type> is Type:
14	  Record{payload: P, outcome: Outcome<R>, at: U32}
15	
16	type Reply<-R: Type> is Type:
17	  Fresh{outcome: Outcome<R>}
18	  Replayed{outcome: Outcome<R>}
19	  Conflict{}
20	
21	# The domain's answer to a new command.
22	type Decision<-S: Type, -R: Type, -E: Type> is Type:
23	  Accept{state: S, result: R, effects: List<E>}
24	  Refuse{result: R}
25	
26	# Write: commit the state, the record, and the effects together. Note: commit only the record. Keep: write nothing.
27	type Step<-S: Type, -P: Type, -R: Type, -E: Type> is Type:
28	  Write{state: S, record: Record<P, R>, reply: Reply<R>, effects: List<E>}
29	  Note{record: Record<P, R>, reply: Reply<R>}
30	  Keep{reply: Reply<R>}
31	
32	def accepted(-S: Type, -P: Type, -R: Type, -E: Type, rr: R & R, s2: S, p: P, at: U32, es: List<E>) -> Step<S, P, R, E>:
33	  (r, r2) = rr
34	  Write{s2, Record{p, Accepted{r}, at}, Fresh{Accepted{r2}}, es}
35	
36	def refused(-S: Type, -P: Type, -R: Type, -E: Type, rr: R & R, p: P, at: U32) -> Step<S, P, R, E>:
37	  (r, r2) = rr
38	  Note{Record{p, Refused{r}, at}, Fresh{Refused{r2}}}
39	
40	def decided(~S: Type, ~P: Type, ~R: Type, ~E: Type, ~dup_r: R -> R & R, d: Decision<S, R, E>, p: P, at: U32) -> Step<S, P, R, E>:
41	  match d:
42	    case Accept{s2, r, es}:
43	      accepted(S, P, R, E, dup_r(r), s2, p, at, es)
44	    case Refuse{r}:
45	      refused(S, P, R, E, dup_r(r), p, at)
46	
47	def fresh(
48	  ~S: Type, ~P: Type, ~R: Type, ~E: Type, ~dup_r: R -> R & R, ~decide: S -> P -> Decision<S, R, E>,
49	  pp: P & P, s: S, at: U32
50	) -> Step<S, P, R, E>:
51	  (p, p2) = pp
52	  decided(~S, ~P, ~R, ~E, ~dup_r, decide(s, p), p2, at)
53	
54	def replayed(-R: Type, same: Bool, o: Outcome<R>) -> Reply<R>:
55	  match same:
56	    case True{}:
57	      Replayed{o}
58	    case False{}:
59	      Conflict{}
60	
61	# One command, given the record that the store holds for its id and the domain state s.
62	# dup_p and dup_r copy a payload and a result: a new command's payload goes to decide and to the record.
63	def step(
64	  ~S: Type, ~P: Type, ~R: Type, ~E: Type, ~eq: P -> P -> Bool, ~dup_p: P -> P & P, ~dup_r: R -> R & R,
65	  ~decide: S -> P -> Decision<S, R, E>, found: Maybe<Record<P, R>>, s: S, p: P, at: U32
66	) -> Step<S, P, R, E>:
67	  match found:
68	    case None{}:
69	      fresh(~S, ~P, ~R, ~E, ~dup_r, ~decide, dup_p(p), s, at)
70	    case Some{Record{q, o, t}}:
71	      Keep{replayed(R, eq(q, p), o)}
72	
73	# ---- IO that can fail with a SQLite error
74	
75	def Run(-A: Type) -> Type:
76	  IO(Result<&1, &1, U32 & String, A>)
77	
78	def Run.pure(-A: Type, x: A) -> Run(A):
79	  IO.pure(Result<&1, &1, U32 & String, A>, Done{x})
80	
81	def Run.fail(-A: Type, e: U32 & String) -> Run(A):
82	  IO.pure(Result<&1, &1, U32 & String, A>, Fail{e})
83	
84	def Run.next(-A: Type, -B: Type, r: Result<&1, &1, U32 & String, A>, f: A -> Run(B)) -> Run(B):
85	  match r:
86	    case Done{x}:
87	      f(x)
88	    case Fail{e}:
89	      Run.fail(B, e)
90	
91	def Run.bind(-A: Type, -B: Type, m: Run(A), f: A -> Run(B)) -> Run(B):
92	  IO.bind(Result<&1, &1, U32 & String, A>, Result<&1, &1, U32 & String, B>, m, r => Run.next(A, B, r, f))
93	
94	# r's value, unless r or then z failed.
95	def first(-A: Type, r: Result<&1, &1, U32 & String, A>, z: Result<&1, &1, U32 & String, Unit>) -> Result<&1, &1, U32 & String, A>:
96	  match r z:
97	    case Fail{e} _:
98	      Fail{e}
99	    case Done{x} Fail{e}:
100	      Fail{e}
101	    case Done{x} Done{u}:
102	      Done{x}
103	
104	def unit(-A: Type, r: Result<&1, &1, U32 & String, A>) -> Result<&1, &1, U32 & String, Unit>:
105	  match r:
106	    case Fail{e}:
107	      Fail{e}
108	    case Done{x}:
109	      Done{Unit{}}
110	
111	# ---- SQLite helpers
112	
113	# Steps a statement and ignores any row, then resets it, also after a failed step.
114	def exec(+stmt: U32) -> Run(Unit):
115	  do IO<Result<&1, &1, U32 & String, Unit>>:
116	    r : Result<&1, &1, U32 & String, Bool> <- Sqlite.stmt.step(stmt)
117	    z : Result<&1, &1, U32 & String, Unit> <- Sqlite.stmt.reset(stmt)
118	    return first(Unit, unit(Bool, r), z)
119	
120	def once.run(-A: Type, +s: U32, body: U32 -> Run(A)) -> Run(A):
121	  do IO<Result<&1, &1, U32 & String, A>>:
122	    r : Result<&1, &1, U32 & String, A> <- body(s)
123	    z : Result<&1, &1, U32 & String, Unit> <- Sqlite.stmt.finalize(s)
124	    return first(A, r, z)
125	
126	# Prepares one statement, runs body on it, and finalizes it, also after body fails.
127	def once(-A: Type, +db: U32, text: String, body: U32 -> Run(A)) -> Run(A):
128	  do Run<A>:
129	    s : U32 <- Sqlite.stmt.prepare(db, text)
130	    once.run(A, s, body)
131	
132	def sql(+db: U32, text: String) -> Run(Unit):
133	  once(Unit, db, text, +s => exec(s))
134	
135	def blob.of(lw: U32 & Array<U32>) -> Bytes.Bytes:
136	  (n, w) = lw
137	  Bytes.Bytes{n, w}
138	
139	def column(+s: U32, i: U32) -> Run(Bytes.Bytes):
140	  do Run<Bytes.Bytes>:
141	    lw : U32 & Array<U32> <- Sqlite.stmt.column.blob(s, i)
142	    return blob.of(lw)
143	
144	def bind(+s: U32, i: U32, b: Bytes.Bytes) -> Run(Unit):
145	  Bytes.Bytes{n, w} = b
146	  Sqlite.stmt.bind.blob(s, i, n, w)
147	
148	def same.of(r: Bytes.Bytes & Bytes.Bytes & Bool) -> Bool:
149	  (a, b, ok) = r
150	  ok
151	
152	# The payload comparison that handle gives step: every byte, NUL included.
153	def same(a: Bytes.Bytes, b: Bytes.Bytes) -> Bool:
154	  same.of(Bytes.eq(a, b))
155	
156	def dup.of(+n: U32, ww: Array<U32> & Array<U32>) -> Bytes.Bytes & Bytes.Bytes:
157	  (w, w2) = ww
158	  (Bytes.Bytes{n, w}, Bytes.Bytes{n, w2})
159	
160	def dup(b: Bytes.Bytes) -> Bytes.Bytes & Bytes.Bytes:
161	  Bytes.Bytes{+n, w} = b
162	  dup.of(n, Array.clone(U32, w))
163	
164	# ---- the store
165	
166	# A database and its prepared statements. One Store serves one task at a time.
167	type Store is Data:
168	  Store{db: U32, begin: U32, commit: U32, rollback: U32, find: U32, read: U32, save: U32, keep: U32, emit: U32}
169	
170	def init.run(+s: U32, init: Bytes.Bytes) -> Run(Unit):
171	  do Run<Unit>:
172	    bind(s, 1, init)
173	    exec(s)
174	
175	# Opens or creates the tables at path. A new database starts with domain state init.
176	# wait is how many ms a write waits for another connection's transaction before it fails with SQLITE_BUSY (5).
177	def open(path: String, init: Bytes.Bytes, wait: U32) -> Run(Store):
178	  do Run<Store>:
179	    +db : U32 <- Sqlite.db.open(path)
180	    sql(db, "PRAGMA busy_timeout = " ++ U32.show(wait))
181	    sql(db, "CREATE TABLE IF NOT EXISTS commands_state (k INTEGER PRIMARY KEY CHECK (k = 0), state BLOB NOT NULL)")
182	    sql(db, "CREATE TABLE IF NOT EXISTS commands_seen (id TEXT PRIMARY KEY, payload BLOB NOT NULL, kind INTEGER NOT NULL, result BLOB NOT NULL, at INTEGER NOT NULL)")
183	    sql(db, "CREATE INDEX IF NOT EXISTS commands_seen_at ON commands_seen (at)")
184	    sql(db, "CREATE TABLE IF NOT EXISTS commands_outbox (seq INTEGER PRIMARY KEY AUTOINCREMENT, id TEXT NOT NULL, effect BLOB NOT NULL)")
185	    once(Unit, db, "INSERT OR IGNORE INTO commands_state (k, state) VALUES (0, ?)", +s => init.run(s, init))
186	    begin : U32 <- Sqlite.stmt.prepare(db, "BEGIN IMMEDIATE")
187	    commit : U32 <- Sqlite.stmt.prepare(db, "COMMIT")
188	    rollback : U32 <- Sqlite.stmt.prepare(db, "ROLLBACK")
189	    find : U32 <- Sqlite.stmt.prepare(db, "SELECT payload, kind, result, at FROM commands_seen WHERE id = ?")
190	    read : U32 <- Sqlite.stmt.prepare(db, "SELECT state FROM commands_state WHERE k = 0")
191	    save : U32 <- Sqlite.stmt.prepare(db, "UPDATE commands_state SET state = ? WHERE k = 0")
192	    keep : U32 <- Sqlite.stmt.prepare(db, "INSERT INTO commands_seen (id, payload, kind, result, at) VALUES (?, ?, ?, ?, ?)")
193	    emit : U32 <- Sqlite.stmt.prepare(db, "INSERT INTO commands_outbox (id, effect) VALUES (?, ?)")
194	    return Store{db, begin, commit, rollback, find, read, save, keep, emit}
195	
196	def close(st: Store) -> Run(Unit):
197	  Store{db, begin, commit, rollback, find, read, save, keep, emit} = st
198	  do Run<Unit>:
199	    Sqlite.stmt.finalize(begin)
200	    Sqlite.stmt.finalize(commit)
201	    Sqlite.stmt.finalize(rollback)
202	    Sqlite.stmt.finalize(find)
203	    Sqlite.stmt.finalize(read)
204	    Sqlite.stmt.finalize(save)
205	    Sqlite.stmt.finalize(keep)
206	    Sqlite.stmt.finalize(emit)
207	    Sqlite.db.close(db)
208	
209	def outcome(accepted: Bool, r: Bytes.Bytes) -> Outcome<Bytes.Bytes>:
210	  match accepted:
211	    case True{}:
212	      Accepted{r}
213	    case False{}:
214	      Refused{r}
215	
216	def seen.row(row: Bool, +find: U32) -> Run(Maybe<Record<Bytes.Bytes, Bytes.Bytes>>):
217	  match row:
218	    case False{}:
219	      Run.pure(Maybe<Record<Bytes.Bytes, Bytes.Bytes>>, None{})
220	    case True{}:
221	      do Run<Maybe<Record<Bytes.Bytes, Bytes.Bytes>>>:
222	        p : Bytes.Bytes <- column(find, 0)
223	        kind : U32 <- Sqlite.stmt.column.int(find, 1)
224	        r : Bytes.Bytes <- column(find, 2)
225	        at : U32 <- Sqlite.stmt.column.int(find, 3)
226	        return Some{Record{p, outcome(U32.is_eq(kind, 1), r), at}}
227	
228	# The record for id, if the store holds one.
229	def seen.get(+find: U32, id: String) -> Run(Maybe<Record<Bytes.Bytes, Bytes.Bytes>>):
230	  do Run<Maybe<Record<Bytes.Bytes, Bytes.Bytes>>>:
231	    Sqlite.stmt.bind.text(find, 1, id)
232	    row : Bool <- Sqlite.stmt.step(find)
233	    m : Maybe<Record<Bytes.Bytes, Bytes.Bytes>> <- seen.row(row, find)
234	    Sqlite.stmt.reset(find)
235	    return m
236	
237	def state.row(row: Bool, +read: U32) -> Run(Bytes.Bytes):
238	  match row:
239	    case True{}:
240	      column(read, 0)
241	    case False{}:
242	      Run.fail(Bytes.Bytes, (1, "commands: the state row is missing"))
243	
244	def state.get(+read: U32) -> Run(Bytes.Bytes):
245	  do Run<Bytes.Bytes>:
246	    row : Bool <- Sqlite.stmt.step(read)
247	    s : Bytes.Bytes <- state.row(row, read)
248	    Sqlite.stmt.reset(read)
249	    return s
250	
251	def seen.put.row(+keep: U32, id: String, p: Bytes.Bytes, kind: U32, r: Bytes.Bytes, at: U32) -> Run(Unit):
252	  do Run<Unit>:
253	    Sqlite.stmt.bind.text(keep, 1, id)
254	    bind(keep, 2, p)
255	    Sqlite.stmt.bind.int(keep, 3, kind)
256	    bind(keep, 4, r)
257	    Sqlite.stmt.bind.int(keep, 5, at)
258	    exec(keep)
259	
260	def seen.put.of(o: Outcome<Bytes.Bytes>, +keep: U32, id: String, p: Bytes.Bytes, at: U32) -> Run(Unit):
261	  match o:
262	    case Accepted{r}:
263	      seen.put.row(keep, id, p, 1, r, at)
264	    case Refused{r}:
265	      seen.put.row(keep, id, p, 2, r, at)
266	
267	def seen.put(rec: Record<Bytes.Bytes, Bytes.Bytes>, +keep: U32, id: String) -> Run(Unit):
268	  Record{p, o, at} = rec
269	  seen.put.of(o, keep, id, p, at)
270	
271	def outbox.put(es: List<Bytes.Bytes>, +emit: U32, +id: String) -> Run(Unit):
272	  match es:
273	    case Nil{}:
274	      Run.pure(Unit, Unit{})
275	    case Con{e, rest}:
276	      do Run<Unit>:
277	        Sqlite.stmt.bind.text(emit, 1, id)
278	        bind(emit, 2, e)
279	        exec(emit)
280	        outbox.put(rest, emit, id)
281	
282	def state.put(+save: U32, s: Bytes.Bytes) -> Run(Unit):
283	  do Run<Unit>:
284	    bind(save, 1, s)
285	    exec(save)
286	
287	# Applies a step inside the open transaction, then commits it.
288	def written(
289	  w: Step<Bytes.Bytes, Bytes.Bytes, Bytes.Bytes, Bytes.Bytes>, +commit: U32, +save: U32, +keep: U32, +emit: U32, +id: String
290	) -> Run(Reply<Bytes.Bytes>):
291	  match w:
292	    case Write{s, rec, reply, es}:
293	      do Run<Reply<Bytes.Bytes>>:
294	        state.put(save, s)
295	        seen.put(rec, keep, id)
296	        outbox.put(es, emit, id)
297	        exec(commit)
298	        return reply
299	    case Note{rec, reply}:
300	      do Run<Reply<Bytes.Bytes>>:
301	        seen.put(rec, keep, id)
302	        exec(commit)
303	        return reply
304	    case Keep{reply}:
305	      do Run<Reply<Bytes.Bytes>>:
306	        exec(commit)
307	        return reply
308	
309	def handle.txn(
310	  ~decide: Bytes.Bytes -> Bytes.Bytes -> Decision<Bytes.Bytes, Bytes.Bytes, Bytes.Bytes>,
311	  st: Store, +id: String, payload: Bytes.Bytes, at: U32
312	) -> Run(Reply<Bytes.Bytes>):
313	  Store{db, begin, commit, rollback, find, read, save, keep, emit} = st
314	  do Run<Reply<Bytes.Bytes>>:
315	    exec(begin)
316	    found : Maybe<Record<Bytes.Bytes, Bytes.Bytes>> <- seen.get(find, id)
317	    s : Bytes.Bytes <- state.get(read)
318	    written(step(~Bytes.Bytes, ~Bytes.Bytes, ~Bytes.Bytes, ~Bytes.Bytes, ~same, ~dup, ~dup, ~decide, found, s, payload, at), commit, save, keep, emit, id)
319	
320	# On failure, resets the read statements and rolls back. Rollback errors are ignored: the transaction may not have started.
321	def handle.end(r: Result<&1, &1, U32 & String, Reply<Bytes.Bytes>>, st: Store) -> Run(Reply<Bytes.Bytes>):
322	  match r:
323	    case Done{x}:
324	      Run.pure(Reply<Bytes.Bytes>, x)
325	    case Fail{e}:
326	      Store{db, begin, commit, rollback, find, read, save, keep, emit} = st
327	      do IO<Result<&1, &1, U32 & String, Reply<Bytes.Bytes>>>:
328	        a : Result<&1, &1, U32 & String, Unit> <- Sqlite.stmt.reset(find)
329	        b : Result<&1, &1, U32 & String, Unit> <- Sqlite.stmt.reset(read)
330	        c : Result<&1, &1, U32 & String, Unit> <- exec(rollback)
331	        return Fail{e}
332	
333	# Runs one command in one transaction: find its record, decide, and commit state, record, and effects.
334	# at is the record's time for prune, in seconds. decide sees the committed state and the payload.
335	def handle(
336	  ~decide: Bytes.Bytes -> Bytes.Bytes -> Decision<Bytes.Bytes, Bytes.Bytes, Bytes.Bytes>,
337	  +st: Store, id: String, payload: Bytes.Bytes, at: U32
338	) -> Run(Reply<Bytes.Bytes>):
339	  do IO<Result<&1, &1, U32 & String, Reply<Bytes.Bytes>>>:
340	    r : Result<&1, &1, U32 & String, Reply<Bytes.Bytes>> <- handle.txn(~decide, st, id, payload, at)
341	    handle.end(r, st)
342	
343	# The committed domain state.
344	def state(st: Store) -> Run(Bytes.Bytes):
345	  Store{db, begin, commit, rollback, find, read, save, keep, emit} = st
346	  state.get(read)
347	
348	def prune.run(+s: U32, before: U32) -> Run(Unit):
349	  do Run<Unit>:
350	    Sqlite.stmt.bind.int(s, 1, before)
351	    exec(s)
352	
353	# Forgets every record made before the time before. A later retry of a forgotten id runs as a new command.
354	def prune(st: Store, before: U32) -> Run(Unit):
355	  Store{db, begin, commit, rollback, find, read, save, keep, emit} = st
356	  once(Unit, db, "DELETE FROM commands_seen WHERE at < ?", +s => prune.run(s, before))
357	
358	# An undelivered effect: its sequence number, its command id, and the effect.
359	type Pending is Type:
360	  Pending{seq: U32, id: String, effect: Bytes.Bytes}
361	
362	def outbox.row(row: Bool, +s: U32) -> Run(Maybe<Pending>):
363	  match row:
364	    case False{}:
365	      Run.pure(Maybe<Pending>, None{})
366	    case True{}:
367	      do Run<Maybe<Pending>>:
368	        seq : U32 <- Sqlite.stmt.column.int(s, 0)
369	        id : String <- Sqlite.stmt.column.text(s, 1)
370	        e : Bytes.Bytes <- column(s, 2)
371	        return Some{Pending{seq, id, e}}
372	
373	def outbox.first(+s: U32) -> Run(Maybe<Pending>):
374	  do Run<Maybe<Pending>>:
375	    row : Bool <- Sqlite.stmt.step(s)
376	    outbox.row(row, s)
377	
378	# The oldest undelivered effect.
379	def outbox.next(st: Store) -> Run(Maybe<Pending>):
380	  Store{db, begin, commit, rollback, find, read, save, keep, emit} = st
381	  once(Maybe<Pending>, db, "SELECT seq, id, effect FROM commands_outbox ORDER BY seq LIMIT 1", +s => outbox.first(s))
382	
383	def outbox.ack.run(+s: U32, seq: U32) -> Run(Unit):
384	  do Run<Unit>:
385	    Sqlite.stmt.bind.int(s, 1, seq)
386	    exec(s)
387	
388	# Removes a delivered effect. Deliver before ack: a crash between them delivers the effect again.
389	def outbox.ack(st: Store, seq: U32) -> Run(Unit):
390	  Store{db, begin, commit, rollback, find, read, save, keep, emit} = st
391	  once(Unit, db, "DELETE FROM commands_outbox WHERE seq = ?", +s => outbox.ack.run(s, seq))