Typed native database responses in Rust

Copy the complete module below into src/native_db_response.rs and add mod native_db_response; to your application. It uses the existing blinkhost-app-support 0.1.0 package. Keep your application manifest and complete lockfile; see Rust application support for configuration. No new package or compiler version is required.

This example decodes strings returned directly by blinkhost_sdk::database::{query,write,transaction,status}. It does not handle the older top-level query API. A native DEFAULT binding and database permission remain required. The native database reference covers parameterized SQL, scope, operation IDs and recovery.

Use the decoder

These snippets run inside application code. raw_read comes from a native query; raw_mutation comes from a write, transaction or status call. Handle SDK errors before decoding.

let read = native_db_response::read(&raw_read, statement_count)?;
let new_scope = read.primary_scope()?.clone();
// Map read.results only after checking expected columns, row count and cell types.
// Persist new_scope with a NEW operation ID and exact payload before first send.

let receipt = native_db_response::mutation(
    &raw_mutation, &saved_scope, &saved_operation_id, saved_statement_count,
)?;
match receipt.outcome {
    native_db_response::Outcome::Committed => { /* commitment is proven */ }
    native_db_response::Outcome::NotCommitted => { /* this attempt rolled back */ }
    native_db_response::Outcome::Unknown => { /* retain the original operation */ }
}

saved_scope is the original persisted native_db_response::Scope, including on status calls. Check receipt.error even after commitment: expired or policy-restricted result bodies can be unavailable while the write remains committed. Empty successful SELECT rows differ from a failed read. Integer cells are exact i64 values; blobs are bytes; duplicate column names retain their positions. Application mapping must still validate the expected schema and record ownership.

Safety and limits

A transport or decoding error after a mutation leaves commitment unknown. Do not replay it with a new scope, ID or payload. The decoder never retries, performs I/O, authorizes users or persists operations. Reusing an operation ID with changed inputs can be rejected even if the original operation committed.

Only decode unmodified native host responses, never customer-supplied receipts or HTTP bodies. The underlying JSON parser uses ordinary last-value handling of duplicate object keys. Additional envelope/result metadata is accepted; tagged SQL cells are strict. This is an application example, not a new SDK API.

Responses are limited to 128 KiB plus 1 KiB of envelope allowance, 32 statements, 128 columns and 1,000 total rows; text and decoded blobs are limited to 64 KiB. Module policy may be lower. JSON and typed results coexist briefly in guest memory: paginate large queries. Local parser tests do not attest live database availability.

Complete reusable module

//! Copy into an app using blinkhost-app-support 0.1.0. Only decode native host
//! responses here, never arbitrary request bodies. This performs no I/O/retries.
use blinkhost_app_support::{parse_json, Value};

const MAX_RESPONSE: usize = 128 * 1024 + 1024;
const MAX_VALUE: usize = 64 * 1024;

#[derive(Debug, PartialEq, Eq)]
pub enum Error {
    InvalidJson,
    InvalidResponse,
    InvalidValue,
    LimitExceeded,
    IdentityMismatch,
    PrimaryRequired,
    Database(String),
}

#[derive(Clone, Debug, PartialEq, Eq)]
pub struct Scope {
    pub database_id: String,
    pub generation: String,
}

#[derive(Debug, PartialEq)]
pub enum SqlValue {
    Null,
    Integer(i64),
    Real(f64),
    Text(String),
    Blob(Vec<u8>),
}

#[derive(Debug, PartialEq)]
pub struct StatementResult {
    pub columns: Vec<String>,
    pub rows: Vec<Vec<SqlValue>>,
    pub affected_rows: u64,
}

#[derive(Debug, PartialEq)]
pub struct ReadResponse {
    pub scope: Scope,
    pub served_by: String,
    pub replica_sync_age_ms: Option<u64>,
    pub results: Vec<StatementResult>,
}

impl ReadResponse {
    /// Only a successful primary read can supply a new mutation's scope.
    pub fn primary_scope(&self) -> Result<&Scope, Error> {
        if self.served_by != "primary" {
            return Err(Error::PrimaryRequired);
        }
        Ok(&self.scope)
    }
}

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum Outcome {
    Committed,
    NotCommitted,
    Unknown,
}

#[derive(Debug, PartialEq)]
pub struct MutationResponse {
    pub scope: Option<Scope>,
    pub operation_id: String,
    pub outcome: Outcome,
    pub results: Vec<StatementResult>,
    /// Commitment can be proven even when the result body is no longer available.
    pub error: Option<String>,
}

fn text(value: &Value) -> Result<&str, Error> {
    value.as_str().ok_or(Error::InvalidResponse)
}

