rainier-drivers 2.0.0

//! Cloudflare D1 executor (feature `d1-http`).
//!
//! D1 is SQLite, so the SQL is rendered with [`Dialect::Sqlite`]; what's
//! different is the wire. The D1 HTTP API and the Worker D1 binding both
//! answer a query with the same JSON shape — an array of row objects plus a
//! `meta` block. This module owns that JSON mapping ([`D1Row`], the
//! response/outcome parsing) and leaves the actual bytes-on-the-wire to a
//! caller-supplied [`D1Transport`].
//!
//! That split is deliberate: the transport is the only part that isn't
//! wasm-safe in every environment (server-side you'd back it with `reqwest`;
//! inside a Worker with the native `env.d1(...)` binding). Keeping it a trait
//! means this executor — and everything above it — compiles to `wasm32` and
//! runs in the Worker unchanged.

use chrono::{DateTime, NaiveDate, Utc};
use rainier_orm::sea_query::Value;
use rainier_orm::{Dialect, Error, ExecOutcome, Executor, Result, Row};
use serde_json::Value as Json;

/// One D1 query's response, already parsed from the `result[0]` object the
/// API/binding returns: the decoded rows plus the write metadata.
#[derive(Debug, Clone, Default)]
pub struct D1Response {
    /// Row objects from `results`, each a JSON object keyed by column name.
    pub rows: Vec<serde_json::Map<String, Json>>,
    /// `meta.last_row_id` — the autogenerated rowid of an INSERT.
    pub last_row_id: i64,
    /// `meta.changes` — rows written.
    pub changes: u64,
}

impl D1Response {
    /// Parse the full envelope a D1 `/query` call returns
    /// (`{ "result": [ { "results": [...], "meta": { ... } } ], ... }`) into
    /// the first statement's [`D1Response`]. Errors if `success` is false or
    /// the shape is unexpected.
    pub fn from_envelope(v: &Json) -> Result<Self> {
        if v.get("success").and_then(Json::as_bool) == Some(false) {
            let errs = v.get("errors").cloned().unwrap_or(Json::Null);
            return Err(Error::msg(format!("D1 query failed: {errs}")));
        }
        let first = v
            .get("result")
            .and_then(Json::as_array)
            .and_then(|a| a.first())
            .ok_or_else(|| Error::msg("D1 response missing result[0]"))?;
        Ok(Self::from_result(first))
    }

    /// Parse a single `result` entry (`{ "results": [...], "meta": {...} }`).
    /// This is the shape the Worker binding's per-statement result yields.
    pub fn from_result(first: &Json) -> Self {
        let rows = first
            .get("results")
            .and_then(Json::as_array)
            .map(|a| a.iter().filter_map(|r| r.as_object().cloned()).collect())
            .unwrap_or_default();
        let meta = first.get("meta");
        let last_row_id =
            meta.and_then(|m| m.get("last_row_id")).and_then(Json::as_i64).unwrap_or(0);
        let changes = meta.and_then(|m| m.get("changes")).and_then(Json::as_u64).unwrap_or(0);
        Self { rows, last_row_id, changes }
    }
}

/// Sends rendered SQL + bound parameters to D1 and returns the parsed
/// [`D1Response`]. Implement over `reqwest` (the HTTP `/query` API) on a
/// server, or over the Worker `env.d1(...)` binding inside a Worker — the
/// executor above is identical either way.
#[allow(async_fn_in_trait)]
pub trait D1Transport {
    /// Run `sql` with `params` bound, and parse what D1 sends back.
    async fn query(&self, sql: &str, params: Vec<Json>) -> Result<D1Response>;
}

/// An [`Executor`] over any [`D1Transport`]. Speaks the SQLite dialect and
/// lowers sea-query bind values to the JSON D1 expects.
#[derive(Clone)]
pub struct D1Executor<T> {
    transport: T,
}

impl<T> D1Executor<T> {
    /// An executor over `transport`.
    pub fn new(transport: T) -> Self {
        Self { transport }
    }

    /// The transport underneath.
    pub fn transport(&self) -> &T {
        &self.transport
    }
}

impl<T: D1Transport> Executor for D1Executor<T> {
    fn dialect(&self) -> Dialect {
        Dialect::Sqlite
    }

    async fn fetch_all(&self, sql: &str, params: Vec<Value>) -> Result<Vec<Box<dyn Row>>> {
        let resp = self.transport.query(sql, to_json_params(params)?).await?;
        Ok(resp.rows.into_iter().map(|obj| Box::new(D1Row(obj)) as Box<dyn Row>).collect())
    }

