1	import Base
2	import ./Protocol.bend as Protocol
3	import ../scan/RangeCursor.bend as RangeCursor
4	import ../ByteMemTable.bend as ByteMemTable
5	import bend-kit-bytes@0.3.2.0/bytes.bend as Bytes
6	
7	# Carries one exact range or prefix request for an owner snapshot cursor.
8	type Request is Type:
9	  Request{snapshot_id: U32, bounds: RangeCursor.Bounds}
10	
11	# Reports a decoded cursor-open request or malformed argument bytes.
12	type Decision is Type:
13	  Decoded{request: Request}
14	  Malformed{}
15	
16	# Carries an input buffer, one parsed optional bound, and its next offset.
17	type BoundResult is Type:
18	  BoundParsed{source: Bytes.Bytes, value: Maybe<&1, Bytes.Bytes>, next: U32}
19	  BoundMalformed{}
20	
21	# Carries both range bounds after parsing the lower bound.
22	type RangeResult is Type:
23	  RangeParsed{lower: Maybe<&1, Bytes.Bytes>,
24	    upper: Maybe<&1, Bytes.Bytes>, next: U32}
25	  RangeMalformed{}
26	
27	# Encodes an absent bound with a reserved length outside the admitted frame cap.
28	def bound.absent() -> Bytes.Bytes:
29	  Protocol.u32.bytes(4294967295)
30	
31	# Encodes one present bound as a length followed by its raw bytes.
32	def bound.present(measured: Bytes.Bytes & U32) -> Bytes.Bytes:
33	  match measured:
34	    case (value, length):
35	      Bytes.concat([Protocol.u32.bytes(length), value])
36	
37	# Encodes an optional bound without interpreting or transforming its bytes.
38	def bound.encode(value: Maybe<&1, Bytes.Bytes>) -> Bytes.Bytes:
39	  match value:
40	    case None{}: bound.absent()
41	    case Some{bytes}: bound.present(Bytes.length(bytes))
42	
43	# Encodes one full range request with an owner snapshot token.
44	def encode.range(
45	  snapshot_id: U32,
46	  lower: Maybe<&1, Bytes.Bytes>,
47	  upper: Maybe<&1, Bytes.Bytes>
48	) -> Bytes.Bytes:
49	  Bytes.concat([Protocol.snapshot.id.bytes(snapshot_id),
50	    Protocol.u32.bytes(0), bound.encode(lower), bound.encode(upper)])
51	
52	# Encodes one byte-prefix request with an owner snapshot token.
53	def encode.prefix(snapshot_id: U32, prefix: Bytes.Bytes) -> Bytes.Bytes:
54	  Bytes.concat([Protocol.snapshot.id.bytes(snapshot_id),
55	    Protocol.u32.bytes(1), bound.present(Bytes.length(prefix))])
56	
57	# Copies a fixed-width word without recursion on its numeric value.
58	def u32.duplicate.triple(+value: U32) -> U32 & U32 & U32:
59	  (value, value, value)
60	
61	# Duplicates one fixed-width word for two bounded offset calculations.
62	def u32.duplicate.pair(+value: U32) -> U32 & U32:
63	  (value, value)
64	
65	# Retains the source buffer while storing one exact bound slice.
66	def field.sized.sliced(
67	  pair: Bytes.Bytes & Bytes.Bytes,
68	  next: U32
69	) -> BoundResult:
70	  match pair:
71	    case (source, value): BoundParsed{source, Some{value}, next}
72	
73	# Slices a validated span using separate copies for its start and end offsets.
74	def field.sized.checked(
75	  payload: Bytes.Bytes,
76	  offsets: U32 & U32 & U32,
77	  lengths: U32 & U32,
78	  fits: Bool
79	) -> BoundResult:
80	  match offsets:
81	    case (slice_offset, (next_offset, _)):
82	      match lengths:
83	        case (slice_length, next_length):
84	          match fits:
85	            case True{}:
86	              field.sized.sliced(Bytes.slice(payload,
87	                U32.from_nat(Nat.add(U32.to_nat(slice_offset), 4n)),
88	                slice_length),
89	                U32.from_nat(Nat.add(Nat.add(U32.to_nat(next_offset), 4n),
90	                  U32.to_nat(next_length))))
91	            case False{}: BoundMalformed{}
92	
93	# Selects a bounded raw-byte slice or rejects a span beyond the frame.
94	def field.sized.decision(
95	  payload: Bytes.Bytes,
96	  offset: U32,
97	  length: U32,
98	  fits: Bool
99	) -> BoundResult:
100	  field.sized.checked(payload,
101	    u32.duplicate.triple(offset), u32.duplicate.pair(length), fits)
102	
103	# Separates validation inputs from the values passed to the slice decision.
104	def field.sized.inputs(
105	  payload: Bytes.Bytes,
106	  totals: U32 & U32,
107	  offsets: U32 & U32,
108	  lengths: U32 & U32
109	) -> BoundResult:
110	  match totals:
111	    case (_, check_total):
112	      match offsets:
113	        case (slice_offset, check_offset):
114	          match lengths:
115	            case (slice_length, check_length):
116	              field.sized.decision(payload, slice_offset, slice_length,
117	                Nat.is_le(Nat.add(Nat.add(U32.to_nat(check_offset), 4n),
118	                  U32.to_nat(check_length)), U32.to_nat(check_total)))
119	
120	# Validates the payload span before slicing one present bound.
121	def field.sized(
122	  payload: Bytes.Bytes,
123	  total: U32,
124	  offset: U32,
125	  length: U32
126	) -> BoundResult:
127	  field.sized.inputs(payload, u32.duplicate.pair(total),
128	    u32.duplicate.pair(offset),
129	    u32.duplicate.pair(length))
130	
131	# Selects the absent sentinel or a validated present bound.
132	def field.value.decision(
133	  payload: Bytes.Bytes,
134	  total: U32,
135	  offset: U32,
136	  length: U32,
137	  absent: Bool
138	) -> BoundResult:
139	  match absent:
140	    case True{}:
141	      BoundParsed{payload, None{}, U32.from_nat(
142	        Nat.add(U32.to_nat(offset), 4n))}
143	    case False{}:
144	      field.sized(payload, total, offset, length)
145	
146	# Separates the sentinel check from the length passed to the bound parser.
147	def field.value.inputs(
148	  payload: Bytes.Bytes,
149	  total: U32,
150	  offset: U32,
151	  lengths: U32 & U32
152	) -> BoundResult:
153	  match lengths:
154	    case (bound_length, check_length):
155	      field.value.decision(payload, total, offset, bound_length,
156	        U32.is_eq(check_length, 4294967295))
157	
158	# Converts one field length into a bounded slice or an absent bound.
159	def field.value(
160	  payload: Bytes.Bytes,
161	  total: U32,
162	  offset: U32,
163	  length: U32
164	) -> BoundResult:
165	  field.value.inputs(payload, total, offset, u32.duplicate.pair(length))
166	
167	# Checks that one field's four-byte length is available at its offset.
168	def field.length(
169	  payload: Bytes.Bytes,
170	  total: U32,
171	  offset: U32,
172	  parsed: Maybe<&2, U32>
173	) -> BoundResult:
174	  match parsed:
175	    case None{}: BoundMalformed{}
176	    case Some{length}:
177	      field.value(payload, total, offset, length)
178	
179	# Parses one optional bound length from the supplied offset.
180	def field.read.parsed(
181	  payload: Bytes.Bytes,
182	  total: U32,
183	  offset: U32,
184	  pair: Bytes.Bytes & Maybe<&2, U32>
185	) -> BoundResult:
186	  match pair:
187	    case (_, parsed): field.length(payload, total, offset, parsed)
188	
189	# Keeps one source for slicing while reading its fixed-width length field.
190	def field.read.copies(
191	  sources: Bytes.Bytes & Bytes.Bytes,
192	  total: U32,
193	  offsets: U32 & U32
194	) -> BoundResult:
195	  match sources:
196	    case (payload, measured):
197	      match offsets:
198	        case (slice_offset, read_offset):
199	          field.read.parsed(payload, total, slice_offset,
200	            Bytes.get.u32be(measured, read_offset))
201	
202	# Parses one optional bound length from the supplied offset.
203	def field.read(
204	  payload: Bytes.Bytes,
205	  total: U32,
206	  offset: U32
207	) -> BoundResult:
208	  field.read.copies(ByteMemTable.bytes.clone(payload), total,
209	    u32.duplicate.pair(offset))
210	
211	# Combines both successfully parsed range bounds.
212	def range.upper.result(
213	  result: BoundResult,
214	  lower: Maybe<&1, Bytes.Bytes>
215	) -> RangeResult:
216	  match result:
217	    case BoundMalformed{}: RangeMalformed{}
218	    case BoundParsed{_, upper, next}:
219	      RangeParsed{lower, upper, next}
220	
221	# Reads the upper field from the source retained by lower-bound parsing.
222	def range.upper.read(
223	  source: Bytes.Bytes,
224	  total: U32,
225	  lower: Maybe<&1, Bytes.Bytes>,
226	  offset: U32
227	) -> RangeResult:
228	  range.upper.result(field.read(source, total, offset), lower)
229	
230	# Parses the upper bound after preserving the lower-bound result.
231	def range.upper(
232	  lower_result: BoundResult,
233	  total: U32
234	) -> RangeResult:
235	  match lower_result:
236	    case BoundMalformed{}: RangeMalformed{}
237	    case BoundParsed{source, lower, next}:
238	      range.upper.read(source, total, lower, next)
239	
240	# Rejects trailing bytes or returns the decoded range cursor request.
241	def range.finish.exact(
242	  exact: Bool,
243	  snapshot_id: U32,
244	  lower: Maybe<&1, Bytes.Bytes>,
245	  upper: Maybe<&1, Bytes.Bytes>
246	) -> Decision:
247	  match exact:
248	    case True{}:
249	      Decoded{Request{snapshot_id,
250	        RangeCursor.Bounds{lower, upper, None{}}}}
251	    case False{}: Malformed{}
252	
253	# Builds a range request only when its fields consume the entire frame.
254	def range.finish(
255	  parsed: RangeResult,
256	  snapshot_id: U32,
257	  total: U32
258	) -> Decision:
259	  match parsed:
260	    case RangeMalformed{}: Malformed{}
261	    case RangeParsed{lower, upper, next}:
262	      range.finish.exact(U32.is_eq(next, total), snapshot_id, lower, upper)
263	
264	# Parses range bounds with separate total lengths for IO-free validation.
265	def range.decode.totals(
266	  payload: Bytes.Bytes,
267	  snapshot_id: U32,
268	  totals: U32 & U32 & U32
269	) -> Decision:
270	  match totals:
271	    case (read_total, (upper_total, finish_total)):
272	      range.finish(range.upper(field.read(payload, read_total, 8), upper_total),
273	        snapshot_id, finish_total)
274	
275	# Parses both optional bounds and requires an exact request boundary.
276	def range.decode(
277	  payload: Bytes.Bytes,
278	  snapshot_id: U32,
279	  total: U32
280	) -> Decision:
281	  range.decode.totals(payload, snapshot_id,
282	    u32.duplicate.triple(total))
283	
284	# Rejects trailing bytes or returns the decoded prefix cursor request.
285	def prefix.finish.exact(
286	  exact: Bool,
287	  snapshot_id: U32,
288	  prefix: Bytes.Bytes
289	) -> Decision:
290	  match exact:
291	    case True{}:
292	      Decoded{Request{snapshot_id,
293	        RangeCursor.Bounds{None{}, None{}, Some{prefix}}}}
294	    case False{}: Malformed{}
295	
296	# Builds a prefix request only when its field consumes the remaining frame.
297	def prefix.finish(
298	  parsed: BoundResult,
299	  snapshot_id: U32,
300	  total: U32
301	) -> Decision:
302	  match parsed:
303	    case BoundMalformed{}: Malformed{}
304	    case BoundParsed{_, Some{prefix}, next}:
305	      prefix.finish.exact(U32.is_eq(next, total), snapshot_id, prefix)
306	    case BoundParsed{_, None{}, _}: Malformed{}
307	
308	# Parses prefix bounds with separate lengths for reading and validation.
309	def prefix.decode.totals(
310	  payload: Bytes.Bytes,
311	  snapshot_id: U32,
312	  totals: U32 & U32
313	) -> Decision:
314	  match totals:
315	    case (read_total, finish_total):
316	      prefix.finish(field.read(payload, read_total, 8), snapshot_id, finish_total)
317	
318	# Parses one present prefix bound and requires an exact request boundary.
319	def prefix.decode(
320	  payload: Bytes.Bytes,
321	  snapshot_id: U32,
322	  total: U32
323	) -> Decision:
324	  prefix.decode.totals(payload, snapshot_id, u32.duplicate.pair(total))
325	
326	# Selects range or prefix decoding from the exact mode word.
327	def mode(
328	  payload: Bytes.Bytes,
329	  snapshot_id: U32,
330	  total: U32,
331	  parsed: Maybe<&2, U32>
332	) -> Decision:
333	  match parsed:
334	    case Some{0}: range.decode(payload, snapshot_id, total)
335	    case Some{1}: prefix.decode(payload, snapshot_id, total)
336	    case _: Malformed{}
337	
338	# Requires the mode field to exist before selecting a cursor-open shape.
339	def mode.read.parsed(
340	  payload: Bytes.Bytes,
341	  snapshot_id: U32,
342	  total: U32,
343	  parsed: Maybe<&2, U32>
344	) -> Decision:
345	  mode(payload, snapshot_id, total, parsed)
346	
347	# Projects one fixed-width mode parse while retaining its request source.
348	def mode.read.field(
349	  payload: Bytes.Bytes,
350	  snapshot_id: U32,
351	  total: U32,
352	  field: Bytes.Bytes & Maybe<&2, U32>
353	) -> Decision:
354	  match field:
355	    case (_, parsed): mode.read.parsed(payload, snapshot_id, total, parsed)
356	
357	# Reads the mode from a copy while retaining the request for bound slicing.
358	def mode.read.copies(
359	  sources: Bytes.Bytes & Bytes.Bytes,
360	  snapshot_id: U32,
361	  total: U32
362	) -> Decision:
363	  match sources:
364	    case (payload, reader):
365	      mode.read.field(payload, snapshot_id, total,
366	        Bytes.get.u32be(reader, 4))
367	
368	# Requires the mode field to exist before selecting a cursor-open shape.
369	def mode.read(
370	  payload: Bytes.Bytes,
371	  snapshot_id: U32,
372	  total: U32,
373	  enough: Bool
374	) -> Decision:
375	  match enough:
376	    case True{}:
377	      mode.read.copies(ByteMemTable.bytes.clone(payload), snapshot_id, total)
378	    case False{}: Malformed{}
379	
380	# Decodes an owner token followed by one supported cursor-open mode.
381	def decode.payload.inputs(
382	  payload: Bytes.Bytes,
383	  totals: U32 & U32,
384	  parsed: Maybe<&2, U32>
385	) -> Decision:
386	  match totals:
387	    case (mode_total, check_total):
388	      match parsed:
389	        case Some{snapshot_id}:
390	          mode.read(payload, snapshot_id, mode_total,
391	            U32.is_le(8, check_total))
392	        case None{}: Malformed{}
393	
394	# Decodes an owner token followed by one supported cursor-open mode.
395	def decode.payload(
396	  payload: Bytes.Bytes,
397	  total: U32,
398	  parsed: Maybe<&2, U32>
399	) -> Decision:
400	  decode.payload.inputs(payload, u32.duplicate.pair(total), parsed)
401	
402	# Projects one snapshot-token parse while retaining the full request.
403	def decode.payload.field(
404	  payload: Bytes.Bytes,
405	  total: U32,
406	  field: Bytes.Bytes & Maybe<&2, U32>
407	) -> Decision:
408	  match field:
409	    case (_, parsed): decode.payload(payload, total, parsed)
410	
411	# Decodes an owner token from one input copy and retains another for mode parsing.
412	def decode.payload.copies(
413	  sources: Bytes.Bytes & Bytes.Bytes,
414	  total: U32
415	) -> Decision:
416	  match sources:
417	    case (payload, reader):
418	      decode.payload.field(payload, total, Bytes.get.u32be(reader, 0))
419	
420	# Decodes one bounded cursor-open request without interpreting its bytes.
421	def decode.measured(measured: Bytes.Bytes & U32) -> Decision:
422	  match measured:
423	    case (payload, total):
424	      decode.payload.copies(ByteMemTable.bytes.clone(payload), total)
425	
426	# Decodes a full-range or prefix request from its exact wire payload.
427	def decode(payload: Bytes.Bytes) -> Decision:
428	  decode.measured(Bytes.length(payload))