fn field<'a>(value: &'a Value, name: &str) -> Result<&'a Value, Error> {
    value.get(name).ok_or(Error::InvalidResponse)
}

fn optional_error(value: &Value) -> Result<Option<String>, Error> {
    value
        .get("error")
        .map(|v| {
            let code = text(v)?;
            if code.is_empty() || code.len() > 128 {
                return Err(Error::InvalidResponse);
            }
            Ok(code.to_owned())
        })
        .transpose()
}

fn uuid(value: &str) -> bool {
    value.len() == 36
        && value.bytes().enumerate().all(|(i, b)| {
            if matches!(i, 8 | 13 | 18 | 23) {
                b == b'-'
            } else {
                b.is_ascii_digit() || (b'a'..=b'f').contains(&b)
            }
        })
}

fn scope(value: &Value) -> Result<Option<Scope>, Error> {
    let id = field(value, "database_id")?;
    let generation = field(value, "generation")?;
    if id.is_null() && generation.is_null() {
        return Ok(None);
    }
    let id = text(id)?;
    let generation = text(generation)?;
    if !uuid(id) || !uuid(generation) {
        return Err(Error::InvalidResponse);
    }
    Ok(Some(Scope {
        database_id: id.into(),
        generation: generation.into(),
    }))
}

fn document(raw: &str) -> Result<Value, Error> {
    if raw.len() > MAX_RESPONSE {
        return Err(Error::LimitExceeded);
    }
    let value = parse_json(raw.as_bytes()).map_err(|_| Error::InvalidJson)?;
    if !value.is_object() || !field(&value, "response")?.is_object() {
        return Err(Error::InvalidResponse);
    }
    Ok(value)
}

fn sextet(b: u8) -> Result<u8, Error> {
    match b {
        b'A'..=b'Z' => Ok(b - b'A'),
        b'a'..=b'z' => Ok(b - b'a' + 26),
        b'0'..=b'9' => Ok(b - b'0' + 52),
        b'+' => Ok(62),
        b'/' => Ok(63),
        _ => Err(Error::InvalidValue),
    }
}

fn blob(encoded: &str) -> Result<Vec<u8>, Error> {
    let bytes = encoded.as_bytes();
    if bytes.len() > MAX_VALUE.div_ceil(3) * 4 {
        return Err(Error::LimitExceeded);
    }
    if bytes.len() % 4 != 0 {
        return Err(Error::InvalidValue);
    }
    let mut out = Vec::with_capacity(bytes.len() / 4 * 3);
    for (i, chunk) in bytes.chunks_exact(4).enumerate() {
        let a = sextet(chunk[0])?;
        let b = sextet(chunk[1])?;
        out.push((a << 2) | (b >> 4));
        if chunk[2] == b'=' {
            if chunk[3] != b'=' || b & 15 != 0 || (i + 1) * 4 != bytes.len() {
                return Err(Error::InvalidValue);
            }
        } else {
            let c = sextet(chunk[2])?;
            out.push((b << 4) | (c >> 2));
            if chunk[3] == b'=' {
                if c & 3 != 0 || (i + 1) * 4 != bytes.len() {
                    return Err(Error::InvalidValue);
                }
            } else {
                out.push((c << 6) | sextet(chunk[3])?);
            }
        }
    }
    if out.len() > MAX_VALUE {
        return Err(Error::LimitExceeded);
    }
    Ok(out)
}

fn cell(value: &Value) -> Result<SqlValue, Error> {
    let object = value.as_object().ok_or(Error::InvalidValue)?;
    let tag = text(field(value, "type")?)?;
    if tag == "null" && object.len() == 1 {
        return Ok(SqlValue::Null);
    }
    if object.len() != 2 {
        return Err(Error::InvalidValue);
    }
    let data = field(value, "value")?;
    match tag {
        "integer" => {
            let raw = text(data)?;
            let number: i64 = raw.parse().map_err(|_| Error::InvalidValue)?;
            if number.to_string() != raw {
                return Err(Error::InvalidValue);
            }
            Ok(SqlValue::Integer(number))
        }
        "real" => data
            .as_f64()
            .filter(|v| v.is_finite())
            .map(SqlValue::Real)
            .ok_or(Error::InvalidValue),
        "text" if text(data)?.len() <= MAX_VALUE => Ok(SqlValue::Text(text(data)?.into())),
        "blob" => blob(text(data)?).map(SqlValue::Blob),
        _ => Err(Error::InvalidValue),
    }
}

