1	# Postgres frontend/backend protocol 3.0 messages over Bytes, and an incremental reader of backend messages.
2	import Base
3	import bend-kit-bytes@0.3.3.0/bytes.bend as Bytes
4	
5	# ---- messages
6	
7	# A StartupMessage parameter, such as ("user", name) or ("database", db).
8	type Param is Type:
9	  Param{name: Bytes.Bytes, value: Bytes.Bytes}
10	
11	# Frontend messages. Describe and CloseMsg take kind 83 ('S', statement) or 80 ('P', portal).
12	# Bind params hold one Maybe per parameter, None for SQL NULL; empty formats mean text.
13	type Front is Type:
14	  Startup{params: List<&1, Param>}
15	  SSLRequest{}
16	  Password{text: Bytes.Bytes}
17	  SASLInitial{mech: Bytes.Bytes, data: Bytes.Bytes}
18	  SASLResponse{data: Bytes.Bytes}
19	  Parse{name: Bytes.Bytes, query: Bytes.Bytes, oids: List<&2, U32>}
20	  Bind{portal: Bytes.Bytes, stmt: Bytes.Bytes, formats: List<&2, U32>, params: List<&1, Maybe<&1, Bytes.Bytes>>, results: List<&2, U32>}
21	  Describe{kind: U32, name: Bytes.Bytes}
22	  Execute{portal: Bytes.Bytes, max: U32}
23	  Sync{}
24	  CloseMsg{kind: U32, name: Bytes.Bytes}
25	  Terminate{}
26	
27	# A RowDescription column. size and modifier are the raw int16 and int32 words.
28	type Field is Type:
29	  Field{name: Bytes.Bytes, table: U32, column: U32, oid: U32, size: U32, modifier: U32, format: U32}
30	
31	# An ErrorResponse or NoticeResponse field: code is its type byte, such as 67 ('C', SQLSTATE) or 77 ('M', message).
32	type Notice is Type:
33	  Notice{code: U32, value: Bytes.Bytes}
34	
35	# Backend messages. AuthOther holds any other authentication code (MD5 is 5) and its payload.
36	# ReadyForQuery status is 73 ('I'), 84 ('T') or 69 ('E'). Other holds a well-framed message of any other tag.
37	type Back is Type:
38	  AuthOk{}
39	  AuthCleartext{}
40	  AuthSASL{mechs: List<&1, Bytes.Bytes>}
41	  AuthSASLContinue{data: Bytes.Bytes}
42	  AuthSASLFinal{data: Bytes.Bytes}
43	  AuthOther{code: U32, data: Bytes.Bytes}
44	  ParameterStatus{name: Bytes.Bytes, value: Bytes.Bytes}
45	  BackendKeyData{pid: U32, key: U32}
46	  ReadyForQuery{status: U32}
47	  RowDescription{fields: List<&1, Field>}
48	  DataRow{cols: List<&1, Maybe<&1, Bytes.Bytes>>}
49	  CommandComplete{tag: Bytes.Bytes}
50	  ErrorResponse{fields: List<&1, Notice>}
51	  NoticeResponse{fields: List<&1, Notice>}
52	  ParseComplete{}
53	  BindComplete{}
54	  CloseComplete{}
55	  NoData{}
56	  EmptyQuery{}
57	  PortalSuspended{}
58	  Other{tag: U32, body: Bytes.Bytes}
59	
60	# The first field whose code is code.
61	def notice.get(+code: U32, fs: List<&1, Notice>) -> Maybe<&1, Bytes.Bytes>:
62	  match fs:
63	    case Nil{}:
64	      None{}
65	    case Con{Notice{+k, v}, t}:
66	      Bool.pick(Maybe<&1, Bytes.Bytes>, U32.is_eq(k, code), Some{v}, notice.get(code, t))
67	
68	# ---- encoder
69	
70	def byte1(+c: U32) -> Bytes.Bytes:
71	  Bytes.set(Bytes.new(1), 0, c)
72	
73	def be16(+v: U32) -> Bytes.Bytes:
74	  Bytes.set.u16be(Bytes.new(2), 0, v)
75	
76	def be32(+v: U32) -> Bytes.Bytes:
77	  Bytes.set.u32be(Bytes.new(4), 0, v)
78	
79	# ponytail: a NUL inside a cstr field ends it early; the message length still frames it, so the server rejects that message only.
80	def cstr.enc(b: Bytes.Bytes) -> Bytes.Bytes:
81	  Bytes.append(b, Bytes.new(1))
82	
83	def frame.of(+tag: U32, r: Bytes.Bytes & U32) -> Bytes.Bytes:
84	  (body, +n) = r
85	  Bytes.concat([byte1(tag), be32((n + 4 : U32)), body])
86	
87	# A tagged message: the tag byte, the int32 length (counting itself), then the body.
88	def frame(+tag: U32, body: Bytes.Bytes) -> Bytes.Bytes:
89	  frame.of(tag, Bytes.length(body))
90	
91	def packet.of(r: Bytes.Bytes & U32) -> Bytes.Bytes:
92	  (body, +n) = r
93	  Bytes.append(be32((n + 4 : U32)), body)
94	
95	# An untagged startup-phase packet.
96	def packet(body: Bytes.Bytes) -> Bytes.Bytes:
97	  packet.of(Bytes.length(body))
98	
99	def n16.con(h: Bytes.Bytes, r: List<&1, Bytes.Bytes> & U32) -> List<&1, Bytes.Bytes> & U32:
100	  (t, +n) = r
101	  (Con{h, t}, (n + 1 : U32))
102	
103	def n16.count(xs: List<&1, Bytes.Bytes>) -> List<&1, Bytes.Bytes> & U32:
104	  match xs:
105	    case Nil{}:
106	      (Nil{}, 0)
107	    case Con{h, t}:
108	      n16.con(h, n16.count(t))
109	
110	def n16.of(r: List<&1, Bytes.Bytes> & U32) -> Bytes.Bytes:
111	  (xs, +n) = r
112	  Bytes.concat(Con{be16(n), xs})
113	
114	# An int16 count, then the pieces.
115	def n16(xs: List<&1, Bytes.Bytes>) -> Bytes.Bytes:
116	  n16.of(n16.count(xs))
117	
118	def be16s(xs: List<&2, U32>) -> List<&1, Bytes.Bytes>:
119	  match xs:
120	    case Nil{}:
121	      Nil{}
122	    case Con{h, t}:
123	      Con{be16(h), be16s(t)}
124	
125	def be32s(xs: List<&2, U32>) -> List<&1, Bytes.Bytes>:
126	  match xs:
127	    case Nil{}:
128	      Nil{}
129	    case Con{h, t}:
130	      Con{be32(h), be32s(t)}
131	
132	def value.of(r: Bytes.Bytes & U32) -> Bytes.Bytes:
133	  (b, +n) = r
134	  Bytes.append(be32(n), b)
135	
136	# An int32 length then the bytes; -1 and nothing for NULL.
137	def value(m: Maybe<&1, Bytes.Bytes>) -> Bytes.Bytes:
138	  match m:
139	    case None{}:
140	      be32(4294967295)
141	    case Some{b}:
142	      value.of(Bytes.length(b))
143	
144	def values(xs: List<&1, Maybe<&1, Bytes.Bytes>>) -> List<&1, Bytes.Bytes>:
145	  match xs:
146	    case Nil{}:
147	      Nil{}
148	    case Con{h, t}:
149	      Con{value(h), values(t)}
150	
151	def params.enc(xs: List<&1, Param>) -> List<&1, Bytes.Bytes>:
152	  match xs:
153	    case Nil{}:
154	      [byte1(0)]
155	    case Con{Param{n, v}, t}:
156	      Con{Bytes.append(cstr.enc(n), cstr.enc(v)), params.enc(t)}
157	
158	def names.enc(xs: List<&1, Bytes.Bytes>) -> List<&1, Bytes.Bytes>:
159	  match xs:
160	    case Nil{}:
161	      [byte1(0)]
162	    case Con{h, t}:
163	      Con{cstr.enc(h), names.enc(t)}
164	
165	def notices.enc(xs: List<&1, Notice>) -> List<&1, Bytes.Bytes>:
166	  match xs:
167	    case Nil{}:
168	      [byte1(0)]
169	    case Con{Notice{+k, v}, t}:
170	      Con{Bytes.append(byte1(k), cstr.enc(v)), notices.enc(t)}
171	
172	def fields.enc(xs: List<&1, Field>) -> List<&1, Bytes.Bytes>:
173	  match xs:
174	    case Nil{}:
175	      Nil{}
176	    case Con{Field{n, +tb, +col, +oid, +sz, +md, +fmt}, t}:
177	      Con{Bytes.concat([cstr.enc(n), be32(tb), be16(col), be32(oid), be16(sz), be32(md), be16(fmt)]), fields.enc(t)}
178	
179	# The wire bytes of a frontend message.
180	def encode(m: Front) -> Bytes.Bytes:
181	  match m:
182	    case Startup{ps}:
183	      packet(Bytes.concat(Con{be32(196608), params.enc(ps)}))
184	    case SSLRequest{}:
185	      packet(be32(80877103))
186	    case Password{t}:
187	      frame(112, cstr.enc(t))
188	    case SASLInitial{mech, d}:
189	      frame(112, Bytes.append(cstr.enc(mech), value(Some{d})))
190	    case SASLResponse{d}:
191	      frame(112, d)
192	    case Parse{n, q, oids}:
193	      frame(80, Bytes.concat([cstr.enc(n), cstr.enc(q), n16(be32s(oids))]))
194	    case Bind{p, s, fs, ps, rs}:
195	      frame(66, Bytes.concat([cstr.enc(p), cstr.enc(s), n16(be16s(fs)), n16(values(ps)), n16(be16s(rs))]))
196	    case Describe{+k, n}:
197	      frame(68, Bytes.append(byte1(k), cstr.enc(n)))
198	    case Execute{p, +max}:
199	      frame(69, Bytes.append(cstr.enc(p), be32(max)))
200	    case Sync{}:
201	      frame(83, Bytes.new(0))
202	    case CloseMsg{+k, n}:
203	      frame(67, Bytes.append(byte1(k), cstr.enc(n)))
204	    case Terminate{}:
205	      frame(88, Bytes.new(0))
206	
207	def auth.enc(+code: U32, payload: Bytes.Bytes) -> Bytes.Bytes:
208	  frame(82, Bytes.append(be32(code), payload))
209	
210	# The wire bytes of a backend message, as a server sends it.
211	def encode.back(m: Back) -> Bytes.Bytes:
212	  match m:
213	    case AuthOk{}:
214	      auth.enc(0, Bytes.new(0))
215	    case AuthCleartext{}:
216	      auth.enc(3, Bytes.new(0))
217	    case AuthSASL{ms}:
218	      auth.enc(10, Bytes.concat(names.enc(ms)))
219	    case AuthSASLContinue{d}:
220	      auth.enc(11, d)
221	    case AuthSASLFinal{d}:
222	      auth.enc(12, d)
223	    case AuthOther{+code, d}:
224	      auth.enc(code, d)
225	    case ParameterStatus{n, v}:
226	      frame(83, Bytes.append(cstr.enc(n), cstr.enc(v)))
227	    case BackendKeyData{+pid, +key}:
228	      frame(75, Bytes.append(be32(pid), be32(key)))
229	    case ReadyForQuery{+s}:
230	      frame(90, byte1(s))
231	    case RowDescription{fs}:
232	      frame(84, n16(fields.enc(fs)))
233	    case DataRow{cs}:
234	      frame(68, n16(values(cs)))
235	    case CommandComplete{t}:
236	      frame(67, cstr.enc(t))
237	    case ErrorResponse{fs}:
238	      frame(69, Bytes.concat(notices.enc(fs)))
239	    case NoticeResponse{fs}:
240	      frame(78, Bytes.concat(notices.enc(fs)))
241	    case ParseComplete{}:
242	      frame(49, Bytes.new(0))
243	    case BindComplete{}:
244	      frame(50, Bytes.new(0))
245	    case CloseComplete{}:
246	      frame(51, Bytes.new(0))
247	    case NoData{}:
248	      frame(110, Bytes.new(0))
249	    case EmptyQuery{}:
250	      frame(73, Bytes.new(0))
251	    case PortalSuspended{}:
252	      frame(115, Bytes.new(0))
253	    case Other{+tag, b}:
254	      frame(tag, b)
255	
256	# ---- body cursor
257	
258	# The buffer, the next byte i, the end e of the message body (e <= len), and whether every read so far fit.
259	# A read that does not fit clears ok and leaves i; the message is Bad unless ok holds and i reaches e.
260	type Cur is Type:
261	  Cur{b: Bytes.Bytes, i: U32, e: U32, ok: Bool}
262	
263	def fail(c: Cur) -> Cur:
264	  Cur{b, +i, +e, ok} = c
265	  Cur{b, i, e, False{}}
266	
267	def need(ok: Bool, c: Cur) -> Cur:
268	  match ok:
269	    case True{}:
270	      c
271	    case False{}:
272	      fail(c)
273	
274	def left(c: Cur) -> Cur & Nat:
275	  Cur{b, +i, +e, +ok} = c
276	  (Cur{b, i, e, ok}, U32.to_nat((e - i : U32)))
277	
278	def uint.got(+i: U32, +n: U32, +e: U32, +ok: Bool, r: Bytes.Bytes & Maybe<&2, U32>) -> Cur & U32:
279	  (b, m) = r
280	  match m:
281	    case Some{+v}:
282	      (Cur{b, (i + n : U32), e, ok}, v)
283	    case None{}:
284	      (Cur{b, i, e, False{}}, 0)
285	
286	def uint.at(fit: Bool, b: Bytes.Bytes, +i: U32, +n: U32, +e: U32, +ok: Bool) -> Cur & U32:
287	  match fit:
288	    case True{}:
289	      uint.got(i, n, e, ok, Bytes.uint(b, i, n, True{}))
290	    case False{}:
291	      (Cur{b, i, e, False{}}, 0)
292	
293	# A big-endian unsigned integer of n bytes (1, 2 or 4).
294	def uint(c: Cur, +n: U32) -> Cur & U32:
295	  Cur{b, +i, +e, +ok} = c
296	  uint.at(Bytes.fits(e, i, n), b, i, n, e, ok)
297	
298	def take.got(+i: U32, +e: U32, +ok: Bool, r: Bytes.Bytes & Bytes.Bytes) -> Cur & Bytes.Bytes:
299	  (b, s) = r
300	  (Cur{b, i, e, ok}, s)
301	
302	def take.at(fit: Bool, b: Bytes.Bytes, +i: U32, +n: U32, +e: U32, +ok: Bool) -> Cur & Bytes.Bytes:
303	  match fit:
304	    case True{}:
305	      take.got((i + n : U32), e, ok, Bytes.slice(b, i, n))
306	    case False{}:
307	      (Cur{b, i, e, False{}}, Bytes.new(0))
308	
309	# The next n bytes, copied out.
310	def take(c: Cur, +n: U32) -> Cur & Bytes.Bytes:
311	  Cur{b, +i, +e, +ok} = c
312	  take.at(Bytes.fits(e, i, n), b, i, n, e, ok)
313	
314	def rest(c: Cur) -> Cur & Bytes.Bytes:
315	  Cur{b, +i, +e, +ok} = c
316	  take(Cur{b, i, e, ok}, (e - i : U32))
317	
318	def nul() -> String:
319	  SCon{Chr{0}, SNil{}}
320	
321	def cstr.in(inside: Bool, b: Bytes.Bytes, +i: U32, +j: U32, +e: U32, +ok: Bool) -> Cur & Bytes.Bytes:
322	  match inside:
323	    case True{}:
324	      take.got((j + 1 : U32), e, ok, Bytes.slice(b, i, (j - i : U32)))
325	    case False{}:
326	      (Cur{b, i, e, False{}}, Bytes.new(0))
327	
328	def cstr.at(+i: U32, +e: U32, +ok: Bool, r: Bytes.Bytes & Maybe<&2, U32>) -> Cur & Bytes.Bytes:
329	  (b, m) = r
330	  match m:
331	    case Some{+j}:
332	      cstr.in(U32.is_lt(j, e), b, i, j, e, ok)
333	    case None{}:
334	      (Cur{b, i, e, False{}}, Bytes.new(0))
335	
336	# The bytes before the next NUL inside the body, and the cursor past it.
337	def cstr(c: Cur) -> Cur & Bytes.Bytes:
338	  Cur{b, +i, +e, +ok} = c
339	  cstr.at(i, e, ok, Bytes.find.from(b, nul(), i))
340	
341	def u16s(c: Cur) -> Cur & U32:
342	  uint(c, 2)
343	
344	def u32s(c: Cur) -> Cur & U32:
345	  uint(c, 4)
346	
347	# n more items after the one in r, oldest last in acc.
348	def many(~A: Type, ~item: Cur -> Cur & A, n: Nat, r: Cur & A, acc: List<&1, A>) -> Cur & List<&1, A>:
349	  match n:
350	    case 0n:
351	      (c, v) = r
352	      (c, List.reverse(&1, A, Con{v, acc}))
353	    case 1n+p:
354	      (c, v) = r
355	      many(~A, ~item, p, item(c), Con{v, acc})
356	
357	def n16.go(~A: Type, ~item: Cur -> Cur & A, zero: Bool, +k: U32, c: Cur) -> Cur & List<&1, A>:
358	  match zero:
359	    case True{}:
360	      (c, Nil{})
361	    case False{}:
362	      many(~A, ~item, U32.to_nat((k - 1 : U32)), item(c), Nil{})
363	
364	# An int16 count, then that many items.
365	def n16.dec(~A: Type, ~item: Cur -> Cur & A, r: Cur & U32) -> Cur & List<&1, A>:
366	  (c, +k) = r
367	  n16.go(~A, ~item, U32.is_eq(k, 0), k, c)
368	
369	def list16(~A: Type, ~item: Cur -> Cur & A, c: Cur) -> Cur & List<&1, A>:
370	  n16.dec(~A, ~item, uint(c, 2))
371	
372	def dup(xs: List<&1, U32>) -> List<&2, U32>:
373	  match xs:
374	    case Nil{}:
375	      Nil{}
376	    case Con{+h, t}:
377	      Con{h, dup(t)}
378	
379	def col.some(r: Cur & Bytes.Bytes) -> Cur & Maybe<&1, Bytes.Bytes>:
380	  (c, b) = r
381	  (c, Some{b})
382	
383	def col.of(null: Bool, +n: U32, c: Cur) -> Cur & Maybe<&1, Bytes.Bytes>:
384	  match null:
385	    case True{}:
386	      (c, None{})
387	    case False{}:
388	      col.some(take(c, n))
389	
390	def col.len(r: Cur & U32) -> Cur & Maybe<&1, Bytes.Bytes>:
391	  (c, +n) = r
392	  col.of(U32.is_eq(n, 4294967295), n, c)
393	
394	# An int32 length then the bytes; -1 is NULL, and any other negative length does not fit.
395	def col(c: Cur) -> Cur & Maybe<&1, Bytes.Bytes>:
396	  col.len(uint(c, 4))
397	
398	# A loop up to a terminator: Go holds an item still to push, Stop the end. acc is reversed.
399	type Loop<-A: Type, -B: Type> is Type:
400	  Go{c: Cur, x: A, acc: List<&1, B>}
401	  Stop{c: Cur, acc: List<&1, B>}
402	
403	# Out of fuel before the terminator is Bad.
404	def loop.end(-A: Type, -B: Type, s: Loop<A, B>) -> Cur & List<&1, B>:
405	  match s:
406	    case Go{c, x, acc}:
407	      (fail(c), List.reverse(&1, B, acc))
408	    case Stop{c, acc}:
409	      (c, List.reverse(&1, B, acc))
410	
411	def loop.pick(-A: Type, -B: Type, stop: Bool, c: Cur, x: A, acc: List<&1, B>) -> Loop<A, B>:
412	  match stop:
413	    case True{}:
414	      Stop{c, acc}
415	    case False{}:
416	      Go{c, x, acc}
417	
418	def strs.len(c: Cur, acc: List<&1, Bytes.Bytes>, r: Bytes.Bytes & U32) -> Loop<Bytes.Bytes, Bytes.Bytes>:
419	  (m, +n) = r
420	  loop.pick(Bytes.Bytes, Bytes.Bytes, U32.is_eq(n, 0), c, m, acc)
421	
422	def strs.read(acc: List<&1, Bytes.Bytes>, r: Cur & Bytes.Bytes) -> Loop<Bytes.Bytes, Bytes.Bytes>:
423	  (c, m) = r
424	  strs.len(c, acc, Bytes.length(m))
425	
426	# Each string takes a byte, so fuel of the bytes left is enough.
427	def strs(f: Nat, s: Loop<Bytes.Bytes, Bytes.Bytes>) -> Cur & List<&1, Bytes.Bytes>:
428	  match f:
429	    case 0n:
430	      loop.end(Bytes.Bytes, Bytes.Bytes, s)
431	    case 1n+p:
432	      match s:
433	        case Go{c, m, acc}:
434	          strs(p, strs.read(Con{m, acc}, cstr(c)))
435	        case Stop{c, acc}:
436	          (c, List.reverse(&1, Bytes.Bytes, acc))
437	
438	def strs.start(r: Cur & Nat) -> Cur & List<&1, Bytes.Bytes>:
439	  (c, f) = r
440	  strs(f, strs.read(Nil{}, cstr(c)))
441	
442	# NUL-terminated strings up to an empty one.
443	def names.dec(c: Cur) -> Cur & List<&1, Bytes.Bytes>:
444	  strs.start(left(c))
445	
446	def pr.len(c: Cur, acc: List<&1, Param>, r: Bytes.Bytes & U32) -> Loop<Bytes.Bytes, Param>:
447	  (m, +n) = r
448	  loop.pick(Bytes.Bytes, Param, U32.is_eq(n, 0), c, m, acc)
449	
450	def pr.read(acc: List<&1, Param>, r: Cur & Bytes.Bytes) -> Loop<Bytes.Bytes, Param>:
451	  (c, m) = r
452	  pr.len(c, acc, Bytes.length(m))
453	
454	def pr.val(name: Bytes.Bytes, acc: List<&1, Param>, r: Cur & Bytes.Bytes) -> Loop<Bytes.Bytes, Param>:
455	  (c, v) = r
456	  pr.read(Con{Param{name, v}, acc}, cstr(c))
457	
458	# Name and value pairs up to an empty name; a value may be empty. Each pair takes two bytes.
459	def prs(f: Nat, s: Loop<Bytes.Bytes, Param>) -> Cur & List<&1, Param>:
460	  match f:
461	    case 0n:
462	      loop.end(Bytes.Bytes, Param, s)
463	    case 1n+p:
464	      match s:
465	        case Go{c, name, acc}:
466	          prs(p, pr.val(name, acc, cstr(c)))
467	        case Stop{c, acc}:
468	          (c, List.reverse(&1, Param, acc))
469	
470	def prs.start(r: Cur & Nat) -> Cur & List<&1, Param>:
471	  (c, f) = r
472	  prs(f, pr.read(Nil{}, cstr(c)))
473	
474	def params.dec(c: Cur) -> Cur & List<&1, Param>:
475	  prs.start(left(c))
476	
477	def notice.code(acc: List<&1, Notice>, r: Cur & U32) -> Loop<U32, Notice>:
478	  (c, +k) = r
479	  loop.pick(U32, Notice, U32.is_eq(k, 0), c, k, acc)
480	
481	def notice.push(+code: U32, acc: List<&1, Notice>, r: Cur & Bytes.Bytes) -> Loop<U32, Notice>:
482	  (c, v) = r
483	  notice.code(Con{Notice{code, v}, acc}, uint(c, 1))
484	
485	# Fields up to a zero code byte; each takes two bytes, so the bytes left are enough fuel.
486	def notices(f: Nat, s: Loop<U32, Notice>) -> Cur & List<&1, Notice>:
487	  match f:
488	    case 0n:
489	      loop.end(U32, Notice, s)
490	    case 1n+p:
491	      match s:
492	        case Go{c, +code, acc}:
493	          notices(p, notice.push(code, acc, cstr(c)))
494	        case Stop{c, acc}:
495	          (c, List.reverse(&1, Notice, acc))
496	
497	def notices.start(r: Cur & Nat) -> Cur & List<&1, Notice>:
498	  (c, f) = r
499	  notices(f, notice.code(Nil{}, uint(c, 1)))
500	
501	def notices.dec(c: Cur) -> Cur & List<&1, Notice>:
502	  notices.start(left(c))
503	
504	def fd.g(name: Bytes.Bytes, +tb: U32, +col: U32, +oid: U32, +sz: U32, +md: U32, r: Cur & U32) -> Cur & Field:
505	  (c, +fmt) = r
506	  (c, Field{name, tb, col, oid, sz, md, fmt})
507	
508	def fd.f(name: Bytes.Bytes, +tb: U32, +col: U32, +oid: U32, +sz: U32, r: Cur & U32) -> Cur & Field:
509	  (c, +md) = r
510	  fd.g(name, tb, col, oid, sz, md, uint(c, 2))
511	
512	def fd.e(name: Bytes.Bytes, +tb: U32, +col: U32, +oid: U32, r: Cur & U32) -> Cur & Field:
513	  (c, +sz) = r
514	  fd.f(name, tb, col, oid, sz, uint(c, 4))
515	
516	def fd.d(name: Bytes.Bytes, +tb: U32, +col: U32, r: Cur & U32) -> Cur & Field:
517	  (c, +oid) = r
518	  fd.e(name, tb, col, oid, uint(c, 2))
519	
520	def fd.c(name: Bytes.Bytes, +tb: U32, r: Cur & U32) -> Cur & Field:
521	  (c, +col) = r
522	  fd.d(name, tb, col, uint(c, 4))
523	
524	def fd.b(name: Bytes.Bytes, r: Cur & U32) -> Cur & Field:
525	  (c, +tb) = r
526	  fd.c(name, tb, uint(c, 2))
527	
528	def fd.a(r: Cur & Bytes.Bytes) -> Cur & Field:
529	  (c, name) = r
530	  fd.b(name, uint(c, 4))
531	
532	def field(c: Cur) -> Cur & Field:
533	  fd.a(cstr(c))
534	
535	# ---- backend bodies
536	
537	def one(-A: Type, f: A -> Back, r: Cur & A) -> Cur & Back:
538	  (c, x) = r
539	  (c, f(x))
540	
541	def ps.b(name: Bytes.Bytes, r: Cur & Bytes.Bytes) -> Cur & Back:
542	  (c, v) = r
543	  (c, ParameterStatus{name, v})
544	
545	def ps.a(r: Cur & Bytes.Bytes) -> Cur & Back:
546	  (c, n) = r
547	  ps.b(n, cstr(c))
548	
549	def bkd.b(+pid: U32, r: Cur & U32) -> Cur & Back:
550	  (c, +k) = r
551	  (c, BackendKeyData{pid, k})
552	
553	def bkd.a(r: Cur & U32) -> Cur & Back:
554	  (c, +p) = r
555	  bkd.b(p, uint(c, 4))
556	
557	def rfq(r: Cur & U32) -> Cur & Back:
558	  (c, +s) = r
559	  (need(Bool.or(U32.is_eq(s, 73), Bool.or(U32.is_eq(s, 84), U32.is_eq(s, 69))), c), ReadyForQuery{s})
560	
561	def auth.data(k: U32, +code: U32, r: Cur & Bytes.Bytes) -> Cur & Back:
562	  match k:
563	    case 11:
564	      (c, d) = r
565	      (c, AuthSASLContinue{d})
566	    case 12:
567	      (c, d) = r
568	      (c, AuthSASLFinal{d})
569	    case _:
570	      (c, d) = r
571	      (c, AuthOther{code, d})
572	
573	def auth.of(k: U32, +code: U32, c: Cur) -> Cur & Back:
574	  match k:
575	    case 0:
576	      (c, AuthOk{})
577	    case 3:
578	      (c, AuthCleartext{})
579	    case 10:
580	      one(List<&1, Bytes.Bytes>, xs => AuthSASL{xs}, names.dec(c))
581	    case _:
582	      auth.data(code, code, rest(c))
583	
584	def auth(r: Cur & U32) -> Cur & Back:
585	  (c, +code) = r
586	  auth.of(code, code, c)
587	
588	def back(t: U32, +tag: U32, c: Cur) -> Cur & Back:
589	  match t:
590	    case 82:
591	      auth(uint(c, 4))
592	    case 83:
593	      ps.a(cstr(c))
594	    case 75:
595	      bkd.a(uint(c, 4))
596	    case 90:
597	      rfq(uint(c, 1))
598	    case 84:
599	      one(List<&1, Field>, xs => RowDescription{xs}, list16(~Field, ~field, c))
600	    case 68:
601	      one(List<&1, Maybe<&1, Bytes.Bytes>>, xs => DataRow{xs}, list16(~Maybe<&1, Bytes.Bytes>, ~col, c))
602	    case 67:
603	      one(Bytes.Bytes, x => CommandComplete{x}, cstr(c))
604	    case 69:
605	      one(List<&1, Notice>, xs => ErrorResponse{xs}, notices.dec(c))
606	    case 78:
607	      one(List<&1, Notice>, xs => NoticeResponse{xs}, notices.dec(c))
608	    case 49:
609	      (c, ParseComplete{})
610	    case 50:
611	      (c, BindComplete{})
612	    case 51:
613	      (c, CloseComplete{})
614	    case 110:
615	      (c, NoData{})
616	    case 73:
617	      (c, EmptyQuery{})
618	    case 115:
619	      (c, PortalSuspended{})
620	    case _:
621	      one(Bytes.Bytes, b => Other{tag, b}, rest(c))
622	
623	# ---- incremental reader
624	
625	# Bytes read from the server, and the index of the first byte not yet decoded.
626	type Reader is Type:
627	  Reader{buf: Bytes.Bytes, pos: U32}
628	
629	def reader() -> Reader:
630	  Reader{Bytes.new(0), 0}
631	
632	def feed.at(zero: Bool, buf: Bytes.Bytes, +pos: U32, chunk: Bytes.Bytes) -> Reader:
633	  match zero:
634	    case True{}:
635	      Reader{Bytes.append(buf, chunk), 0}
636	    case False{}:
637	      Reader{Bytes.append(Bytes.snd(Bytes.slice(buf, pos, 4294967295)), chunk), 0}
638	
639	# Drops the decoded bytes, then adds chunk.
640	def feed(r: Reader, chunk: Bytes.Bytes) -> Reader:
641	  Reader{buf, +pos} = r
642	  feed.at(U32.is_zero(pos), buf, pos, chunk)
643	
644	# One message; Want until it has all arrived; Bad for a length below 4 or above 1 GiB,
645	# or a body its tag does not fill exactly. After Bad the stream has lost its framing.
646	type Next is Type:
647	  Got{m: Back}
648	  Want{}
649	  Bad{}
650	
651	def next.done(ok: Bool, +pos: U32, +e: U32, buf: Bytes.Bytes, m: Back) -> Reader & Next:
652	  match ok:
653	    case True{}:
654	      (Reader{buf, e}, Got{m})
655	    case False{}:
656	      (Reader{buf, pos}, Bad{})
657	
658	def next.body(+pos: U32, r: Cur & Back) -> Reader & Next:
659	  (c, m) = r
660	  Cur{buf, +i, +e, +ok} = c
661	  next.done(Bool.and(ok, U32.is_eq(i, e)), pos, e, buf, m)
662	
663	def next.whole(ok: Bool, +t: U32, +pos: U32, +n: U32, buf: Bytes.Bytes) -> Reader & Next:
664	  match ok:
665	    case True{}:
666	      next.body(pos, back(t, t, Cur{buf, (pos + 5 : U32), (pos + 1 + n : U32), True{}}))
667	    case False{}:
668	      (Reader{buf, pos}, Want{})
669	
670	def next.size(ok: Bool, +t: U32, +pos: U32, +len: U32, +n: U32, buf: Bytes.Bytes) -> Reader & Next:
671	  match ok:
672	    case True{}:
673	      next.whole(Bytes.fits(len, (pos + 1 : U32), n), t, pos, n, buf)
674	    case False{}:
675	      (Reader{buf, pos}, Bad{})
676	
677	def next.n(+t: U32, +pos: U32, +len: U32, r: Cur & U32) -> Reader & Next:
678	  (c, +n) = r
679	  Cur{buf, i, e, ok} = c
680	  next.size(Bool.and(U32.is_le(4, n), U32.is_le(n, 1073741824)), t, pos, len, n, buf)
681	
682	def next.tag(+pos: U32, +len: U32, r: Cur & U32) -> Reader & Next:
683	  (c, +t) = r
684	  next.n(t, pos, len, uint(c, 4))
685	
686	def next.room(ok: Bool, +pos: U32, +len: U32, buf: Bytes.Bytes) -> Reader & Next:
687	  match ok:
688	    case True{}:
689	      next.tag(pos, len, uint(Cur{buf, pos, len, True{}}, 1))
690	    case False{}:
691	      (Reader{buf, pos}, Want{})
692	
693	def next.len(+pos: U32, r: Bytes.Bytes & U32) -> Reader & Next:
694	  (buf, +len) = r
695	  next.room(Bytes.fits(len, pos, 5), pos, len, buf)
696	
697	# The next backend message.
698	def next(r: Reader) -> Reader & Next:
699	  Reader{buf, +pos} = r
700	  next.len(pos, Bytes.length(buf))
701	
702	def pending.of(+pos: U32, r: Bytes.Bytes & U32) -> Reader & U32:
703	  (b, +len) = r
704	  (Reader{b, pos}, (len - pos : U32))
705	
706	# The count of bytes read but not yet decoded.
707	def pending(r: Reader) -> Reader & U32:
708	  Reader{buf, +pos} = r
709	  pending.of(pos, Bytes.length(buf))
710	
711	# ---- frontend decoder, for tests and mock servers
712	
713	def one.f(-A: Type, f: A -> Front, r: Cur & A) -> Cur & Front:
714	  (c, x) = r
715	  (c, f(x))
716	
717	def fe.sasl(mech: Bytes.Bytes, r: Cur & Maybe<&1, Bytes.Bytes>) -> Cur & Front:
718	  (c, m) = r
719	  match m:
720	    case Some{d}:
721	      (c, SASLInitial{mech, d})
722	    case None{}:
723	      (fail(c), SASLInitial{mech, Bytes.new(0)})
724	
725	def fe.sasl.a(r: Cur & Bytes.Bytes) -> Cur & Front:
726	  (c, mech) = r
727	  fe.sasl(mech, col(c))
728	
729	def fe.pass(k: U32, c: Cur) -> Cur & Front:
730	  match k:
731	    case 1:
732	      one.f(Bytes.Bytes, t => Password{t}, cstr(c))
733	    case 2:
734	      fe.sasl.a(cstr(c))
735	    case _:
736	      one.f(Bytes.Bytes, d => SASLResponse{d}, rest(c))
737	
738	def fe.parse.c(name: Bytes.Bytes, q: Bytes.Bytes, r: Cur & List<&1, U32>) -> Cur & Front:
739	  (c, oids) = r
740	  (c, Parse{name, q, dup(oids)})
741	
742	def fe.parse.b(name: Bytes.Bytes, r: Cur & Bytes.Bytes) -> Cur & Front:
743	  (c, q) = r
744	  fe.parse.c(name, q, list16(~U32, ~u32s, c))
745	
746	def fe.parse.a(r: Cur & Bytes.Bytes) -> Cur & Front:
747	  (c, name) = r
748	  fe.parse.b(name, cstr(c))
749	
750	def fe.bind.e(p: Bytes.Bytes, s: Bytes.Bytes, fs: List<&1, U32>, ps: List<&1, Maybe<&1, Bytes.Bytes>>, r: Cur & List<&1, U32>) -> Cur & Front:
751	  (c, rs) = r
752	  (c, Bind{p, s, dup(fs), ps, dup(rs)})
753	
754	def fe.bind.d(p: Bytes.Bytes, s: Bytes.Bytes, fs: List<&1, U32>, r: Cur & List<&1, Maybe<&1, Bytes.Bytes>>) -> Cur & Front:
755	  (c, ps) = r
756	  fe.bind.e(p, s, fs, ps, list16(~U32, ~u16s, c))
757	
758	def fe.bind.c(p: Bytes.Bytes, s: Bytes.Bytes, r: Cur & List<&1, U32>) -> Cur & Front:
759	  (c, fs) = r
760	  fe.bind.d(p, s, fs, list16(~Maybe<&1, Bytes.Bytes>, ~col, c))
761	
762	def fe.bind.b(p: Bytes.Bytes, r: Cur & Bytes.Bytes) -> Cur & Front:
763	  (c, s) = r
764	  fe.bind.c(p, s, list16(~U32, ~u16s, c))
765	
766	def fe.bind.a(r: Cur & Bytes.Bytes) -> Cur & Front:
767	  (c, p) = r
768	  fe.bind.b(p, cstr(c))
769	
770	def fe.kind(close: Bool, +k: U32, r: Cur & Bytes.Bytes) -> Cur & Front:
771	  match close:
772	    case True{}:
773	      (c, n) = r
774	      (need(Bool.or(U32.is_eq(k, 83), U32.is_eq(k, 80)), c), CloseMsg{k, n})
775	    case False{}:
776	      (c, n) = r
777	      (need(Bool.or(U32.is_eq(k, 83), U32.is_eq(k, 80)), c), Describe{k, n})
778	
779	def fe.kind.a(+close: Bool, r: Cur & U32) -> Cur & Front:
780	  (c, +k) = r
781	  fe.kind(close, k, cstr(c))
782	
783	def fe.exec(p: Bytes.Bytes, r: Cur & U32) -> Cur & Front:
784	  (c, +max) = r
785	  (c, Execute{p, max})
786	
787	def fe.exec.a(r: Cur & Bytes.Bytes) -> Cur & Front:
788	  (c, p) = r
789	  fe.exec(p, uint(c, 4))
790	
791	def fe.tagged(t: U32, +mode: U32, c: Cur) -> Cur & Front:
792	  match t:
793	    case 112:
794	      fe.pass(mode, c)
795	    case 80:
796	      fe.parse.a(cstr(c))
797	    case 66:
798	      fe.bind.a(cstr(c))
799	    case 68:
800	      fe.kind.a(False{}, uint(c, 1))
801	    case 69:
802	      fe.exec.a(cstr(c))
803	    case 83:
804	      (c, Sync{})
805	    case 67:
806	      fe.kind.a(True{}, uint(c, 1))
807	    case 88:
808	      (c, Terminate{})
809	    case _:
810	      (fail(c), Sync{})
811	
812	def fe.len(+t: U32, +mode: U32, r: Cur & U32) -> Cur & Front:
813	  (c, +n) = r
814	  Cur{b, +i, +e, +ok} = c
815	  fe.tagged(t, mode, Cur{b, i, e, Bool.and(ok, U32.is_eq((n + 1 : U32), e))})
816	
817	def fe.tag(+mode: U32, r: Cur & U32) -> Cur & Front:
818	  (c, +t) = r
819	  fe.len(t, mode, uint(c, 4))
820	
821	def fe.pkt(k: U32, c: Cur) -> Cur & Front:
822	  match k:
823	    case 80877103:
824	      (c, SSLRequest{})
825	    case 196608:
826	      one.f(List<&1, Param>, ps => Startup{ps}, params.dec(c))
827	    case _:
828	      (fail(c), SSLRequest{})
829	
830	def fe.code(r: Cur & U32) -> Cur & Front:
831	  (c, +k) = r
832	  fe.pkt(k, c)
833	
834	def fe.plen(r: Cur & U32) -> Cur & Front:
835	  (c, +n) = r
836	  Cur{b, +i, +e, +ok} = c
837	  fe.code(uint(Cur{b, i, e, Bool.and(ok, U32.is_eq(n, e))}, 4))
838	
839	def fe.start(k: U32, +mode: U32, c: Cur) -> Cur & Front:
840	  match k:
841	    case 0:
842	      fe.plen(uint(c, 4))
843	    case _:
844	      fe.tag(mode, uint(c, 1))
845	
846	def fe.fin(r: Cur & Front) -> Maybe<&1, Front>:
847	  (c, m) = r
848	  Cur{b, +i, +e, ok} = c
849	  match ok:
850	    case True{}:
851	      Bool.pick(Maybe<&1, Front>, U32.is_eq(i, e), Some{m}, None{})
852	    case False{}:
853	      None{}
854	
855	# One frontend message that fills b exactly, or None. The protocol leaves the phase to context:
856	# mode 0 reads an untagged startup packet, and 'p' reads as Password (1), SASLInitial (2) or SASLResponse (3).
857	def decode.front(+mode: U32, b: Bytes.Bytes) -> Maybe<&1, Front>:
858	  Bytes.Bytes{+len, buf} = b
859	  fe.fin(fe.start(mode, mode, Cur{Bytes.Bytes{len, buf}, 0, len, True{}}))