1	import Base
2	import ../ByteMemTable.bend as ByteMemTable
3	import ../ByteWal.bend as ByteWal
4	import ../Wal.bend as Wal
5	import ./Protocol.bend as Protocol
6	import ./CursorOpenCodec.bend as CursorOpenCodec
7	import ../scan/RangeCursor.bend as RangeCursor
8	import bend-kit-bytes@0.3.2.0/bytes.bend as Bytes
9	
10	# Represents a decoded byte-oriented owner database operation.
11	type Operation is Type:
12	  Get{key: Bytes.Bytes}
13	  Put{key: Bytes.Bytes, value: Bytes.Bytes}
14	  Delete{key: Bytes.Bytes}
15	  WriteBatch{mutations: List<&1, ByteWal.Mutation>}
16	  Flush{}
17	  Compact{}
18	  Stats{}
19	  SnapshotOpen{}
20	  SnapshotGet{snapshot_id: U32, key: Bytes.Bytes}
21	  SnapshotClose{snapshot_id: U32}
22	  SnapshotCursorOpen{snapshot_id: U32, bounds: RangeCursor.Bounds}
23	  SnapshotCursorNext{cursor_id: U32}
24	  SnapshotCursorClose{cursor_id: U32}
25	  SnapshotCursorSeek{cursor_id: U32, lower: Maybe<&1, Bytes.Bytes>}
26	  DescribeOwner{}
27	
28	# Reports decoded arguments or a malformed/unsupported operation.
29	type Decision is Type:
30	  Decoded{operation: Operation}
31	  MalformedArguments{}
32	  UnsupportedOperation{}
33	
34	def bytes.triple.second(
35	  first: Bytes.Bytes,
36	  pair: Bytes.Bytes & Bytes.Bytes
37	) -> Bytes.Bytes & Bytes.Bytes & Bytes.Bytes:
38	  match pair:
39	    case (second, third): (first, second, third)
40	
41	def bytes.triple.pair(
42	  pair: Bytes.Bytes & Bytes.Bytes
43	) -> Bytes.Bytes & Bytes.Bytes & Bytes.Bytes:
44	  match pair:
45	    case (first, second):
46	      bytes.triple.second(first, ByteMemTable.bytes.clone(second))
47	
48	def bytes.triple(bytes: Bytes.Bytes) -> Bytes.Bytes & Bytes.Bytes & Bytes.Bytes:
49	  bytes.triple.pair(ByteMemTable.bytes.clone(bytes))
50	
51	def put.result(
52	  key: Bytes.Bytes & Bytes.Bytes,
53	  value: Bytes.Bytes & Bytes.Bytes
54	) -> Decision:
55	  match key:
56	    case (_, owned_key):
57	      match value:
58	        case (_, owned_value): Decoded{Put{owned_key, owned_value}}
59	
60	def put.sliced(
61	  copies: Bytes.Bytes & Bytes.Bytes,
62	  key_lengths: U32 & U32,
63	  value_lengths: U32 & U32
64	) -> Decision:
65	  match copies:
66	    case (key_source, value_source):
67	      match key_lengths:
68	        case (key_length, key_offset):
69	          match value_lengths:
70	            case (value_length, _):
71	              put.result(Bytes.slice(key_source, 8, key_length),
72	                Bytes.slice(value_source, (8 + key_offset : U32), value_length))
73	
74	def put.bound.result(
75	  valid: Bool,
76	  payload: Bytes.Bytes,
77	  key_lengths: U32 & U32,
78	  value_lengths: U32 & U32
79	) -> Decision:
80	  match valid:
81	    case False{}: MalformedArguments{}
82	    case True{}:
83	      put.sliced(ByteMemTable.bytes.clone(payload), key_lengths, value_lengths)
84	
85	def put.bound.checked(
86	  payload: Bytes.Bytes,
87	  actual_length: U32,
88	  key_lengths: U32 & U32 & U32,
89	  value_lengths: U32 & U32 & U32
90	) -> Decision:
91	  match key_lengths:
92	    case (key_check, (key_length, key_offset)):
93	      match value_lengths:
94	        case (value_check, (value_length, value_copy)):
95	          put.bound.result(
96	            Nat.is_eq(U32.to_nat(actual_length),
97	              Nat.add(8n, Nat.add(U32.to_nat(key_check),
98	                U32.to_nat(value_check)))),
99	            payload, (key_length, key_offset), (value_length, value_copy))
100	
101	def put.measure(
102	  pair: Bytes.Bytes & U32,
103	  key_lengths: U32 & U32 & U32,
104	  value_lengths: U32 & U32 & U32
105	) -> Decision:
106	  match pair:
107	    case (payload, actual_length):
108	      put.bound.checked(payload, actual_length, key_lengths, value_lengths)
109	
110	def put.value.parsed(
111	  pair: Bytes.Bytes & Maybe<&2, U32>,
112	  measure: Bytes.Bytes,
113	  key_length: U32
114	) -> Decision:
115	  match pair:
116	    case (_, Some{value_length}):
117	      put.measure(Bytes.length(measure),
118	        Protocol.u32.duplicate.triple(key_length),
119	        Protocol.u32.duplicate.triple(value_length))
120	    case _: MalformedArguments{}
121	
122	def put.value.length.keyed(
123	  pair: Bytes.Bytes & Maybe<&2, U32>,
124	  measure: Bytes.Bytes
125	) -> Decision:
126	  match pair:
127	    case (value_head, Some{key_length}):
128	      put.value.parsed(Bytes.get.u32be(value_head, 4), measure, key_length)
129	    case _: MalformedArguments{}
130	
131	def put.decode.copies(copies: Bytes.Bytes & Bytes.Bytes & Bytes.Bytes) -> Decision:
132	  match copies:
133	    case (key_head, (value_head, measure)):
134	      put.value.length.keyed(Bytes.get.u32be(key_head, 0), measure)
135	
136	def put.decode(payload: Bytes.Bytes) -> Decision:
137	  put.decode.copies(bytes.triple(payload))
138	
139	def empty.result(
140	  is_empty: Bool,
141	  operation: Operation
142	) -> Decision:
143	  match is_empty:
144	    case True{}: Decoded{operation}
145	    case False{}: MalformedArguments{}
146	
147	def empty.operation(
148	  pair: Bytes.Bytes & U32,
149	  operation: Operation
150	) -> Decision:
151	  match pair:
152	    case (_, length): empty.result(U32.is_eq(length, 0), operation)
153	
154	def batch.result(
155	  decoded: Result<&1, &1, Wal.Error, ByteWal.Batch>
156	) -> Decision:
157	  match decoded:
158	    case Fail{_}: MalformedArguments{}
159	    case Done{ByteWal.Batch{mutations}}:
160	      Decoded{WriteBatch{mutations}}
161	
162	def snapshot.get.key.sliced(pair: Bytes.Bytes & Bytes.Bytes) -> Bytes.Bytes:
163	  match pair:
164	    case (_, key): key
165	
166	def snapshot.get.decision(snapshot_id: U32, key: Bytes.Bytes) -> Decision:
167	  Decoded{SnapshotGet{snapshot_id, key}}
168	
169	def snapshot.get.decoded(
170	  payload: Bytes.Bytes,
171	  snapshot_id: U32,
172	  length: U32
173	) -> Decision:
174	  snapshot.get.decision(snapshot_id,
175	    snapshot.get.key.sliced(Bytes.slice(payload, 4, U32.sub(length, 4))))
176	
177	def snapshot.get.header.result(
178	  payload: Bytes.Bytes,
179	  snapshot_id: U32,
180	  length: U32,
181	  has_header: Bool
182	) -> Decision:
183	  match has_header:
184	    case True{}: snapshot.get.decoded(payload, snapshot_id, length)
185	    case False{}: MalformedArguments{}
186	
187	def snapshot.get.parsed(
188	  parsed: Bytes.Bytes & Maybe<&2, U32>,
189	  length: U32,
190	  has_header: Bool
191	) -> Decision:
192	  match parsed:
193	    case (payload, Some{snapshot_id}):
194	      snapshot.get.header.result(payload, snapshot_id, length, has_header)
195	    case _: MalformedArguments{}
196	
197	def snapshot.get.length.checked(
198	  parsed: Bytes.Bytes & Maybe<&2, U32>,
199	  lengths: U32 & U32 & U32
200	) -> Decision:
201	  match lengths:
202	    case (length, (check_length, _)):
203	      snapshot.get.parsed(parsed, length, U32.is_le(4, check_length))
204	
205	def snapshot.get.measured(measured: Bytes.Bytes & U32) -> Decision:
206	  match measured:
207	    case (payload, length):
208	      snapshot.get.length.checked(Bytes.get.u32be(payload, 0),
209	        Protocol.u32.duplicate.triple(length))
210	
211	def snapshot.get.decode(payload: Bytes.Bytes) -> Decision:
212	  snapshot.get.measured(Bytes.length(payload))
213	
214	# Decodes the exact four-byte owner token used to close a snapshot.
215	def snapshot.close.decision(parsed: Maybe<&2, U32>) -> Decision:
216	  match parsed:
217	    case Some{snapshot_id}: Decoded{SnapshotClose{snapshot_id}}
218	    case None{}: MalformedArguments{}
219	
220	# Rejects truncated and extended close-token payloads.
221	def snapshot.close.decode(payload: Bytes.Bytes) -> Decision:
222	  snapshot.close.decision(Protocol.snapshot.id.read(payload))
223	
224	# Converts one pure cursor-open codec decision to an owner operation.
225	def snapshot.cursor.open.decision(
226	  decision: CursorOpenCodec.Decision
227	) -> Decision:
228	  match decision:
229	    case CursorOpenCodec.Decoded{CursorOpenCodec.Request{snapshot_id, bounds}}:
230	      Decoded{SnapshotCursorOpen{snapshot_id, bounds}}
231	    case CursorOpenCodec.Malformed{}: MalformedArguments{}
232	
233	# Decodes exact range and prefix fields for cursor opening.
234	def snapshot.cursor.open.decode(payload: Bytes.Bytes) -> Decision:
235	  snapshot.cursor.open.decision(CursorOpenCodec.decode(payload))
236	
237	# Decodes the owner-issued cursor token used for one incremental read.
238	def snapshot.cursor.next.parsed(parsed: Maybe<&2, U32>) -> Decision:
239	  match parsed:
240	    case Some{cursor_id}: Decoded{SnapshotCursorNext{cursor_id}}
241	    case None{}: MalformedArguments{}
242	
243	# Decodes the exact four-byte cursor token for one incremental read.
244	def snapshot.cursor.next.decode(payload: Bytes.Bytes) -> Decision:
245	  snapshot.cursor.next.parsed(Protocol.snapshot.id.read(payload))
246	
247	# Decodes the owner-issued cursor token used for explicit closure.
248	def snapshot.cursor.close.parsed(parsed: Maybe<&2, U32>) -> Decision:
249	  match parsed:
250	    case Some{cursor_id}: Decoded{SnapshotCursorClose{cursor_id}}
251	    case None{}: MalformedArguments{}
252	
253	# Decodes the exact four-byte cursor token for explicit closure.
254	def snapshot.cursor.close.decode(payload: Bytes.Bytes) -> Decision:
255	  snapshot.cursor.close.parsed(Protocol.snapshot.id.read(payload))
256	
257	def snapshot.cursor.seek.range(
258	  snapshot_id: U32,
259	  bounds: RangeCursor.Bounds
260	) -> Decision:
261	  match bounds:
262	    case RangeCursor.Bounds{lower, None{}, None{}}:
263	      Decoded{SnapshotCursorSeek{snapshot_id, lower}}
264	    case _: MalformedArguments{}
265	
266	def snapshot.cursor.seek.decision(
267	  decision: CursorOpenCodec.Decision
268	) -> Decision:
269	  match decision:
270	    case CursorOpenCodec.Decoded{CursorOpenCodec.Request{snapshot_id, bounds}}:
271	      snapshot.cursor.seek.range(snapshot_id, bounds)
272	    case CursorOpenCodec.Malformed{}: MalformedArguments{}
273	
274	def snapshot.cursor.seek.decode(payload: Bytes.Bytes) -> Decision:
275	  snapshot.cursor.seek.decision(CursorOpenCodec.decode(payload))
276	
277	def decode.payload(
278	  operation_code: U32,
279	  payload: Bytes.Bytes
280	) -> Decision:
281	  match operation_code:
282	    case 1: Decoded{Get{payload}}
283	    case 2: put.decode(payload)
284	    case 3: Decoded{Delete{payload}}
285	    case 4: empty.operation(Bytes.length(payload), Flush{})
286	    case 5: empty.operation(Bytes.length(payload), Compact{})
287	    case 6: empty.operation(Bytes.length(payload), Stats{})
288	    case 8: batch.result(ByteWal.decode(payload))
289	    case 9: empty.operation(Bytes.length(payload), SnapshotOpen{})
290	    case 10: snapshot.get.decode(payload)
291	    case 11: snapshot.close.decode(payload)
292	    case 12: snapshot.cursor.open.decode(payload)
293	    case 13: snapshot.cursor.next.decode(payload)
294	    case 14: snapshot.cursor.close.decode(payload)
295	    case 15: snapshot.cursor.seek.decode(payload)
296	    case 16: empty.operation(Bytes.length(payload), DescribeOwner{})
297	    case _: UnsupportedOperation{}
298	
299	# Decodes the tagged operation payload without interpreting keys as text.
300	def decode(request: Protocol.Request) -> Decision:
301	  match request:
302	    case Protocol.Request{_, _, _, operation, payload}:
303	      decode.payload(operation, payload)
304	
305	def u32.bytes(value: U32) -> Bytes.Bytes:
306	  Bytes.set.u32be(Bytes.new(4), 0, value)
307	
308	def encode.put.sized(
309	  key_pair: Bytes.Bytes & U32,
310	  value_pair: Bytes.Bytes & U32
311	) -> Bytes.Bytes:
312	  match key_pair:
313	    case (key, key_length):
314	      match value_pair:
315	        case (value, value_length):
316	          Bytes.concat([u32.bytes(key_length), u32.bytes(value_length), key, value])
317	
318	def encode.put(key: Bytes.Bytes, value: Bytes.Bytes) -> Bytes.Bytes:
319	  encode.put.sized(Bytes.length(key), Bytes.length(value))