helix_core/
port_codec.rs

1//! Pure, runtime-free codecs shared by every platform driver.
2//!
3//! Drivers own I/O execution, while this module owns the bytes fed back through
4//! `Tick::PortReply`. Keeping the wire shape here prevents native, FFI and Web
5//! adapters from inventing subtly different HTTP or storage reply envelopes.
6
7use bytes::Bytes;
8use serde_json::{json, Value as JsonValue};
9
10use crate::effect::{HttpResponse, Row, SqlValue};
11use crate::error::PortError as DriverPortError;
12use crate::tick::{PortError, PortOutcome, ReplyBytes};
13
14/// Convert one completed HTTP port call into the canonical module-visible reply.
15///
16/// HTTP 4xx/5xx remain successful port calls. Their status is carried inside
17/// the envelope so the business module, rather than the driver, decides how to
18/// interpret the response.
19pub fn http_result_to_outcome(result: Result<HttpResponse, DriverPortError>) -> PortOutcome {
20    match result {
21        Ok(response) => PortOutcome::Ok(ReplyBytes(encode_http_response(&response))),
22        Err(error) => PortOutcome::Err(classify_port_error(error)),
23    }
24}
25
26/// Encode the canonical HTTP response envelope.
27///
28/// Shape: `{"status":u16,"headers":[[name,value],...],"body":"base64"}`.
29pub fn encode_http_response(response: &HttpResponse) -> Bytes {
30    let value = json!({
31        "status": response.status,
32        "headers": response.headers,
33        "body": base64_encode(&response.body),
34    });
35    serde_json::to_vec(&value)
36        .map(Bytes::from)
37        .unwrap_or_default()
38}
39
40/// Encode storage rows as the canonical JSON array used by `PortReply`.
41pub fn rows_to_reply_bytes(rows: &[Row]) -> Bytes {
42    let json_rows: Vec<JsonValue> = rows
43        .iter()
44        .map(|row| {
45            let object = row
46                .iter()
47                .map(|(key, value)| {
48                    let value = match value {
49                        SqlValue::Text(value) => JsonValue::String(value.clone()),
50                        SqlValue::Integer(value) => json!(value),
51                        SqlValue::Real(value) => json!(value),
52                        SqlValue::Blob(value) => json!(value),
53                        SqlValue::Null => JsonValue::Null,
54                    };
55                    (key.clone(), value)
56                })
57                .collect();
58            JsonValue::Object(object)
59        })
60        .collect();
61
62    serde_json::to_vec(&json_rows)
63        .map(Bytes::from)
64        .unwrap_or_default()
65}
66
67/// Decode [`rows_to_reply_bytes`] output for driver compatibility paths.
68pub fn rows_from_reply_bytes(bytes: &Bytes) -> Result<Vec<Row>, String> {
69    let json: Vec<serde_json::Map<String, JsonValue>> =
70        serde_json::from_slice(bytes).map_err(|error| error.to_string())?;
71
72    Ok(json
73        .into_iter()
74        .map(|object| {
75            object
76                .into_iter()
77                .map(|(key, value)| {
78                    let value = match value {
79                        JsonValue::String(value) => SqlValue::Text(value),
80                        JsonValue::Number(value) => value.as_i64().map_or_else(
81                            || SqlValue::Real(value.as_f64().unwrap_or(0.0)),
82                            SqlValue::Integer,
83                        ),
84                        JsonValue::Null => SqlValue::Null,
85                        JsonValue::Array(values) => SqlValue::Blob(
86                            values
87                                .iter()
88                                .filter_map(|value| value.as_u64().map(|value| value as u8))
89                                .collect(),
90                        ),
91                        other => SqlValue::Text(other.to_string()),
92                    };
93                    (key, value)
94                })
95                .collect()
96        })
97        .collect())
98}
99
100/// Convert a driver-facing port failure into the stable module-visible class.
101pub fn classify_port_error(error: DriverPortError) -> PortError {
102    match &error {
103        DriverPortError::Storage(_) => PortError::Storage(0),
104        DriverPortError::StorageConflict => PortError::Storage(1),
105        DriverPortError::Transport(detail) if detail.starts_with("timeout") => PortError::Timeout,
106        DriverPortError::Transport(_) => PortError::Network,
107        DriverPortError::Http(detail) => PortError::Http(
108            detail
109                .split_whitespace()
110                .find_map(|word| word.parse::<u16>().ok())
111                .unwrap_or(0),
112        ),
113        _ => PortError::Other(0),
114    }
115}
116
117/// Standard padded base64 used by the HTTP response envelope.
118pub fn base64_encode(bytes: &[u8]) -> String {
119    const TABLE: &[u8] = b"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/";
120    let mut output = String::with_capacity(bytes.len().div_ceil(3) * 4);
121    for chunk in bytes.chunks(3) {
122        let first = chunk[0] as u32;
123        let second = chunk.get(1).copied().unwrap_or_default() as u32;
124        let third = chunk.get(2).copied().unwrap_or_default() as u32;
125        let value = (first << 16) | (second << 8) | third;
126        output.push(TABLE[((value >> 18) & 63) as usize] as char);
127        output.push(TABLE[((value >> 12) & 63) as usize] as char);
128        output.push(
129            chunk
130                .get(1)
131                .map_or('=', |_| TABLE[((value >> 6) & 63) as usize] as char),
132        );
133        output.push(
134            chunk
135                .get(2)
136                .map_or('=', |_| TABLE[(value & 63) as usize] as char),
137        );
138    }
139    output
140}
141
142#[cfg(test)]
143mod tests {
144    use super::*;
145
146    #[test]
147    fn http_envelope_keeps_binary_body_and_repeated_headers() {
148        let encoded = encode_http_response(&HttpResponse {
149            status: 418,
150            headers: vec![
151                ("set-cookie".into(), "a=1".into()),
152                ("set-cookie".into(), "b=2".into()),
153            ],
154            body: Bytes::from_static(&[0, 1, 2, 255]),
155        });
156        let value: JsonValue = serde_json::from_slice(&encoded).expect("valid envelope");
157        assert_eq!(value["status"], 418);
158        assert_eq!(value["headers"].as_array().map(Vec::len), Some(2));
159        assert_eq!(value["body"], "AAEC/w==");
160    }
161
162    #[test]
163    fn row_codec_round_trips_every_sql_value() {
164        let rows = vec![vec![
165            ("text".into(), SqlValue::Text("helix".into())),
166            ("integer".into(), SqlValue::Integer(7)),
167            ("real".into(), SqlValue::Real(1.5)),
168            ("blob".into(), SqlValue::Blob(vec![0, 127, 255])),
169            ("null".into(), SqlValue::Null),
170        ]];
171        let mut decoded = rows_from_reply_bytes(&rows_to_reply_bytes(&rows)).expect("row decode");
172        let mut expected = rows;
173        for row in &mut decoded {
174            row.sort_unstable_by(|left, right| left.0.cmp(&right.0));
175        }
176        for row in &mut expected {
177            row.sort_unstable_by(|left, right| left.0.cmp(&right.0));
178        }
179        assert_eq!(format!("{decoded:?}"), format!("{expected:?}"));
180    }
181
182    #[test]
183    fn error_mapping_is_shared_by_all_drivers() {
184        assert_eq!(
185            classify_port_error(DriverPortError::Transport("timeout: slow".into())),
186            PortError::Timeout
187        );
188        assert_eq!(
189            classify_port_error(DriverPortError::Http("http status 503".into())),
190            PortError::Http(503)
191        );
192    }
193}
helix_core/port_codec.rs

helix_core/
port_codec.rs
