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