1	import Base
2	import bend-kit-bytes@0.3.2.0/bytes.bend as Bytes
3	
4	# Represents one incremental cursor response frame.
5	type Frame is Type:
6	  CursorEnd{}
7	  CursorItem{key: Bytes.Bytes, value: Bytes.Bytes}
8	
9	# Reports a decoded cursor frame or malformed wire bytes.
10	type Decision is Type:
11	  Decoded{frame: Frame}
12	  Malformed{}
13	
14	# Encodes one exact byte offset and length into a fixed-width word.
15	def u32.bytes(value: U32) -> Bytes.Bytes:
16	  Bytes.set.u32be(Bytes.new(4), 0, value)
17	
18	def nat.duplicate.step(pair: Nat & Nat) -> Nat & Nat:
19	  match pair:
20	    case (first, second): (Nat.add(1n, first), Nat.add(1n, second))
21	
22	def nat.duplicate(value: Nat) -> Nat & Nat:
23	  match value:
24	    case 0n: (0n, 0n)
25	    case 1n+rest: nat.duplicate.step(nat.duplicate(rest))
26	
27	def u32.duplicate.result(pair: Nat & Nat) -> U32 & U32:
28	  match pair:
29	    case (first, second): (U32.from_nat(first), U32.from_nat(second))
30	
31	def u32.duplicate(value: U32) -> U32 & U32:
32	  u32.duplicate.result(nat.duplicate(U32.to_nat(value)))
33	
34	# Extracts the owned byte sequence from the bytes package slice result.
35	def slice.result(pair: Bytes.Bytes & Bytes.Bytes) -> Bytes.Bytes:
36	  match pair:
37	    case (_, sliced): sliced
38	
39	# Returns the one-byte tag for terminal cursor state.
40	def end.tag() -> Bytes.Bytes:
41	  Bytes.new(1)
42	
43	# Returns the one-byte tag for an item carrying raw key and value bytes.
44	def item.tag() -> Bytes.Bytes:
45	  slice.result(Bytes.slice(u32.bytes(1), 3, 1))
46	
47	# Encodes retained key/value buffers after measuring each exact byte length.
48	def item.measured(
49	  key_size: Bytes.Bytes & U32,
50	  value_size: Bytes.Bytes & U32
51	) -> Bytes.Bytes:
52	  match key_size:
53	    case (key, key_length):
54	      match value_size:
55	        case (value, value_length):
56	          Bytes.concat([item.tag(), u32.bytes(key_length), key,
57	            u32.bytes(value_length), value])
58	
59	# Encodes a cursor item using explicit lengths and no text conversion.
60	def item(
61	  key: Bytes.Bytes,
62	  value: Bytes.Bytes
63	) -> Bytes.Bytes:
64	  item.measured(Bytes.length(key), Bytes.length(value))
65	
66	# Encodes End separately from an item whose key and value are empty.
67	def encode(frame: Frame) -> Bytes.Bytes:
68	  match frame:
69	    case CursorEnd{}: end.tag()
70	    case CursorItem{key, value}: item(key, value)
71	
72	# Extracts exact byte equality from the bytes package comparison result.
73	def tag.matches(pair: Bytes.Bytes & Bytes.Bytes & Bool) -> Bool:
74	  match pair:
75	    case (_, _, equal): equal
76	
77	# Decodes the terminal marker only when no trailing bytes remain.
78	def end.checked(
79	  is_end: Bool,
80	  exact: Bool
81	) -> Decision:
82	  match is_end:
83	    case True{}:
84	      match exact:
85	        case True{}: Decoded{CursorEnd{}}
86	        case False{}: Malformed{}
87	    case False{}: Malformed{}
88	
89	# Combines the exact key and value byte slices into one cursor item.
90	def item.value.paired(
91	  pair: Bytes.Bytes & Bytes.Bytes,
92	  key: Bytes.Bytes
93	) -> Decision:
94	  match pair:
95	    case (_, value): Decoded{CursorItem{key, value}}
96	
97	# Creates the value slice from the source retained by the key slice.
98	def item.value.key.sliced(
99	  pair: Bytes.Bytes & Bytes.Bytes,
100	  value_offset: U32,
101	  value_length: U32
102	) -> Decision:
103	  match pair:
104	    case (source, key):
105	      item.value.paired(Bytes.slice(source, value_offset, value_length), key)
106	
107	# Slices the key first so the source remains available for the value slice.
108	def item.value.slices(
109	  payload: Bytes.Bytes,
110	  key_lengths: U32 & U32,
111	  value_length: U32
112	) -> Decision:
113	  match key_lengths:
114	    case (key_length, value_key_length):
115	      item.value.key.sliced(Bytes.slice(payload, 5, key_length),
116	        U32.from_nat(Nat.add(9n, U32.to_nat(value_key_length))), value_length)
117	
118	# Accepts only an exact frame length after reading the value length.
119	def item.value.bound.checked(
120	  exact: Bool,
121	  payload: Bytes.Bytes,
122	  key_length: U32,
123	  value_length: U32
124	) -> Decision:
125	  match exact:
126	    case True{}:
127	      item.value.slices(payload, u32.duplicate(key_length), value_length)
128	    case False{}: Malformed{}
129	
130	# Accepts only a value length that ends exactly at the frame boundary.
131	def item.value.bound.compare(
132	  payload: Bytes.Bytes,
133	  size: U32,
134	  key_lengths: U32 & U32,
135	  value_lengths: U32 & U32
136	) -> Decision:
137	  match key_lengths:
138	    case (checked_key_length, retained_key_length):
139	      match value_lengths:
140	        case (checked_value_length, retained_value_length):
141	          item.value.bound.checked(Nat.is_eq(U32.to_nat(size),
142	            Nat.add(9n, Nat.add(U32.to_nat(checked_key_length),
143	              U32.to_nat(checked_value_length)))), payload,
144	            retained_key_length, retained_value_length)
145	
146	# Duplicates lengths for exact validation and the returned value slice.
147	def item.value.bound(
148	  payload: Bytes.Bytes,
149	  size: U32,
150	  key_length: U32,
151	  value_length: U32
152	) -> Decision:
153	  item.value.bound.compare(payload, size, u32.duplicate(key_length),
154	    u32.duplicate(value_length))
155	
156	# Distinguishes a complete value length from a truncated header.
157	def item.value.parsed(
158	  payload: Bytes.Bytes,
159	  size: U32,
160	  key_length: U32,
161	  parsed: Maybe<&2, U32>
162	) -> Decision:
163	  match parsed:
164	    case Some{value_length}:
165	      item.value.bound(payload, size, key_length, value_length)
166	    case None{}: Malformed{}
167	
168	# Reads the value length at the boundary immediately after the key.
169	def item.value.length(
170	  payload: Bytes.Bytes,
171	  size: U32,
172	  key_length: U32,
173	  parsed: Maybe<&2, U32>
174	) -> Decision:
175	  item.value.parsed(payload, size, key_length, parsed)
176	
177	# Checks that the key length leaves room for the value length word.
178	def item.key.bound.read(
179	  read: Bytes.Bytes & Maybe<&2, U32>,
180	  size: U32,
181	  key_length: U32
182	) -> Decision:
183	  match read:
184	    case (payload, parsed):
185	      item.value.length(payload, size, key_length, parsed)
186	
187	# Checks that the key length leaves room for the value length word.
188	def item.key.bound.read.offset(
189	  payload: Bytes.Bytes,
190	  size: U32,
191	  key_lengths: U32 & U32
192	) -> Decision:
193	  match key_lengths:
194	    case (offset_length, retained_length):
195	      item.key.bound.read(Bytes.get.u32be(payload,
196	        U32.from_nat(Nat.add(5n, U32.to_nat(offset_length)))), size,
197	        retained_length)
198	
199	# Checks that the key length leaves room for the value length word.
200	def item.key.bound.checked(
201	  has_value_header: Bool,
202	  payload: Bytes.Bytes,
203	  size: U32,
204	  key_length: U32
205	) -> Decision:
206	  match has_value_header:
207	    case True{}:
208	      item.key.bound.read.offset(payload, size,
209	        u32.duplicate(key_length))
210	    case False{}: Malformed{}
211	
212	# Accepts key lengths that leave a complete value length word.
213	def item.key.bound.check(
214	  payload: Bytes.Bytes,
215	  sizes: U32 & U32,
216	  key_lengths: U32 & U32
217	) -> Decision:
218	  match sizes:
219	    case (checked_size, retained_size):
220	      match key_lengths:
221	        case (checked_key_length, retained_key_length):
222	          item.key.bound.checked(Nat.is_le(
223	            Nat.add(9n, U32.to_nat(checked_key_length)),
224	            U32.to_nat(checked_size)), payload, retained_size,
225	            retained_key_length)
226	
227	# Accepts key lengths that leave a complete value length word.
228	def item.key.bound.compare(
229	  payload: Bytes.Bytes,
230	  size: U32,
231	  key_lengths: U32 & U32
232	) -> Decision:
233	  item.key.bound.check(payload, u32.duplicate(size), key_lengths)
234	
235	# Accepts key lengths that leave a complete value length word.
236	def item.key.bound(
237	  payload: Bytes.Bytes,
238	  size: U32,
239	  key_length: U32
240	) -> Decision:
241	  item.key.bound.compare(payload, size, u32.duplicate(key_length))
242	
243	# Accepts the key-length field only when all required bytes are present.
244	def item.key.parsed(
245	  payload: Bytes.Bytes,
246	  size: U32,
247	  parsed: Maybe<&2, U32>
248	) -> Decision:
249	  match parsed:
250	    case Some{key_length}: item.key.bound(payload, size, key_length)
251	    case None{}: Malformed{}
252	
253	# Reads the key length before deriving the value-length offset.
254	def item.key.length(
255	  payload: Bytes.Bytes,
256	  size: U32,
257	  parsed: Maybe<&2, U32>
258	) -> Decision:
259	  item.key.parsed(payload, size, parsed)
260	
261	# Routes complete item headers to the key-length decoder.
262	def item.header.read(
263	  read: Bytes.Bytes & Maybe<&2, U32>,
264	  size: U32
265	) -> Decision:
266	  match read:
267	    case (payload, parsed): item.key.length(payload, size, parsed)
268	
269	# Routes complete item headers to the key-length decoder.
270	def item.header.checked(
271	  has_header: Bool,
272	  payload: Bytes.Bytes,
273	  size: U32
274	) -> Decision:
275	  match has_header:
276	    case True{}:
277	      item.header.read(Bytes.get.u32be(payload, 1), size)
278	    case False{}: Malformed{}
279	
280	# Checks header capacity while retaining its total for later framing checks.
281	def item.header.compare(
282	  payload: Bytes.Bytes,
283	  sizes: U32 & U32,
284	  is_item: Bool
285	) -> Decision:
286	  match sizes:
287	    case (checked_size, retained_size):
288	      item.header.checked(U32.is_le(9, checked_size) && is_item, payload,
289	        retained_size)
290	
291	# Rejects item headers that cannot contain both length words.
292	def item.header.size(
293	  payload: Bytes.Bytes,
294	  size: U32,
295	  is_item: Bool
296	) -> Decision:
297	  item.header.compare(payload, u32.duplicate(size), is_item)
298	
299	# Routes the item tag to exact length-prefixed decoding.
300	def decode.item(
301	  payload: Bytes.Bytes,
302	  size: U32,
303	  is_item: Bool
304	) -> Decision:
305	  item.header.size(payload, size, is_item)
306	
307	# Decodes the item marker from the retained response source.
308	def decode.item.tag.sliced(
309	  sliced: Bytes.Bytes & Bytes.Bytes,
310	  size: U32
311	) -> Decision:
312	  match sliced:
313	    case (source, tag_bytes):
314	      decode.item(source, size,
315	        tag.matches(Bytes.eq(tag_bytes, item.tag())))
316	
317	# Retains the response source while comparing the item marker.
318	def decode.item.tag(
319	  payload: Bytes.Bytes,
320	  size: U32
321	) -> Decision:
322	  decode.item.tag.sliced(Bytes.slice(payload, 0, 1), size)
323	
324	# Selects End validation or item decoding after the terminal tag check.
325	def decode.tag.end.checked(
326	  is_end: Bool,
327	  payload: Bytes.Bytes,
328	  sizes: U32 & U32
329	) -> Decision:
330	  match is_end:
331	    case True{}:
332	      match sizes:
333	        case (checked_size, _):
334	          end.checked(True{}, U32.is_eq(checked_size, 1))
335	    case False{}:
336	      match sizes:
337	        case (_, retained_size): decode.item.tag(payload, retained_size)
338	
339	# Checks the terminal marker while retaining bytes for item-tag decoding.
340	def decode.tag.end(
341	  sliced: Bytes.Bytes & Bytes.Bytes,
342	  size: U32
343	) -> Decision:
344	  match sliced:
345	    case (payload, tag_bytes):
346	      decode.tag.end.checked(tag.matches(Bytes.eq(tag_bytes, end.tag())),
347	        payload, u32.duplicate(size))
348	
349	# Validates that a nonempty encoded response has a supported tag.
350	def decode.nonempty(
351	  payload: Bytes.Bytes,
352	  size: U32,
353	  is_nonempty: Bool
354	) -> Decision:
355	  match is_nonempty:
356	    case True{}: decode.tag.end(Bytes.slice(payload, 0, 1), size)
357	    case False{}: Malformed{}
358	
359	# Decodes a complete cursor response, rejecting unknown tags and trailing data.
360	def decode.size.checked(
361	  payload: Bytes.Bytes,
362	  sizes: U32 & U32
363	) -> Decision:
364	  match sizes:
365	    case (checked_size, retained_size):
366	      decode.nonempty(payload, retained_size, U32.is_le(1, checked_size))
367	
368	# Decodes a complete cursor response, rejecting unknown tags and trailing data.
369	def decode.measured(measured: Bytes.Bytes & U32) -> Decision:
370	  match measured:
371	    case (payload, size): decode.size.checked(payload, u32.duplicate(size))
372	
373	# Decodes a complete cursor response from its exact byte payload.
374	def decode(payload: Bytes.Bytes) -> Decision:
375	  decode.measured(Bytes.length(payload))