    async fn execute(&self, sql: &str, params: Vec<Value>) -> Result<ExecOutcome> {
        let resp = self.transport.query(sql, to_json_params(params)?).await?;
        Ok(ExecOutcome { rows_affected: resp.changes, last_insert_id: resp.last_row_id })
    }
}

/// One D1 result row — a JSON object keyed by column name.
///
/// D1's JSON typing is SQLite's: integers and reals are JSON numbers, text and
/// dates are strings, booleans come back as `0`/`1`, and BLOBs as a JSON array
/// of byte values. Each getter coerces accordingly; a JSON `null` (or absent
/// key) is `None`.
pub struct D1Row(serde_json::Map<String, Json>);

impl D1Row {
    fn field(&self, col: &str) -> Option<&Json> {
        self.0.get(col).filter(|v| !v.is_null())
    }
}

impl Row for D1Row {
    fn get_bool(&self, col: &str) -> Result<Option<bool>> {
        match self.field(col) {
            None => Ok(None),
            Some(Json::Bool(b)) => Ok(Some(*b)),
            Some(Json::Number(n)) => Ok(Some(n.as_i64().unwrap_or(0) != 0)),
            Some(other) => Err(type_err(col, "bool", other)),
        }
    }

    fn get_i32(&self, col: &str) -> Result<Option<i32>> {
        Ok(self.get_i64(col)?.map(|v| v as i32))
    }

    fn get_i64(&self, col: &str) -> Result<Option<i64>> {
        match self.field(col) {
            None => Ok(None),
            Some(Json::Number(n)) => n
                .as_i64()
                .map(Some)
                .ok_or_else(|| Error::msg(format!("column `{col}` is not an integer"))),
            Some(other) => Err(type_err(col, "integer", other)),
        }
    }

    fn get_u32(&self, col: &str) -> Result<Option<u32>> {
        Ok(self.get_i64(col)?.map(|v| v as u32))
    }

    fn get_u64(&self, col: &str) -> Result<Option<u64>> {
        Ok(self.get_i64(col)?.map(|v| v as u64))
    }

    fn get_f64(&self, col: &str) -> Result<Option<f64>> {
        match self.field(col) {
            None => Ok(None),
            Some(Json::Number(n)) => Ok(n.as_f64()),
            Some(other) => Err(type_err(col, "real", other)),
        }
    }

    fn get_string(&self, col: &str) -> Result<Option<String>> {
        match self.field(col) {
            None => Ok(None),
            Some(Json::String(s)) => Ok(Some(s.clone())),
            Some(other) => Err(type_err(col, "text", other)),
        }
    }

    fn get_bytes(&self, col: &str) -> Result<Option<Vec<u8>>> {
        match self.field(col) {
            None => Ok(None),
            // D1 hands a BLOB back as a JSON array of byte values.
            Some(Json::Array(a)) => {
                let bytes = a
                    .iter()
                    .map(|x| {
                        x.as_u64().filter(|n| *n <= 255).map(|n| n as u8).ok_or_else(|| {
                            Error::msg(format!("column `{col}` has a non-byte in its BLOB array"))
                        })
                    })
                    .collect::<Result<Vec<u8>>>()?;
                Ok(Some(bytes))
            }
            Some(other) => Err(type_err(col, "blob", other)),
        }
    }

    fn get_datetime(&self, col: &str) -> Result<Option<DateTime<Utc>>> {
        match self.get_string(col)? {
            None => Ok(None),
            Some(s) => {
                DateTime::parse_from_rfc3339(&s).map(|dt| Some(dt.with_timezone(&Utc))).map_err(
                    |e| Error::msg(format!("column `{col}` is not an RFC3339 timestamp: {e}")),
                )
            }
        }
    }

    fn get_naive_date(&self, col: &str) -> Result<Option<NaiveDate>> {
        match self.get_string(col)? {
            None => Ok(None),
            Some(s) => NaiveDate::parse_from_str(&s, "%Y-%m-%d")
                .map(Some)
                .map_err(|e| Error::msg(format!("column `{col}` is not a YYYY-MM-DD date: {e}"))),
        }
    }
}

fn type_err(col: &str, want: &str, got: &Json) -> Error {
    Error::msg(format!("column `{col}`: expected {want}, got {got}"))
}

/// Lower sea-query bind values to the JSON D1 accepts as `params`. D1 binds
/// null / number / string / and (for BLOBs) an array of byte values; chrono
/// values go as their SQLite text form.
fn to_json_params(values: Vec<Value>) -> Result<Vec<Json>> {
    values.into_iter().map(value_to_json).collect()
}

