codec.bend

bend-kit-postgres@0.3.0.0 · 859 lines · 134 definitions

1# Postgres frontend/backend protocol 3.0 messages over Bytes, and an incremental reader of backend messages.
2import Base
3import 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).
8type 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.
13type 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.
28type 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).
32type 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.
37type 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.
61def 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
70def byte1(+c: U32) -> Bytes.Bytes:
71  Bytes.set(Bytes.new(1), 0, c)
72
73def be16(+v: U32) -> Bytes.Bytes:
74  Bytes.set.u16be(Bytes.new(2), 0, v)
75
76def 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.
80def cstr.enc(b: Bytes.Bytes) -> Bytes.Bytes:
81  Bytes.append(b, Bytes.new(1))
82
83def 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.
88def frame(+tag: U32, body: Bytes.Bytes) -> Bytes.Bytes:
89  frame.of(tag, Bytes.length(body))
90
91def 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.
96def packet(body: Bytes.Bytes) -> Bytes.Bytes:
97  packet.of(Bytes.length(body))
98
99def 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
103def 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
110def 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.
115def n16(xs: List<&1, Bytes.Bytes>) -> Bytes.Bytes:
116  n16.of(n16.count(xs))
117
118def 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
125def 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
132def 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.
137def 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
144def 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
151def 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
158def 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
165def 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
172def 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.
180def 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
207def 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.
211def 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.
260type Cur is Type:
261  Cur{b: Bytes.Bytes, i: U32, e: U32, ok: Bool}
262
263def fail(c: Cur) -> Cur:
264  Cur{b, +i, +e, ok} = c
265  Cur{b, i, e, False{}}
266
267def need(ok: Bool, c: Cur) -> Cur:
268  match ok:
269    case True{}:
270      c
271    case False{}:
272      fail(c)
273
274def 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
278def 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
286def 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).
294def 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
298def 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
302def 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.
310def 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
314def rest(c: Cur) -> Cur & Bytes.Bytes:
315  Cur{b, +i, +e, +ok} = c
316  take(Cur{b, i, e, ok}, (e - i : U32))
317
318def nul() -> String:
319  SCon{Chr{0}, SNil{}}
320
321def 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
328def 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.
337def 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
341def u16s(c: Cur) -> Cur & U32:
342  uint(c, 2)
343
344def u32s(c: Cur) -> Cur & U32:
345  uint(c, 4)
346
347# n more items after the one in r, oldest last in acc.
348def 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
357def 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.
365def 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
369def list16(~A: Type, ~item: Cur -> Cur & A, c: Cur) -> Cur & List<&1, A>:
370  n16.dec(~A, ~item, uint(c, 2))
371
372def 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
379def col.some(r: Cur & Bytes.Bytes) -> Cur & Maybe<&1, Bytes.Bytes>:
380  (c, b) = r
381  (c, Some{b})
382
383def 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
390def 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.
395def 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.
399type 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.
404def 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
411def 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
418def 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
422def 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.
427def 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
438def 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.
443def names.dec(c: Cur) -> Cur & List<&1, Bytes.Bytes>:
444  strs.start(left(c))
445
446def 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
450def 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
454def 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.
459def 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
470def prs.start(r: Cur & Nat) -> Cur & List<&1, Param>:
471  (c, f) = r
472  prs(f, pr.read(Nil{}, cstr(c)))
473
474def params.dec(c: Cur) -> Cur & List<&1, Param>:
475  prs.start(left(c))
476
477def 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
481def 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.
486def 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
497def notices.start(r: Cur & Nat) -> Cur & List<&1, Notice>:
498  (c, f) = r
499  notices(f, notice.code(Nil{}, uint(c, 1)))
500
501def notices.dec(c: Cur) -> Cur & List<&1, Notice>:
502  notices.start(left(c))
503
504def 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
508def 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
512def 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
516def 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
520def 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
524def fd.b(name: Bytes.Bytes, r: Cur & U32) -> Cur & Field:
525  (c, +tb) = r
526  fd.c(name, tb, uint(c, 2))
527
528def fd.a(r: Cur & Bytes.Bytes) -> Cur & Field:
529  (c, name) = r
530  fd.b(name, uint(c, 4))
531
532def field(c: Cur) -> Cur & Field:
533  fd.a(cstr(c))
534
535# ---- backend bodies
536
537def one(-A: Type, f: A -> Back, r: Cur & A) -> Cur & Back:
538  (c, x) = r
539  (c, f(x))
540
541def ps.b(name: Bytes.Bytes, r: Cur & Bytes.Bytes) -> Cur & Back:
542  (c, v) = r
543  (c, ParameterStatus{name, v})
544
545def ps.a(r: Cur & Bytes.Bytes) -> Cur & Back:
546  (c, n) = r
547  ps.b(n, cstr(c))
548
549def bkd.b(+pid: U32, r: Cur & U32) -> Cur & Back:
550  (c, +k) = r
551  (c, BackendKeyData{pid, k})
552
553def bkd.a(r: Cur & U32) -> Cur & Back:
554  (c, +p) = r
555  bkd.b(p, uint(c, 4))
556
557def 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
561def 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
573def 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
584def auth(r: Cur & U32) -> Cur & Back:
585  (c, +code) = r
586  auth.of(code, code, c)
587
588def 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.
626type Reader is Type:
627  Reader{buf: Bytes.Bytes, pos: U32}
628
629def reader() -> Reader:
630  Reader{Bytes.new(0), 0}
631
632def 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.
640def 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.
646type Next is Type:
647  Got{m: Back}
648  Want{}
649  Bad{}
650
651def 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
658def 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
663def 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
670def 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
677def 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
682def 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
686def 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
693def 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.
698def next(r: Reader) -> Reader & Next:
699  Reader{buf, +pos} = r
700  next.len(pos, Bytes.length(buf))
701
702def 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.
707def 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
713def one.f(-A: Type, f: A -> Front, r: Cur & A) -> Cur & Front:
714  (c, x) = r
715  (c, f(x))
716
717def 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
725def fe.sasl.a(r: Cur & Bytes.Bytes) -> Cur & Front:
726  (c, mech) = r
727  fe.sasl(mech, col(c))
728
729def 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
738def 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
742def 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
746def fe.parse.a(r: Cur & Bytes.Bytes) -> Cur & Front:
747  (c, name) = r
748  fe.parse.b(name, cstr(c))
749
750def 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
754def 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
758def 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
762def 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
766def fe.bind.a(r: Cur & Bytes.Bytes) -> Cur & Front:
767  (c, p) = r
768  fe.bind.b(p, cstr(c))
769
770def 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
779def fe.kind.a(+close: Bool, r: Cur & U32) -> Cur & Front:
780  (c, +k) = r
781  fe.kind(close, k, cstr(c))
782
783def fe.exec(p: Bytes.Bytes, r: Cur & U32) -> Cur & Front:
784  (c, +max) = r
785  (c, Execute{p, max})
786
787def fe.exec.a(r: Cur & Bytes.Bytes) -> Cur & Front:
788  (c, p) = r
789  fe.exec(p, uint(c, 4))
790
791def 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
812def 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
817def fe.tag(+mode: U32, r: Cur & U32) -> Cur & Front:
818  (c, +t) = r
819  fe.len(t, mode, uint(c, 4))
820
821def 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
830def fe.code(r: Cur & U32) -> Cur & Front:
831  (c, +k) = r
832  fe.pkt(k, c)
833
834def 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
839def 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
846def 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).
857def 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{}}))