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.3.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	# ---- the store
157	
158	# A database and its prepared statements. One Store serves one task at a time.
159	type Store is Data:
160	  Store{db: U32, begin: U32, commit: U32, rollback: U32, find: U32, read: U32, save: U32, keep: U32, emit: U32}
161	
162	def init.run(+s: U32, init: Bytes.Bytes) -> Run(Unit):
163	  do Run<Unit>:
164	    bind(s, 1, init)
165	    exec(s)
166	
167	# Opens or creates the tables at path. A new database starts with domain state init.
168	# wait is how many ms a write waits for another connection's transaction before it fails with SQLITE_BUSY (5).
169	def open(path: String, init: Bytes.Bytes, wait: U32) -> Run(Store):
170	  do Run<Store>:
171	    +db : U32 <- Sqlite.db.open(path)
172	    sql(db, "PRAGMA busy_timeout = " ++ U32.show(wait))
173	    sql(db, "CREATE TABLE IF NOT EXISTS commands_state (k INTEGER PRIMARY KEY CHECK (k = 0), state BLOB NOT NULL)")
174	    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)")
175	    sql(db, "CREATE INDEX IF NOT EXISTS commands_seen_at ON commands_seen (at)")
176	    sql(db, "CREATE TABLE IF NOT EXISTS commands_outbox (seq INTEGER PRIMARY KEY AUTOINCREMENT, id TEXT NOT NULL, effect BLOB NOT NULL)")
177	    once(Unit, db, "INSERT OR IGNORE INTO commands_state (k, state) VALUES (0, ?)", +s => init.run(s, init))
178	    begin : U32 <- Sqlite.stmt.prepare(db, "BEGIN IMMEDIATE")
179	    commit : U32 <- Sqlite.stmt.prepare(db, "COMMIT")
180	    rollback : U32 <- Sqlite.stmt.prepare(db, "ROLLBACK")
181	    find : U32 <- Sqlite.stmt.prepare(db, "SELECT payload, kind, result, at FROM commands_seen WHERE id = ?")
182	    read : U32 <- Sqlite.stmt.prepare(db, "SELECT state FROM commands_state WHERE k = 0")
183	    save : U32 <- Sqlite.stmt.prepare(db, "UPDATE commands_state SET state = ? WHERE k = 0")
184	    keep : U32 <- Sqlite.stmt.prepare(db, "INSERT INTO commands_seen (id, payload, kind, result, at) VALUES (?, ?, ?, ?, ?)")
185	    emit : U32 <- Sqlite.stmt.prepare(db, "INSERT INTO commands_outbox (id, effect) VALUES (?, ?)")
186	    return Store{db, begin, commit, rollback, find, read, save, keep, emit}
187	
188	def close(st: Store) -> Run(Unit):
189	  Store{db, begin, commit, rollback, find, read, save, keep, emit} = st
190	  do Run<Unit>:
191	    Sqlite.stmt.finalize(begin)
192	    Sqlite.stmt.finalize(commit)
193	    Sqlite.stmt.finalize(rollback)
194	    Sqlite.stmt.finalize(find)
195	    Sqlite.stmt.finalize(read)
196	    Sqlite.stmt.finalize(save)
197	    Sqlite.stmt.finalize(keep)
198	    Sqlite.stmt.finalize(emit)
199	    Sqlite.db.close(db)
200	
201	def outcome(accepted: Bool, r: Bytes.Bytes) -> Outcome<Bytes.Bytes>:
202	  match accepted:
203	    case True{}:
204	      Accepted{r}
205	    case False{}:
206	      Refused{r}
207	
208	def seen.row(row: Bool, +find: U32) -> Run(Maybe<Record<Bytes.Bytes, Bytes.Bytes>>):
209	  match row:
210	    case False{}:
211	      Run.pure(Maybe<Record<Bytes.Bytes, Bytes.Bytes>>, None{})
212	    case True{}:
213	      do Run<Maybe<Record<Bytes.Bytes, Bytes.Bytes>>>:
214	        p : Bytes.Bytes <- column(find, 0)
215	        kind : U32 <- Sqlite.stmt.column.int(find, 1)
216	        r : Bytes.Bytes <- column(find, 2)
217	        at : U32 <- Sqlite.stmt.column.int(find, 3)
218	        return Some{Record{p, outcome(U32.is_eq(kind, 1), r), at}}
219	
220	# The record for id, if the store holds one.
221	def seen.get(+find: U32, id: String) -> Run(Maybe<Record<Bytes.Bytes, Bytes.Bytes>>):
222	  do Run<Maybe<Record<Bytes.Bytes, Bytes.Bytes>>>:
223	    Sqlite.stmt.bind.text(find, 1, id)
224	    row : Bool <- Sqlite.stmt.step(find)
225	    m : Maybe<Record<Bytes.Bytes, Bytes.Bytes>> <- seen.row(row, find)
226	    Sqlite.stmt.reset(find)
227	    return m
228	
229	def state.row(row: Bool, +read: U32) -> Run(Bytes.Bytes):
230	  match row:
231	    case True{}:
232	      column(read, 0)
233	    case False{}:
234	      Run.fail(Bytes.Bytes, (1, "commands: the state row is missing"))
235	
236	def state.get(+read: U32) -> Run(Bytes.Bytes):
237	  do Run<Bytes.Bytes>:
238	    row : Bool <- Sqlite.stmt.step(read)
239	    s : Bytes.Bytes <- state.row(row, read)
240	    Sqlite.stmt.reset(read)
241	    return s
242	
243	def seen.put.row(+keep: U32, id: String, p: Bytes.Bytes, kind: U32, r: Bytes.Bytes, at: U32) -> Run(Unit):
244	  do Run<Unit>:
245	    Sqlite.stmt.bind.text(keep, 1, id)
246	    bind(keep, 2, p)
247	    Sqlite.stmt.bind.int(keep, 3, kind)
248	    bind(keep, 4, r)
249	    Sqlite.stmt.bind.int(keep, 5, at)
250	    exec(keep)
251	
252	def seen.put.of(o: Outcome<Bytes.Bytes>, +keep: U32, id: String, p: Bytes.Bytes, at: U32) -> Run(Unit):
253	  match o:
254	    case Accepted{r}:
255	      seen.put.row(keep, id, p, 1, r, at)
256	    case Refused{r}:
257	      seen.put.row(keep, id, p, 2, r, at)
258	
259	def seen.put(rec: Record<Bytes.Bytes, Bytes.Bytes>, +keep: U32, id: String) -> Run(Unit):
260	  Record{p, o, at} = rec
261	  seen.put.of(o, keep, id, p, at)
262	
263	def outbox.put(es: List<Bytes.Bytes>, +emit: U32, +id: String) -> Run(Unit):
264	  match es:
265	    case Nil{}:
266	      Run.pure(Unit, Unit{})
267	    case Con{e, rest}:
268	      do Run<Unit>:
269	        Sqlite.stmt.bind.text(emit, 1, id)
270	        bind(emit, 2, e)
271	        exec(emit)
272	        outbox.put(rest, emit, id)
273	
274	def state.put(+save: U32, s: Bytes.Bytes) -> Run(Unit):
275	  do Run<Unit>:
276	    bind(save, 1, s)
277	    exec(save)
278	
279	# Applies a step inside the open transaction, then commits it.
280	def written(
281	  w: Step<Bytes.Bytes, Bytes.Bytes, Bytes.Bytes, Bytes.Bytes>, +commit: U32, +save: U32, +keep: U32, +emit: U32, +id: String
282	) -> Run(Reply<Bytes.Bytes>):
283	  match w:
284	    case Write{s, rec, reply, es}:
285	      do Run<Reply<Bytes.Bytes>>:
286	        state.put(save, s)
287	        seen.put(rec, keep, id)
288	        outbox.put(es, emit, id)
289	        exec(commit)
290	        return reply
291	    case Note{rec, reply}:
292	      do Run<Reply<Bytes.Bytes>>:
293	        seen.put(rec, keep, id)
294	        exec(commit)
295	        return reply
296	    case Keep{reply}:
297	      do Run<Reply<Bytes.Bytes>>:
298	        exec(commit)
299	        return reply
300	
301	# A duplicate needs no state: step ignores s when found is Some, so it gets an empty one.
302	def handle.found(
303	  ~decide: Bytes.Bytes -> Bytes.Bytes -> Decision<Bytes.Bytes, Bytes.Bytes, Bytes.Bytes>,
304	  found: Maybe<Record<Bytes.Bytes, Bytes.Bytes>>, +read: U32, payload: Bytes.Bytes, at: U32,
305	  +commit: U32, +save: U32, +keep: U32, +emit: U32, +id: String
306	) -> Run(Reply<Bytes.Bytes>):
307	  match found:
308	    case None{}:
309	      do Run<Reply<Bytes.Bytes>>:
310	        s : Bytes.Bytes <- state.get(read)
311	        written(step(~Bytes.Bytes, ~Bytes.Bytes, ~Bytes.Bytes, ~Bytes.Bytes, ~same, ~Bytes.clone, ~Bytes.clone, ~decide, None{}, s, payload, at), commit, save, keep, emit, id)
312	    case Some{rec}:
313	      written(step(~Bytes.Bytes, ~Bytes.Bytes, ~Bytes.Bytes, ~Bytes.Bytes, ~same, ~Bytes.clone, ~Bytes.clone, ~decide, Some{rec}, Bytes.new(0), payload, at), commit, save, keep, emit, id)
314	
315	def handle.txn(
316	  ~decide: Bytes.Bytes -> Bytes.Bytes -> Decision<Bytes.Bytes, Bytes.Bytes, Bytes.Bytes>,
317	  st: Store, +id: String, payload: Bytes.Bytes, at: U32
318	) -> Run(Reply<Bytes.Bytes>):
319	  Store{db, begin, commit, rollback, find, read, save, keep, emit} = st
320	  do Run<Reply<Bytes.Bytes>>:
321	    exec(begin)
322	    found : Maybe<Record<Bytes.Bytes, Bytes.Bytes>> <- seen.get(find, id)
323	    handle.found(~decide, found, read, payload, at, commit, save, keep, emit, id)
324	
325	# On failure, resets the read statements and rolls back. Rollback errors are ignored: the transaction may not have started.
326	def handle.end(r: Result<&1, &1, U32 & String, Reply<Bytes.Bytes>>, st: Store) -> Run(Reply<Bytes.Bytes>):
327	  match r:
328	    case Done{x}:
329	      Run.pure(Reply<Bytes.Bytes>, x)
330	    case Fail{e}:
331	      Store{db, begin, commit, rollback, find, read, save, keep, emit} = st
332	      do IO<Result<&1, &1, U32 & String, Reply<Bytes.Bytes>>>:
333	        a : Result<&1, &1, U32 & String, Unit> <- Sqlite.stmt.reset(find)
334	        b : Result<&1, &1, U32 & String, Unit> <- Sqlite.stmt.reset(read)
335	        c : Result<&1, &1, U32 & String, Unit> <- exec(rollback)
336	        return Fail{e}
337	
338	# Runs one command in one transaction: find its record, decide, and commit state, record, and effects.
339	# at is the record's time for prune, in seconds. decide sees the committed state and the payload.
340	def handle(
341	  ~decide: Bytes.Bytes -> Bytes.Bytes -> Decision<Bytes.Bytes, Bytes.Bytes, Bytes.Bytes>,
342	  +st: Store, id: String, payload: Bytes.Bytes, at: U32
343	) -> Run(Reply<Bytes.Bytes>):
344	  do IO<Result<&1, &1, U32 & String, Reply<Bytes.Bytes>>>:
345	    r : Result<&1, &1, U32 & String, Reply<Bytes.Bytes>> <- handle.txn(~decide, st, id, payload, at)
346	    handle.end(r, st)
347	
348	# The committed domain state.
349	def state(st: Store) -> Run(Bytes.Bytes):
350	  Store{db, begin, commit, rollback, find, read, save, keep, emit} = st
351	  state.get(read)
352	
353	def prune.run(+s: U32, before: U32) -> Run(Unit):
354	  do Run<Unit>:
355	    Sqlite.stmt.bind.int(s, 1, before)
356	    exec(s)
357	
358	# Forgets every record made before the time before. A later retry of a forgotten id runs as a new command.
359	def prune(st: Store, before: U32) -> Run(Unit):
360	  Store{db, begin, commit, rollback, find, read, save, keep, emit} = st
361	  once(Unit, db, "DELETE FROM commands_seen WHERE at < ?", +s => prune.run(s, before))
362	
363	# An undelivered effect: its sequence number, its command id, and the effect.
364	type Pending is Type:
365	  Pending{seq: U32, id: String, effect: Bytes.Bytes}
366	
367	def outbox.row(row: Bool, +s: U32) -> Run(Maybe<Pending>):
368	  match row:
369	    case False{}:
370	      Run.pure(Maybe<Pending>, None{})
371	    case True{}:
372	      do Run<Maybe<Pending>>:
373	        seq : U32 <- Sqlite.stmt.column.int(s, 0)
374	        id : String <- Sqlite.stmt.column.text(s, 1)
375	        e : Bytes.Bytes <- column(s, 2)
376	        return Some{Pending{seq, id, e}}
377	
378	def outbox.first(+s: U32) -> Run(Maybe<Pending>):
379	  do Run<Maybe<Pending>>:
380	    row : Bool <- Sqlite.stmt.step(s)
381	    outbox.row(row, s)
382	
383	# The oldest undelivered effect.
384	def outbox.next(st: Store) -> Run(Maybe<Pending>):
385	  Store{db, begin, commit, rollback, find, read, save, keep, emit} = st
386	  once(Maybe<Pending>, db, "SELECT seq, id, effect FROM commands_outbox ORDER BY seq LIMIT 1", +s => outbox.first(s))
387	
388	def outbox.ack.run(+s: U32, seq: U32) -> Run(Unit):
389	  do Run<Unit>:
390	    Sqlite.stmt.bind.int(s, 1, seq)
391	    exec(s)
392	
393	# Removes a delivered effect. Deliver before ack: a crash between them delivers the effect again.
394	def outbox.ack(st: Store, seq: U32) -> Run(Unit):
395	  Store{db, begin, commit, rollback, find, read, save, keep, emit} = st
396	  once(Unit, db, "DELETE FROM commands_outbox WHERE seq = ?", +s => outbox.ack.run(s, seq))