fn results(response: &Value) -> Result<Vec<StatementResult>, Error> {
    let items = field(response, "results")?
        .as_array()
        .ok_or(Error::InvalidResponse)?;
    if items.len() > 32 {
        return Err(Error::LimitExceeded);
    }
    let mut total_rows = 0;
    items
        .iter()
        .map(|item| {
            let names = field(item, "columns")?
                .as_array()
                .ok_or(Error::InvalidResponse)?;
            if names.len() > 128 {
                return Err(Error::LimitExceeded);
            }
            let columns = names
                .iter()
                .map(|v| text(v).map(str::to_owned))
                .collect::<Result<Vec<_>, _>>()?;
            let input = field(item, "rows")?
                .as_array()
                .ok_or(Error::InvalidResponse)?;
            total_rows += input.len();
            if total_rows > 1000 {
                return Err(Error::LimitExceeded);
            }
            let rows = input
                .iter()
                .map(|row| {
                    let cells = row.as_array().ok_or(Error::InvalidResponse)?;
                    if cells.len() != columns.len() {
                        return Err(Error::InvalidResponse);
                    }
                    cells.iter().map(cell).collect()
                })
                .collect::<Result<Vec<_>, Error>>()?;
            let affected_rows = field(item, "affected_rows")?
                .as_u64()
                .ok_or(Error::InvalidResponse)?;
            Ok(StatementResult {
                columns,
                rows,
                affected_rows,
            })
        })
        .collect()
}

fn statement_count(count: usize) -> Result<(), Error> {
    if !(1..=32).contains(&count) {
        return Err(Error::InvalidResponse);
    }
    Ok(())
}

/// Decode the host response to a read of exactly expected_statements statements.
/// An empty SELECT is a success; a failed read never becomes an empty success.
pub fn read(raw: &str, expected_statements: usize) -> Result<ReadResponse, Error> {
    statement_count(expected_statements)?;
    let doc = document(raw)?;
    let outer_scope = scope(&doc)?;
    let response = field(&doc, "response")?;
    let results = results(response)?;
    if let Some(code) = optional_error(response)? {
        if !results.is_empty() {
            return Err(Error::InvalidResponse);
        }
        return Err(Error::Database(code));
    }
    let scope = outer_scope.ok_or(Error::InvalidResponse)?;
    if self::scope(response)?.as_ref() != Some(&scope) {
        return Err(Error::IdentityMismatch);
    }
    let served_by = text(field(response, "served_by")?)?;
    if !matches!(served_by, "primary" | "replica")
        || results.len() != expected_statements
        || results.iter().any(|r| r.affected_rows != 0)
    {
        return Err(Error::InvalidResponse);
    }
    let replica_sync_age_ms = response
        .get("replica_sync_age_ms")
        .map(|v| v.as_u64().ok_or(Error::InvalidResponse))
        .transpose()?;
    Ok(ReadResponse {
        scope,
        served_by: served_by.into(),
        replica_sync_age_ms,
        results,
    })
}

/// Decode write/transaction/status using the ORIGINAL stored scope, operation ID
/// and statement count. Any decoding/transport failure leaves commitment unknown.
pub fn mutation(
    raw: &str,
    expected: &Scope,
    operation_id: &str,
    expected_statements: usize,
) -> Result<MutationResponse, Error> {
    statement_count(expected_statements)?;
    if !uuid(&expected.database_id)
        || !uuid(&expected.generation)
        || !(16..=128).contains(&operation_id.len())
        || !operation_id
            .bytes()
            .all(|b| b.is_ascii_alphanumeric() || b"-_:.".contains(&b))
    {
        return Err(Error::InvalidResponse);
    }
    let doc = document(raw)?;
    let scope = scope(&doc)?;
    if scope.as_ref().is_some_and(|actual| actual != expected) {
        return Err(Error::IdentityMismatch);
    }
    let response = field(&doc, "response")?;
    if text(field(response, "operation_id")?)? != operation_id {
        return Err(Error::IdentityMismatch);
    }
    let outcome = match text(field(response, "outcome")?)? {
        "committed" => Outcome::Committed,
        "not_committed" => Outcome::NotCommitted,
        "unknown" => Outcome::Unknown,
        _ => return Err(Error::InvalidResponse),
    };
    let results = results(response)?;
    let error = optional_error(response)?;
    if scope.is_none() && outcome != Outcome::Unknown {
        return Err(Error::InvalidResponse);
    }
    if error.is_some() {
        if !results.is_empty() {
            return Err(Error::InvalidResponse);
        }
    } else if outcome != Outcome::Committed || results.len() != expected_statements {
        return Err(Error::InvalidResponse);
    }
    Ok(MutationResponse {
        scope,
        operation_id: operation_id.into(),
        outcome,
        results,
        error,
    })
}