reddb_wire/redwire/
operations.rs

1//! RedWire data-plane JSON payload contracts.
2//!
3//! The server owns execution. This module owns only the wire-visible
4//! request/reply envelopes carried by BulkInsert, Get, Delete, BulkOk,
5//! and DeleteOk frames.
6
7use super::frame::MessageKind;
8use serde_json::{Map as JsonMap, Value as JsonValue};
9use std::fmt;
10
11#[derive(Debug, Clone, PartialEq)]
12pub struct InsertDispatchPayload {
13    pub collection: String,
14    pub payload: Option<JsonValue>,
15    pub payloads: Option<Vec<JsonValue>>,
16    pub idempotency_key: Option<String>,
17    pub batch: bool,
18}
19
20#[derive(Debug, Clone, PartialEq, Eq)]
21pub struct KeyPayload {
22    pub collection: String,
23    pub id: String,
24}
25
26#[derive(Debug, Clone, PartialEq, Eq)]
27pub struct BulkOkPayload {
28    pub affected: u64,
29    pub rids: Vec<String>,
30    pub ids: Vec<String>,
31}
32
33#[derive(Debug, Clone, PartialEq, Eq)]
34pub enum OperationPayloadError {
35    InvalidJson { op: &'static str, message: String },
36    ExpectedObject { op: &'static str },
37    MissingCollection { op: &'static str },
38    MissingId { op: &'static str },
39    TruncatedBulkOkCount,
40}
41
42impl fmt::Display for OperationPayloadError {
43    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
44        match self {
45            Self::InvalidJson { op, message } => write!(f, "{op}: invalid JSON: {message}"),
46            Self::ExpectedObject { op } => write!(f, "{op}: payload must be a JSON object"),
47            Self::MissingCollection { op } => {
48                write!(f, "{op}: missing 'collection' string")
49            }
50            Self::MissingId { op } => write!(f, "{op}: missing 'id' string"),
51            Self::TruncatedBulkOkCount => write!(f, "BulkOk truncated: expected 8-byte count"),
52        }
53    }
54}
55
56impl std::error::Error for OperationPayloadError {}
57
58#[derive(Debug, Clone, PartialEq, Eq)]
59pub enum OperationReplyError {
60    Engine(String),
61    UnexpectedKind {
62        expected: &'static str,
63        actual: MessageKind,
64    },
65}
66
67impl fmt::Display for OperationReplyError {
68    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
69        match self {
70            Self::Engine(message) => write!(f, "{message}"),
71            Self::UnexpectedKind { expected, actual } => {
72                write!(f, "expected {expected}, got {actual:?}")
73            }
74        }
75    }
76}
77
78impl std::error::Error for OperationReplyError {}
79
80pub fn encode_insert_payload(collection: &str, payload: JsonValue) -> Vec<u8> {
81    let mut obj = JsonMap::new();
82    obj.insert(
83        "collection".into(),
84        JsonValue::String(collection.to_string()),
85    );
86    obj.insert("payload".into(), payload);
87    serde_json::to_vec(&JsonValue::Object(obj)).expect("insert payload JSON is serializable")
88}
89
90pub fn encode_bulk_insert_payload(collection: &str, payloads: Vec<JsonValue>) -> Vec<u8> {
91    let mut obj = JsonMap::new();
92    obj.insert(
93        "collection".into(),
94        JsonValue::String(collection.to_string()),
95    );
96    obj.insert("payloads".into(), JsonValue::Array(payloads));
97    serde_json::to_vec(&JsonValue::Object(obj)).expect("bulk insert payload JSON is serializable")
98}
99
100pub fn decode_insert_dispatch_payload(
101    bytes: &[u8],
102) -> Result<InsertDispatchPayload, OperationPayloadError> {
103    let obj = object_from_payload("Insert", bytes)?;
104    let collection = required_collection("Insert", &obj)?;
105    let payload = obj.get("payload").cloned();
106    let payloads = obj
107        .get("payloads")
108        .and_then(JsonValue::as_array)
109        .map(|items| items.to_vec());
110    let idempotency_key = obj
111        .get("idempotency_key")
112        .and_then(JsonValue::as_str)
113        .map(String::from);
114    let batch = obj
115        .get("batch")
116        .and_then(JsonValue::as_bool)
117        .unwrap_or(false);
118    Ok(InsertDispatchPayload {
119        collection,
120        payload,
121        payloads,
122        idempotency_key,
123        batch,
124    })
125}
126
127pub fn encode_key_payload(collection: &str, id: &str) -> Vec<u8> {
128    let mut obj = JsonMap::new();
129    obj.insert(
130        "collection".into(),
131        JsonValue::String(collection.to_string()),
132    );
133    obj.insert("id".into(), JsonValue::String(id.to_string()));
134    serde_json::to_vec(&JsonValue::Object(obj)).expect("key payload JSON is serializable")
135}
136
137pub fn decode_get_payload(bytes: &[u8]) -> Result<KeyPayload, OperationPayloadError> {
138    decode_key_payload("Get", bytes)
139}
140
141pub fn decode_delete_payload(bytes: &[u8]) -> Result<KeyPayload, OperationPayloadError> {
142    decode_key_payload("Delete", bytes)
143}
144
145pub fn encode_query_result_summary_payload(statement: &str, affected: u64) -> Vec<u8> {
146    let statement = serde_json::to_string(statement).expect("statement string is serializable");
147    format!(r#"{{"affected":{affected},"ok":true,"statement":{statement}}}"#).into_bytes()
148}
149
150pub fn decode_query_result_payload(bytes: &[u8]) -> Result<JsonValue, OperationPayloadError> {
151    json_value_from_payload("QueryResult", bytes)
152}
153
154pub fn encode_get_result_payload(found: bool) -> Vec<u8> {
155    let mut obj = JsonMap::new();
156    obj.insert("ok".into(), JsonValue::Bool(true));
157    obj.insert("found".into(), JsonValue::Bool(found));
158    serde_json::to_vec(&JsonValue::Object(obj)).expect("get result payload JSON is serializable")
159}
160
161pub fn decode_get_result_payload(bytes: &[u8]) -> Result<JsonValue, OperationPayloadError> {
162    json_value_from_payload("GetResult", bytes)
163}
164
165pub fn decode_text_payload(bytes: &[u8]) -> String {
166    String::from_utf8_lossy(bytes).into_owned()
167}
168
169pub fn decode_error_payload(bytes: &[u8]) -> String {
170    decode_text_payload(bytes)
171}
172
173pub fn expect_result_or_error(
174    kind: MessageKind,
175    payload: &[u8],
176) -> Result<&[u8], OperationReplyError> {
177    expect_payload_or_error(kind, payload, MessageKind::Result, "Result/Error")
178}
179
180pub fn expect_bulk_ok_or_error(
181    kind: MessageKind,
182    payload: &[u8],
183) -> Result<&[u8], OperationReplyError> {
184    expect_payload_or_error(kind, payload, MessageKind::BulkOk, "BulkOk/Error")
185}
186
187pub fn expect_delete_ok_or_error(
188    kind: MessageKind,
189    payload: &[u8],
190) -> Result<&[u8], OperationReplyError> {
191    expect_payload_or_error(kind, payload, MessageKind::DeleteOk, "DeleteOk/Error")
192}
193
194pub fn expect_pong_reply(kind: MessageKind) -> Result<(), OperationReplyError> {
195    if kind == MessageKind::Pong {
196        Ok(())
197    } else {
198        Err(OperationReplyError::UnexpectedKind {
199            expected: "Pong",
200            actual: kind,
201        })
202    }
203}
204
205pub fn encode_bulk_ok_payload(affected: u64, ids: Vec<JsonValue>) -> Vec<u8> {
206    let mut obj = JsonMap::new();
207    obj.insert("affected".into(), JsonValue::Number(affected.into()));
208    obj.insert("ids".into(), JsonValue::Array(ids));
209    serde_json::to_vec(&JsonValue::Object(obj)).expect("bulk ok payload JSON is serializable")
210}
211
212pub fn encode_bulk_ok_payload_from_json_ids_bytes(affected: u64, ids: &[u8]) -> Vec<u8> {
213    let ids = match serde_json::from_slice::<JsonValue>(ids) {
214        Ok(JsonValue::Array(items)) => items,
215        _ => Vec::new(),
216    };
217    encode_bulk_ok_payload(affected, ids)
218}
219
220pub fn encode_bulk_ok_payload_from_json_id_literals<I, S>(affected: u64, ids: I) -> Vec<u8>
221where
222    I: IntoIterator<Item = S>,
223    S: AsRef<str>,
224{
225    let ids = ids
226        .into_iter()
227        .map(|id| {
228            serde_json::from_str::<JsonValue>(id.as_ref())
229                .unwrap_or_else(|_| JsonValue::String(id.as_ref().to_string()))
230        })
231        .collect();
232    encode_bulk_ok_payload(affected, ids)
233}
234
235pub fn decode_bulk_ok_payload(bytes: &[u8]) -> Result<BulkOkPayload, OperationPayloadError> {
236    let obj = object_from_payload("BulkOk", bytes)?;
237    let affected = obj.get("affected").and_then(JsonValue::as_u64).unwrap_or(0);
238    let rids: Vec<String> = obj
239        .get("rids")
240        .or_else(|| obj.get("ids"))
241        .and_then(JsonValue::as_array)
242        .map(|items| items.iter().filter_map(json_id_to_string).collect())
243        .unwrap_or_default();
244    let ids: Vec<String> = obj
245        .get("ids")
246        .and_then(JsonValue::as_array)
247        .map(|items| items.iter().filter_map(json_id_to_string).collect())
248        .unwrap_or_else(|| rids.clone());
249    Ok(BulkOkPayload {
250        affected,
251        rids,
252        ids,
253    })
254}
255
256pub fn encode_bulk_ok_count_payload(count: u64) -> Vec<u8> {
257    count.to_le_bytes().to_vec()
258}
259
260pub fn decode_bulk_ok_count_payload(bytes: &[u8]) -> Result<u64, OperationPayloadError> {
261    if bytes.len() < 8 {
262        return Err(OperationPayloadError::TruncatedBulkOkCount);
263    }
264    let mut count = [0u8; 8];
265    count.copy_from_slice(&bytes[..8]);
266    Ok(u64::from_le_bytes(count))
267}
268
269pub fn decode_delete_ok_affected(bytes: &[u8]) -> Result<u64, OperationPayloadError> {
270    let obj = object_from_payload("DeleteOk", bytes)?;
271    Ok(obj.get("affected").and_then(JsonValue::as_u64).unwrap_or(0))
272}
273
274pub fn encode_delete_ok_payload(affected: u64) -> Vec<u8> {
275    let mut obj = JsonMap::new();
276    obj.insert("affected".into(), JsonValue::Number(affected.into()));
277    serde_json::to_vec(&JsonValue::Object(obj)).expect("delete ok payload JSON is serializable")
278}
279
280fn expect_payload_or_error<'a>(
281    actual: MessageKind,
282    payload: &'a [u8],
283    ok: MessageKind,
284    expected: &'static str,
285) -> Result<&'a [u8], OperationReplyError> {
286    match actual {
287        kind if kind == ok => Ok(payload),
288        MessageKind::Error => Err(OperationReplyError::Engine(decode_error_payload(payload))),
289        other => Err(OperationReplyError::UnexpectedKind {
290            expected,
291            actual: other,
292        }),
293    }
294}
295
296fn decode_key_payload(op: &'static str, bytes: &[u8]) -> Result<KeyPayload, OperationPayloadError> {
297    let obj = object_from_payload(op, bytes)?;
298    let collection = required_collection(op, &obj)?;
299    let id = match obj.get("id").and_then(JsonValue::as_str) {
300        Some(value) if !value.is_empty() => value.to_string(),
301        _ => return Err(OperationPayloadError::MissingId { op }),
302    };
303    Ok(KeyPayload { collection, id })
304}
305
306fn json_value_from_payload(
307    op: &'static str,
308    bytes: &[u8],
309) -> Result<JsonValue, OperationPayloadError> {
310    let value: JsonValue =
311        serde_json::from_slice(bytes).map_err(|err| OperationPayloadError::InvalidJson {
312            op,
313            message: err.to_string(),
314        })?;
315    match value {
316        JsonValue::Object(_) => Ok(value),
317        _ => Err(OperationPayloadError::ExpectedObject { op }),
318    }
319}
320
321fn object_from_payload(
322    op: &'static str,
323    bytes: &[u8],
324) -> Result<JsonMap<String, JsonValue>, OperationPayloadError> {
325    let value: JsonValue =
326        serde_json::from_slice(bytes).map_err(|err| OperationPayloadError::InvalidJson {
327            op,
328            message: err.to_string(),
329        })?;
330    match value {
331        JsonValue::Object(obj) => Ok(obj),
332        _ => Err(OperationPayloadError::ExpectedObject { op }),
333    }
334}
335
336fn required_collection(
337    op: &'static str,
338    obj: &JsonMap<String, JsonValue>,
339) -> Result<String, OperationPayloadError> {
340    match obj.get("collection").and_then(JsonValue::as_str) {
341        Some(value) if !value.is_empty() => Ok(value.to_string()),
342        _ => Err(OperationPayloadError::MissingCollection { op }),
343    }
344}
345
346fn json_id_to_string(value: &JsonValue) -> Option<String> {
347    value
348        .as_str()
349        .map(String::from)
350        .or_else(|| value.as_u64().map(|n| n.to_string()))
351}
352
353#[cfg(test)]
354mod tests {
355    use super::*;
356
357    #[test]
358    fn insert_payload_round_trips_single_and_bulk_shapes() {
359        let single = decode_insert_dispatch_payload(&encode_insert_payload(
360            "users",
361            serde_json::json!({"name":"Ada"}),
362        ))
363        .unwrap();
364        assert_eq!(single.collection, "users");
365        assert_eq!(single.payload.unwrap(), serde_json::json!({"name":"Ada"}));
366        assert!(single.payloads.is_none());
367
368        let bulk = decode_insert_dispatch_payload(&encode_bulk_insert_payload(
369            "users",
370            vec![serde_json::json!({"name":"Ada"})],
371        ))
372        .unwrap();
373        assert_eq!(bulk.collection, "users");
374        assert_eq!(bulk.payloads.unwrap().len(), 1);
375        assert!(bulk.payload.is_none());
376    }
377
378    #[test]
379    fn key_payload_round_trips_get_and_delete_contracts() {
380        let bytes = encode_key_payload("users", "42");
381        assert_eq!(
382            decode_get_payload(&bytes).unwrap(),
383            KeyPayload {
384                collection: "users".into(),
385                id: "42".into(),
386            }
387        );
388        assert_eq!(
389            decode_delete_payload(&bytes).unwrap(),
390            KeyPayload {
391                collection: "users".into(),
392                id: "42".into(),
393            }
394        );
395    }
396
397    #[test]
398    fn bulk_ok_decodes_ids_and_affected_count() {
399        let payload = encode_bulk_ok_payload(2, vec![JsonValue::Number(1.into()), "2".into()]);
400        assert_eq!(
401            decode_bulk_ok_payload(&payload).unwrap(),
402            BulkOkPayload {
403                affected: 2,
404                rids: vec!["1".into(), "2".into()],
405                ids: vec!["1".into(), "2".into()],
406            }
407        );
408
409        let payload = encode_bulk_ok_payload_from_json_ids_bytes(2, br#"[1,"2"]"#);
410        assert_eq!(decode_bulk_ok_payload(&payload).unwrap().ids.len(), 2);
411
412        let payload = encode_bulk_ok_payload_from_json_id_literals(2, ["1", r#""2""#]);
413        assert_eq!(
414            decode_bulk_ok_payload(&payload).unwrap().ids,
415            vec!["1".to_string(), "2".to_string()]
416        );
417    }
418
419    #[test]
420    fn operation_reply_payloads_encode_wire_visible_json_contracts() {
421        let query_bytes = encode_query_result_summary_payload("INSERT", 3);
422        assert_eq!(
423            query_bytes,
424            br#"{"affected":3,"ok":true,"statement":"INSERT"}"#
425        );
426        let query = decode_query_result_payload(&query_bytes).unwrap();
427        assert_eq!(query["ok"], JsonValue::Bool(true));
428        assert_eq!(query["statement"], JsonValue::String("INSERT".into()));
429        assert_eq!(query["affected"], JsonValue::Number(3.into()));
430
431        let get = decode_get_result_payload(&encode_get_result_payload(false)).unwrap();
432        assert_eq!(get["ok"], JsonValue::Bool(true));
433        assert_eq!(get["found"], JsonValue::Bool(false));
434        assert_eq!(decode_text_payload(b"raw result"), "raw result");
435        assert_eq!(decode_error_payload(b"engine failed"), "engine failed");
436        assert_eq!(
437            expect_result_or_error(MessageKind::Result, b"ok").unwrap(),
438            b"ok"
439        );
440        assert_eq!(
441            expect_bulk_ok_or_error(MessageKind::Error, b"failed").unwrap_err(),
442            OperationReplyError::Engine("failed".to_string())
443        );
444        assert_eq!(
445            expect_delete_ok_or_error(MessageKind::Pong, b"").unwrap_err(),
446            OperationReplyError::UnexpectedKind {
447                expected: "DeleteOk/Error",
448                actual: MessageKind::Pong
449            }
450        );
451        assert!(expect_pong_reply(MessageKind::Pong).is_ok());
452
453        assert_eq!(
454            decode_delete_ok_affected(&encode_delete_ok_payload(7)).unwrap(),
455            7
456        );
457    }
458
459    #[test]
460    fn bulk_ok_count_payload_round_trips_legacy_binary_shape() {
461        let payload = encode_bulk_ok_count_payload(42);
462        assert_eq!(payload.len(), 8);
463        assert_eq!(decode_bulk_ok_count_payload(&payload).unwrap(), 42);
464        assert_eq!(
465            decode_bulk_ok_count_payload(&payload[..7])
466                .unwrap_err()
467                .to_string(),
468            "BulkOk truncated: expected 8-byte count"
469        );
470    }
471}
reddb_wire/redwire/operations.rs

reddb_wire/redwire/
operations.rs
