cratefield_runtime_cloudflare/ports/
d1.rs

1//! `Database` over `worker::D1Database` (ADR 0004): sea-query renders the
2//! statement, values bind positionally, `batch` uses D1's atomic batch.
3
4use async_trait::async_trait;
5use cratefield_core::{Database, DbError, Row, Rows, Statement};
6use sea_query::Value as SeaValue;
7use serde_json::Value as Json;
8use worker::D1Database as WorkerD1;
9use worker::send::IntoSendFuture;
10
11pub struct D1Database(pub WorkerD1);
12
13/// A value ready to hand to D1's `bind()`. The decision is pure and unit
14/// tested natively; only the `JsValue` construction in [`bind_statement`]
15/// needs wasm.
16#[derive(Debug)]
17enum Bind {
18    Null,
19    Json(Json),
20    Bytes(Vec<u8>),
21}
22
23/// D1 BLOB marshalling, verified empirically under workerd/miniflare:
24/// - `bind()` stores an `ArrayBuffer` or typed array as BLOB; we build a real
25///   `Uint8Array`. A plain array of numbers happens to work locally but is
26///   undocumented, and a `DataView` silently stores zero bytes.
27/// - `serde_wasm_bindgen` can never produce a typed array out of a
28///   `serde_json::Value` (it has no bytes variant, so `serialize_bytes` is
29///   never called), so bytes must bypass the JSON hop entirely.
30/// - Reads are unambiguous: BLOB comes back as a plain array of numbers, and
31///   because D1 is SQLite (storage classes NULL/INTEGER/REAL/TEXT/BLOB map to
32///   null/number/string/array on read), a JSON array in a result row can only
33///   be a BLOB — TEXT containing `[1,2,3]` comes back as a string. So
34///   [`json_to_sea`] decodes arrays as bytes; no `typeof(col)` SQL is needed.
35fn decide_bind(value: &SeaValue) -> Bind {
36    // Bytes bypass the JSON hop entirely: `serde_wasm_bindgen` has no bytes
37    // variant, so a `serde_json::Value` can never come back out as a typed
38    // array (see the module comment). `Bytes(None)` falls through and joins
39    // every other null in [`Bind::Null`].
40    if let SeaValue::Bytes(Some(bytes)) = value {
41        return Bind::Bytes(bytes.to_vec());
42    }
43    let json = sea_to_json(value);
44    if json.is_null() {
45        Bind::Null
46    } else {
47        Bind::Json(json)
48    }
49}
50
51fn bind_statement(
52    db: &WorkerD1,
53    stmt: &Statement,
54) -> worker::Result<worker::d1::D1PreparedStatement> {
55    let prepared = db.prepare(&stmt.sql);
56    if stmt.values.0.is_empty() {
57        return Ok(prepared);
58    }
59    // Bind through serde_json + serde_wasm_bindgen rather than `bind_refs`
60    // with `D1Type`: the `D1Type` conversion path fails (integer binds
61    // error, text binds hang) under workerd/miniflare, while the
62    // JSON->JsValue path is the one workers-rs itself uses for D1 results.
63    //
64    // `decide_bind` routes every null (JSON null and `Bytes(None)`) to a
65    // single [`Bind::Null`]: `serde_wasm_bindgen` renders JSON null as
66    // `undefined`, and D1 rejects `undefined` with `D1_TYPE_ERROR: Type
67    // 'undefined' not supported`, so any write with a NULL column (a
68    // nullable field left unset) would fail. Bytes bypass the JSON hop (see
69    // [`decide_bind`]); everything else crosses as JSON.
70    let js_values: Vec<worker::wasm_bindgen::JsValue> = stmt
71        .values
72        .0
73        .iter()
74        .map(decide_bind)
75        .map(|bind| match bind {
76            Bind::Null => Ok(worker::wasm_bindgen::JsValue::NULL),
77            Bind::Json(json) => worker::d1::serde_wasm_bindgen::to_value(&json),
78            Bind::Bytes(bytes) => Ok(worker::js_sys::Uint8Array::from(bytes.as_slice()).into()),
79        })
80        .collect::<Result<Vec<_>, _>>()
81        .map_err(worker::Error::SerdeWasmBindgenError)?;
82    prepared.bind(&js_values)
83}
84
85fn sea_to_json(value: &SeaValue) -> serde_json::Value {
86    match value {
87        SeaValue::Bool(Some(v)) => serde_json::Value::Bool(*v),
88        SeaValue::TinyInt(Some(v)) => (*v).into(),
89        SeaValue::SmallInt(Some(v)) => (*v).into(),
90        SeaValue::Int(Some(v)) => (*v).into(),
91        SeaValue::BigInt(Some(v)) => (*v).into(),
92        SeaValue::TinyUnsigned(Some(v)) => (*v).into(),
93        SeaValue::SmallUnsigned(Some(v)) => (*v).into(),
94        SeaValue::Unsigned(Some(v)) => (*v).into(),
95        // u64 > JSON safe range is not representable; the portable subset
96        // stores ids as TEXT, so this never triggers in practice.
97        SeaValue::BigUnsigned(Some(v)) => (*v).into(),
98        SeaValue::Float(Some(v)) => f64::from(*v).into(),
99        SeaValue::Double(Some(v)) => (*v).into(),
100        SeaValue::String(Some(v)) => v.as_str().into(),
101        _ => serde_json::Value::Null,
102    }
103}
104
105fn json_to_sea(value: &Json) -> SeaValue {
106    match value {
107        Json::Null => SeaValue::String(None),
108        Json::Bool(v) => SeaValue::Bool(Some(*v)),
109        Json::Number(n) => {
110            if let Some(i) = n.as_i64() {
111                SeaValue::BigInt(Some(i))
112            } else {
113                SeaValue::Double(Some(n.as_f64().unwrap_or_default()))
114            }
115        }
116        Json::String(s) => SeaValue::String(Some(Box::new(s.clone()))),
117        Json::Array(items) => match array_to_bytes(items) {
118            Some(bytes) => SeaValue::Bytes(Some(Box::new(bytes))),
119            // Should be unreachable: D1 is SQLite, and its read mapping
120            // (NULL/INTEGER/REAL/TEXT/BLOB -> null/number/string/array)
121            // means a JSON array in a result row can only be a BLOB, which
122            // `array_to_bytes` accepts. If a future D1 ever returns a
123            // non-byte array here, stringify rather than silently null the
124            // column.
125            None => SeaValue::String(Some(Box::from(value.to_string()))),
126        },
127        Json::Object(_) => SeaValue::String(Some(Box::from(value.to_string()))),
128    }
129}
130
131/// A D1 BLOB reads back as a plain array of numbers, so an array of integers
132/// in `0..=255` is bytes (an empty array is empty bytes). Anything else
133/// (256, -1, 2.5, a nested array) is not a BLOB read.
134fn array_to_bytes(items: &[Json]) -> Option<Vec<u8>> {
135    items
136        .iter()
137        .map(|item| item.as_u64().and_then(|n| u8::try_from(n).ok()))
138        .collect()
139}
140
141fn d1_rows_to_rows(result: &worker::d1::D1Result) -> Result<Rows, DbError> {
142    let values: Vec<Json> = result.results().map_err(|err| {
143        log_d1(&err);
144        DbError::Query(err.to_string())
145    })?;
146    let rows = values
147        .into_iter()
148        .map(|value| {
149            let Json::Object(map) = value else {
150                return Row::new(Vec::new());
151            };
152            Row::new(
153                map.into_iter()
154                    .map(|(column, value)| (column, json_to_sea(&value)))
155                    .collect(),
156            )
157        })
158        .collect();
159    Ok(Rows::new(rows))
160}
161
162/// Logs a D1 error straight to `console_error!` before it becomes a
163/// [`DbError`]. `install_tracing` is a no-op on wasm (installing a dispatcher
164/// hangs the isolate), so the `tracing::error!` the harness emits when it maps
165/// a `DbError` to a 500 is dropped on Workers; without this, a failed query or
166/// write is an opaque `500` in `wrangler tail`. Errors are the exceptional
167/// path, and D1's messages are structural (no row data), so logging them here
168/// is safe.
169fn log_d1(err: &worker::Error) {
170    worker::console_error!("[d1] {err}");
171}
172
173#[async_trait]
174impl Database for D1Database {
175    async fn execute(&self, stmt: &Statement) -> Result<u64, DbError> {
176        let prepared = bind_statement(&self.0, stmt).map_err(|err| {
177            log_d1(&err);
178            DbError::Execute(err.to_string())
179        })?;
180        // Writes go through `batch` (single-statement): plain `.run()` /
181        // `.all()` promises never resolve for writes under local
182        // workerd/miniflare (verified empirically); batch resolves.
183        let results = self
184            .0
185            .batch(vec![prepared])
186            .into_send()
187            .await
188            .map_err(|err| {
189                log_d1(&err);
190                DbError::Execute(err.to_string())
191            })?;
192        let changed = results
193            .first()
194            .and_then(|result| result.meta().ok().flatten())
195            .and_then(|meta| meta.changes)
196            .unwrap_or_default();
197        Ok(u64::try_from(changed).unwrap_or(u64::MAX))
198    }
199
200    async fn query(&self, stmt: &Statement) -> Result<Rows, DbError> {
201        let prepared = bind_statement(&self.0, stmt).map_err(|err| {
202            log_d1(&err);
203            DbError::Query(err.to_string())
204        })?;
205        let result = prepared.all().into_send().await.map_err(|err| {
206            log_d1(&err);
207            DbError::Query(err.to_string())
208        })?;
209        d1_rows_to_rows(&result)
210    }
211
212    /// D1 batches are atomic (single implicit transaction).
213    async fn batch_atomic(&self, stmts: &[Statement]) -> Result<(), DbError> {
214        let mut prepared = Vec::with_capacity(stmts.len());
215        for stmt in stmts {
216            prepared.push(bind_statement(&self.0, stmt).map_err(|err| {
217                log_d1(&err);
218                DbError::Batch(err.to_string())
219            })?);
220        }
221        self.0.batch(prepared).into_send().await.map_err(|err| {
222            log_d1(&err);
223            DbError::Batch(err.to_string())
224        })?;
225        Ok(())
226    }
227}
228
229#[cfg(test)]
230mod tests {
231    use super::*;
232
233    /// The byte values a BLOB round trip must survive: NUL, the byte-255
234    /// i64/i32-sign trap, and two bytes that are invalid UTF-8 on their own.
235    fn hostile_payload() -> Vec<u8> {
236        vec![0x00, 0x41, 0xFF, 0x80, 0xFE]
237    }
238
239    /// The row JSON workerd produces for a BLOB column: a plain array of
240    /// numbers, as read back after the serde hop.
241    fn blob_row(payload: &[u8]) -> Json {
242        Json::Array(payload.iter().map(|byte| Json::from(*byte)).collect())
243    }
244
245    fn decoded(value: &Json) -> Vec<u8> {
246        match json_to_sea(value) {
247            SeaValue::Bytes(Some(bytes)) => *bytes,
248            SeaValue::String(Some(text)) => panic!("stringified, not decoded: {text}"),
249            other => panic!("unexpected sea value: {other:?}"),
250        }
251    }
252
253    #[test]
254    fn bind_sends_every_byte_of_a_blob() {
255        let payload = hostile_payload();
256        match decide_bind(&SeaValue::Bytes(Some(Box::new(payload.clone())))) {
257            Bind::Bytes(bytes) => assert_eq!(bytes, payload),
258            other => panic!("expected Bind::Bytes, got {other:?}"),
259        }
260    }
261
262    #[test]
263    fn bind_maps_bytes_none_to_null() {
264        assert!(matches!(decide_bind(&SeaValue::Bytes(None)), Bind::Null));
265    }
266
267    #[test]
268    fn decode_survives_nul_and_invalid_utf8() {
269        let payload = hostile_payload();
270        assert_eq!(decoded(&blob_row(&payload)), payload);
271    }
272
273    #[test]
274    fn decode_empty_blob_to_empty_bytes() {
275        assert_eq!(decoded(&blob_row(&[])), Vec::<u8>::new());
276    }
277
278    #[test]
279    fn non_byte_arrays_are_not_blobs() {
280        for row in [
281            Json::Array(vec![Json::from(256)]),
282            Json::Array(vec![Json::from(-1)]),
283            Json::Array(vec![Json::from(2.5)]),
284            Json::Array(vec![Json::from(1), Json::Array(vec![Json::from(2)])]),
285        ] {
286            match json_to_sea(&row) {
287                SeaValue::String(Some(_)) => {}
288                other => panic!("{row} decoded as {other:?}"),
289            }
290        }
291    }
292
293    #[test]
294    fn non_blob_binds_are_unchanged() {
295        for (sea, json) in [
296            (SeaValue::Bool(Some(true)), Json::Bool(true)),
297            (SeaValue::Int(Some(7)), Json::from(7)),
298            (
299                SeaValue::BigInt(Some(i64::from(i32::MAX))),
300                Json::from(i64::from(i32::MAX)),
301            ),
302            (SeaValue::Double(Some(2.5)), Json::from(2.5)),
303            (
304                SeaValue::String(Some(Box::new("x".to_owned()))),
305                Json::String("x".to_owned()),
306            ),
307        ] {
308            match decide_bind(&sea) {
309                Bind::Json(value) => assert_eq!(value, json),
310                other => panic!("{sea:?} bound as {other:?}"),
311            }
312        }
313    }
314
315    #[test]
316    fn non_blob_decodes_are_unchanged() {
317        assert!(matches!(json_to_sea(&Json::Null), SeaValue::String(None)));
318        assert!(matches!(
319            json_to_sea(&Json::Bool(false)),
320            SeaValue::Bool(Some(false))
321        ));
322        assert!(matches!(
323            json_to_sea(&Json::from(7)),
324            SeaValue::BigInt(Some(7))
325        ));
326        assert!(matches!(
327            json_to_sea(&Json::from(2.5)),
328            SeaValue::Double(Some(f)) if (f - 2.5).abs() < f64::EPSILON
329        ));
330        assert!(matches!(
331            json_to_sea(&Json::String("x".to_owned())),
332            SeaValue::String(Some(_))
333        ));
334        // A TEXT column holding the characters "[1,2]" returns as a string,
335        // not an array, and must not be mistaken for a BLOB.
336        assert!(matches!(
337            json_to_sea(&Json::String("[1,2]".to_owned())),
338            SeaValue::String(Some(_))
339        ));
340    }
341}
cratefield_runtime_cloudflare/ports/d1.rs

cratefield_runtime_cloudflare/ports/
d1.rs
