use cratefield_core::{Row, Statement};
use sea_query::{Alias, Cond, Expr, Order, Query, SimpleExpr, Value as SeaValue};
use serde_json::{Map, Value};
use crate::schema::{FieldDef, FieldKind, TableDef};
use crate::validate::{RowErrors, validate_row};
use crate::value::ErrorCode;
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct DecodeError {
pub table: String,
pub column: String,
pub detail: String,
}
impl std::fmt::Display for DecodeError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(
f,
"table `{}`, column `{}`: {}",
self.table, self.column, self.detail
)
}
}
impl std::error::Error for DecodeError {}
#[must_use]
pub fn to_sql(field: &FieldDef, value: Option<&Value>) -> Option<SeaValue> {
let Some(value) = value.filter(|value| !value.is_null()) else {
return Some(null_for(&field.kind));
};
match &field.kind {
FieldKind::Boolean => value.as_bool().map(|flag| SeaValue::Bool(Some(flag))),
FieldKind::Integer { .. } => {
crate::value::integral(value).map(|n| SeaValue::BigInt(Some(n)))
}
FieldKind::Real { .. } => value.as_f64().map(|n| SeaValue::Double(Some(n))),
FieldKind::Json => Some(SeaValue::String(Some(Box::new(value.to_string())))),
FieldKind::Text { .. }
| FieldKind::Timestamp
| FieldKind::Uuid
| FieldKind::Enum { .. } => value
.as_str()
.map(|text| SeaValue::String(Some(Box::new(text.to_owned())))),
}
}
fn bound(field: &FieldDef, value: Option<&Value>) -> Result<SeaValue, RowErrors> {
to_sql(field, value).ok_or_else(|| {
RowErrors::one(
&field.name,
ErrorCode::WrongType,
format!("must be {}", field.kind.as_str()),
)
})
}
fn null_for(kind: &FieldKind) -> SeaValue {
match kind {
FieldKind::Boolean => SeaValue::Bool(None),
FieldKind::Integer { .. } => SeaValue::BigInt(None),
FieldKind::Real { .. } => SeaValue::Double(None),
FieldKind::Text { .. }
| FieldKind::Timestamp
| FieldKind::Uuid
| FieldKind::Json
| FieldKind::Enum { .. } => SeaValue::String(None),
}
}
pub fn from_sql(table: &str, field: &FieldDef, value: &SeaValue) -> Result<Value, DecodeError> {
let wrong = |detail: &str| DecodeError {
table: table.to_owned(),
column: field.name.clone(),
detail: detail.to_owned(),
};
if matches!(value, SeaValue::String(None)) {
return Ok(Value::Null);
}
match &field.kind {
FieldKind::Boolean => match value {
SeaValue::Bool(None) | SeaValue::Int(None) | SeaValue::BigInt(None) => Ok(Value::Null),
SeaValue::Bool(Some(flag)) => Ok(Value::Bool(*flag)),
SeaValue::Int(Some(n)) => Ok(Value::Bool(*n != 0)),
SeaValue::BigInt(Some(n)) => Ok(Value::Bool(*n != 0)),
_ => Err(wrong(
"declared boolean, and the database did not answer with one",
)),
},
FieldKind::Integer { .. } => match integer(value) {
Integral::Whole(n) => Ok(Value::from(n)),
Integral::Null => Ok(Value::Null),
Integral::Other => Err(wrong(
"declared integer, and the database did not answer with one",
)),
},
FieldKind::Real { .. } => match value {
SeaValue::Double(None) | SeaValue::Float(None) => Ok(Value::Null),
SeaValue::Double(Some(n)) => Ok(number(*n).ok_or_else(|| {
wrong("declared real, and the database answered with a value JSON cannot hold")
})?),
SeaValue::Float(Some(n)) => Ok(number(f64::from(*n)).ok_or_else(|| {
wrong("declared real, and the database answered with a value JSON cannot hold")
})?),
_ => match integer(value) {
#[expect(
clippy::cast_precision_loss,
reason = "a REAL column's value was written as an f64; coming back through \
SQLite's integer affinity, f64 is the kind it is declared as"
)]
Integral::Whole(n) => {
Ok(number(n as f64).ok_or_else(|| wrong("not a JSON number"))?)
}
Integral::Null => Ok(Value::Null),
Integral::Other => Err(wrong(
"declared real, and the database did not answer with a number",
)),
},
},
FieldKind::Json => match value {
SeaValue::String(None) => Ok(Value::Null),
SeaValue::String(Some(text)) => serde_json::from_str(text).map_err(|_ignored| {
wrong("declared json, and the stored text is not JSON")
}),
_ => Err(wrong(
"declared json, and the database did not answer with text",
)),
},
FieldKind::Text { .. }
| FieldKind::Timestamp
| FieldKind::Uuid
| FieldKind::Enum { .. } => match value {
SeaValue::String(None) => Ok(Value::Null),
SeaValue::String(Some(text)) => Ok(Value::String((**text).clone())),
_ => Err(wrong(
"declared text, and the database did not answer with text",
)),
},
}
}
enum Integral {
Whole(i64),
Null,
Other,
}
fn integer(value: &SeaValue) -> Integral {
let inner = match value {
SeaValue::TinyInt(inner) => inner.map(i64::from),
SeaValue::SmallInt(inner) => inner.map(i64::from),
SeaValue::Int(inner) => inner.map(i64::from),
SeaValue::BigInt(inner) => *inner,
_ => return Integral::Other,
};
inner.map_or(Integral::Null, Integral::Whole)
}
fn number(n: f64) -> Option<Value> {
serde_json::Number::from_f64(n).map(Value::Number)
}
pub fn row_json(table: &TableDef, row: &Row) -> Result<Value, DecodeError> {
let mut out = Map::new();
for field in &table.fields {
let value = row
.get::<SeaValue>(&field.name)
.ok_or_else(|| DecodeError {
table: table.name.clone(),
column: field.name.clone(),
detail: "declared, and the row does not have it".to_owned(),
})?;
out.insert(field.name.clone(), from_sql(&table.name, field, &value)?);
}
Ok(Value::Object(out))
}
pub fn insert(table: &TableDef, row: &Value) -> Result<Statement, RowErrors> {
validate_row(table, row)?;
let object = row.as_object().cloned().unwrap_or_default();
let present: Vec<&FieldDef> = table
.fields
.iter()
.filter(|field| object.contains_key(&field.name))
.collect();
let mut values = Vec::with_capacity(present.len());
for field in &present {
values.push(bound(field, object.get(&field.name))?.into());
}
let mut insert = Query::insert();
insert
.into_table(Alias::new(&table.name))
.columns(present.iter().map(|field| Alias::new(&field.name)))
.values_panic(values);
Ok(Statement::render(&insert))
}
#[derive(Debug, Clone, Copy)]
pub struct Owned<'a> {
pub column: &'a str,
pub subject: &'a Value,
}
pub fn select_one(
table: &TableDef,
key: &Value,
owned: Option<Owned<'_>>,
) -> Result<Statement, DecodeError> {
let mut select = Query::select();
select
.columns(table.fields.iter().map(|field| Alias::new(&field.name)))
.from(Alias::new(&table.name))
.cond_where(key_predicate(table, key)?.add_option(owned_predicate(table, owned)?))
.limit(1);
Ok(Statement::render(&select))
}
fn owned_predicate(
table: &TableDef,
owned: Option<Owned<'_>>,
) -> Result<Option<sea_query::SimpleExpr>, DecodeError> {
let Some(owned) = owned else {
return Ok(None);
};
let field = table
.fields
.iter()
.find(|field| field.name == owned.column)
.ok_or_else(|| DecodeError {
table: table.name.clone(),
column: owned.column.to_owned(),
detail: "the subject column this read is scoped to is not a field of the table"
.to_owned(),
})?;
let value = to_sql(field, Some(owned.subject)).ok_or_else(|| DecodeError {
table: table.name.clone(),
column: owned.column.to_owned(),
detail: format!(
"the subject this read is scoped to is not {}",
field.kind.as_str()
),
})?;
Ok(Some(Expr::col(Alias::new(owned.column)).eq(value)))
}
#[derive(Debug, Clone)]
pub struct Filter {
pub column: String,
pub value: Value,
}
#[derive(Debug, Clone, Copy)]
pub struct Page<'a> {
pub limit: u64,
pub after: Option<&'a Value>,
pub owned: Option<Owned<'a>>,
pub filters: &'a [Filter],
pub sort: Option<Sort<'a>>,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct Sort<'a> {
pub column: &'a str,
pub descending: bool,
}
pub fn select_page(table: &TableDef, page: Page<'_>) -> Result<Statement, DecodeError> {
let mut select = Query::select();
select
.columns(table.fields.iter().map(|field| Alias::new(&field.name)))
.from(Alias::new(&table.name))
.limit(page.limit);
let direction = if page.sort.is_some_and(|sort| sort.descending) {
Order::Desc
} else {
Order::Asc
};
if let Some(sort) = page.sort {
sortable(table, sort)?;
select.order_by(Alias::new(sort.column), direction.clone());
}
for column in &table.primary_key {
select.order_by(Alias::new(column), direction.clone());
}
let mut all = Cond::all().add_option(owned_predicate(table, page.owned)?);
if let Some(after) = page.after {
all = all.add(after_predicate(table, after, page.sort)?);
}
for filter in page.filters {
all = all.add(filter_predicate(table, filter)?);
}
select.cond_where(all);
Ok(Statement::render(&select))
}
fn filter_predicate(table: &TableDef, filter: &Filter) -> Result<SimpleExpr, DecodeError> {
let field = table
.fields
.iter()
.find(|field| field.name == filter.column)
.ok_or_else(|| DecodeError {
table: table.name.clone(),
column: filter.column.clone(),
detail: "is not a field of this table".to_owned(),
})?;
if filter.value.is_null() {
return Ok(Expr::col(Alias::new(&filter.column)).is_null());
}
if matches!(field.kind, FieldKind::Json) {
return Err(DecodeError {
table: table.name.clone(),
column: filter.column.clone(),
detail: "is json, which has no single text form to compare for equality; reading \
inside a JSON column is a query a module or a sidecar writes"
.to_owned(),
});
}
let value = to_sql(field, Some(&filter.value)).ok_or_else(|| DecodeError {
table: table.name.clone(),
column: filter.column.clone(),
detail: format!("is {} and the value given is not", field.kind.as_str()),
})?;
Ok(Expr::col(Alias::new(&filter.column)).eq(value))
}
pub fn update(
table: &TableDef,
key: &Value,
row: &Value,
owned: Option<Owned<'_>>,
) -> Result<Statement, UpdateError> {
validate_row(table, row).map_err(UpdateError::Row)?;
let object = row.as_object().cloned().unwrap_or_default();
let predicate = key_predicate(table, key)
.map_err(UpdateError::Key)?
.add_option(owned_predicate(table, owned).map_err(UpdateError::Key)?);
let mut update = Query::update();
update.table(Alias::new(&table.name));
for field in &table.fields {
if table.primary_key.contains(&field.name) {
continue;
}
update.value(
Alias::new(&field.name),
bound(field, object.get(&field.name)).map_err(UpdateError::Row)?,
);
}
update.cond_where(predicate);
Ok(Statement::render(&update))
}
#[derive(Debug, Clone, PartialEq, Eq)]
#[non_exhaustive]
pub enum UpdateError {
Row(RowErrors),
Key(DecodeError),
}
impl std::fmt::Display for UpdateError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Self::Row(errors) => write!(f, "{errors}"),
Self::Key(error) => write!(f, "{error}"),
}
}
}
impl std::error::Error for UpdateError {}
pub fn delete(
table: &TableDef,
key: &Value,
owned: Option<Owned<'_>>,
) -> Result<Statement, DecodeError> {
let mut delete = Query::delete();
delete
.from_table(Alias::new(&table.name))
.cond_where(key_predicate(table, key)?.add_option(owned_predicate(table, owned)?));
Ok(Statement::render(&delete))
}
fn key_predicate(table: &TableDef, key: &Value) -> Result<Cond, DecodeError> {
let mut all = Cond::all();
for column in &table.primary_key {
all = all.add(Expr::col(Alias::new(column)).eq(key_part(table, key, column)?));
}
Ok(all)
}
fn after_predicate(
table: &TableDef,
after: &Value,
sort: Option<Sort<'_>>,
) -> Result<Cond, DecodeError> {
let mut columns: Vec<&str> = Vec::with_capacity(table.primary_key.len() + 1);
if let Some(sort) = sort {
columns.push(sort.column);
}
columns.extend(table.primary_key.iter().map(String::as_str));
let descending = sort.is_some_and(|sort| sort.descending);
let mut any = Cond::any();
for (at, column) in columns.iter().enumerate() {
let mut branch = Cond::all();
for earlier in &columns[..at] {
branch =
branch.add(Expr::col(Alias::new(*earlier)).eq(cursor_part(table, after, earlier)?));
}
let value = cursor_part(table, after, column)?;
let beyond = if descending {
Expr::col(Alias::new(*column)).lt(value)
} else {
Expr::col(Alias::new(*column)).gt(value)
};
any = any.add(branch.add(beyond));
}
Ok(any)
}
fn sortable(table: &TableDef, sort: Sort<'_>) -> Result<(), DecodeError> {
let field = table
.fields
.iter()
.find(|field| field.name == sort.column)
.ok_or_else(|| DecodeError {
table: table.name.clone(),
column: sort.column.to_owned(),
detail: "is not a field of this table".to_owned(),
})?;
let keyed = table.primary_key.iter().any(|key| key == sort.column);
if field.required || keyed {
return Ok(());
}
Err(DecodeError {
table: table.name.clone(),
column: sort.column.to_owned(),
detail: "is optional, and the engines disagree about where nulls sort — a page ordered by \
it would differ between two deployments of one declaration"
.to_owned(),
})
}
fn cursor_part(table: &TableDef, cursor: &Value, column: &str) -> Result<SeaValue, DecodeError> {
key_part(table, cursor, column)
}
fn key_part(table: &TableDef, key: &Value, column: &str) -> Result<SeaValue, DecodeError> {
let missing = |detail: &str| DecodeError {
table: table.name.clone(),
column: column.to_owned(),
detail: detail.to_owned(),
};
let field = table
.fields
.iter()
.find(|field| field.name == column)
.ok_or_else(|| missing("a primary-key column that is not a declared field"))?;
let value = key
.get(column)
.filter(|value| !value.is_null())
.ok_or_else(|| missing("part of the primary key, and the key does not give it"))?;
to_sql(field, Some(value)).ok_or_else(|| {
missing(&format!(
"part of the primary key, and the value given is not {}",
field.kind.as_str()
))
})
}