fn value_to_json(v: Value) -> Result<Json> {
    use rainier_orm::sea_query::Value as V;
    Ok(match v {
        V::Bool(o) => o.map(|b| Json::from(b as i64)).unwrap_or(Json::Null),
        V::TinyInt(o) => num(o),
        V::SmallInt(o) => num(o),
        V::Int(o) => num(o),
        V::BigInt(o) => num(o),
        V::TinyUnsigned(o) => num(o),
        V::SmallUnsigned(o) => num(o),
        V::Unsigned(o) => num(o),
        V::BigUnsigned(o) => o.map(Json::from).unwrap_or(Json::Null),
        V::Float(o) => o.map(|f| Json::from(f as f64)).unwrap_or(Json::Null),
        V::Double(o) => o.map(Json::from).unwrap_or(Json::Null),
        V::String(o) => o.map(|s| Json::String(*s)).unwrap_or(Json::Null),
        V::Char(o) => o.map(|c| Json::String(c.to_string())).unwrap_or(Json::Null),
        V::Bytes(o) => {
            o.map(|b| Json::Array(b.iter().map(|x| Json::from(*x)).collect())).unwrap_or(Json::Null)
        }
        V::ChronoDateTimeUtc(o) => o.map(|dt| Json::String(dt.to_rfc3339())).unwrap_or(Json::Null),
        V::ChronoDate(o) => {
            o.map(|d| Json::String(d.format("%Y-%m-%d").to_string())).unwrap_or(Json::Null)
        }
        other => return Err(Error::msg(format!("D1 bind: unsupported sea-query value {other:?}"))),
    })
}

fn num<T: Into<i64>>(o: Option<T>) -> Json {
    o.map(|n| Json::from(n.into())).unwrap_or(Json::Null)
}

#[cfg(test)]
mod tests {
    use super::*;
    use serde_json::json;

    #[test]
    fn parses_query_envelope() {
        let env = json!({
            "success": true,
            "result": [{
                "results": [
                    {"id": 1, "name": "a", "note": null},
                    {"id": 2, "name": "b", "note": "x"}
                ],
                "meta": {"last_row_id": 2, "changes": 0}
            }]
        });
        let resp = D1Response::from_envelope(&env).unwrap();
        assert_eq!(resp.rows.len(), 2);
        assert_eq!(resp.last_row_id, 2);
    }

    #[test]
    fn surfaces_failed_query() {
        let env = json!({"success": false, "errors": [{"message": "boom"}]});
        assert!(D1Response::from_envelope(&env).is_err());
    }

    #[test]
    fn d1row_decodes_sqlite_json_typing() {
        let obj = json!({
            "flag": 1,            // bool as 0/1
            "n": 42,
            "big": 9000000000i64,
            "r": 1.5,
            "s": "hello",
            "blob": [1, 2, 255],
            "ts": "2026-06-03T12:00:00+00:00",
            "d": "2026-06-03",
            "missing": null
        })
        .as_object()
        .unwrap()
        .clone();
        let row = D1Row(obj);
        assert_eq!(row.get_bool("flag").unwrap(), Some(true));
        assert_eq!(row.get_i32("n").unwrap(), Some(42));
        assert_eq!(row.get_u64("big").unwrap(), Some(9_000_000_000));
        assert_eq!(row.get_f64("r").unwrap(), Some(1.5));
        assert_eq!(row.get_string("s").unwrap().as_deref(), Some("hello"));
        assert_eq!(row.get_bytes("blob").unwrap(), Some(vec![1u8, 2, 255]));
        assert!(row.get_datetime("ts").unwrap().is_some());
        assert!(row.get_naive_date("d").unwrap().is_some());
        assert_eq!(row.get_string("missing").unwrap(), None);
        assert_eq!(row.get_string("absent").unwrap(), None);
    }

    #[test]
    fn lowers_bind_values_to_json() {
        let params = to_json_params(vec![
            Value::BigUnsigned(Some(7)),
            Value::String(Some(Box::new("hi".into()))),
            Value::Bool(Some(true)),
            Value::Bytes(Some(Box::new(vec![0u8, 9, 255]))),
            Value::Int(None),
        ])
        .unwrap();
        assert_eq!(params[0], json!(7));
        assert_eq!(params[1], json!("hi"));
        assert_eq!(params[2], json!(1)); // bools bind as 0/1 for SQLite
        assert_eq!(params[3], json!([0, 9, 255]));
        assert_eq!(params[4], Json::Null);
    }
}