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::Transport(detail) if detail.starts_with("timeout") => PortError::Timeout,
105        DriverPortError::Transport(_) => PortError::Network,
106        DriverPortError::Http(detail) => PortError::Http(
107            detail
108                .split_whitespace()
109                .find_map(|word| word.parse::<u16>().ok())
110                .unwrap_or(0),
111        ),
112        _ => PortError::Other(0),
113    }
114}
115
116/// Standard padded base64 used by the HTTP response envelope.
117pub fn base64_encode(bytes: &[u8]) -> String {
118    const TABLE: &[u8] = b"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/";
119    let mut output = String::with_capacity(bytes.len().div_ceil(3) * 4);
120    for chunk in bytes.chunks(3) {
121        let first = chunk[0] as u32;
122        let second = chunk.get(1).copied().unwrap_or_default() as u32;
123        let third = chunk.get(2).copied().unwrap_or_default() as u32;
124        let value = (first << 16) | (second << 8) | third;
125        output.push(TABLE[((value >> 18) & 63) as usize] as char);
126        output.push(TABLE[((value >> 12) & 63) as usize] as char);
127        output.push(
128            chunk
129                .get(1)
130                .map_or('=', |_| TABLE[((value >> 6) & 63) as usize] as char),
131        );
132        output.push(
133            chunk
134                .get(2)
135                .map_or('=', |_| TABLE[(value & 63) as usize] as char),
136        );
137    }
138    output
139}
140
141#[cfg(test)]
142mod tests {
143    use super::*;
144
145    #[test]
146    fn http_envelope_keeps_binary_body_and_repeated_headers() {
147        let encoded = encode_http_response(&HttpResponse {
148            status: 418,
149            headers: vec![
150                ("set-cookie".into(), "a=1".into()),
151                ("set-cookie".into(), "b=2".into()),
152            ],
153            body: Bytes::from_static(&[0, 1, 2, 255]),
154        });
155        let value: JsonValue = serde_json::from_slice(&encoded).expect("valid envelope");
156        assert_eq!(value["status"], 418);
157        assert_eq!(value["headers"].as_array().map(Vec::len), Some(2));
158        assert_eq!(value["body"], "AAEC/w==");
159    }
160
161    #[test]
162    fn row_codec_round_trips_every_sql_value() {
163        let rows = vec![vec![
164            ("text".into(), SqlValue::Text("helix".into())),
165            ("integer".into(), SqlValue::Integer(7)),
166            ("real".into(), SqlValue::Real(1.5)),
167            ("blob".into(), SqlValue::Blob(vec![0, 127, 255])),
168            ("null".into(), SqlValue::Null),
169        ]];
170        let mut decoded = rows_from_reply_bytes(&rows_to_reply_bytes(&rows)).expect("row decode");
171        let mut expected = rows;
172        for row in &mut decoded {
173            row.sort_unstable_by(|left, right| left.0.cmp(&right.0));
174        }
175        for row in &mut expected {
176            row.sort_unstable_by(|left, right| left.0.cmp(&right.0));
177        }
178        assert_eq!(format!("{decoded:?}"), format!("{expected:?}"));
179    }
180
181    #[test]
182    fn error_mapping_is_shared_by_all_drivers() {
183        assert_eq!(
184            classify_port_error(DriverPortError::Transport("timeout: slow".into())),
185            PortError::Timeout
186        );
187        assert_eq!(
188            classify_port_error(DriverPortError::Http("http status 503".into())),
189            PortError::Http(503)
190        );
191    }
192}
helix_core/port_codec.rs

helix_core/
port_codec.rs
