teaql_provider_sqlite/
lib.rs

1use std::collections::{BTreeMap, BTreeSet};
2use std::future::Future;
3use std::pin::Pin;
4use std::str::FromStr;
5use std::sync::{Arc, Mutex, MutexGuard};
6
7use chrono::{DateTime, NaiveDate, NaiveDateTime};
8use rusqlite::types::{Value as SqliteValue, ValueRef};
9use rusqlite::{Connection, Row, params_from_iter};
10use rust_decimal::Decimal;
11use teaql_core::{
12    DataType, EntityDescriptor, Expr, InsertCommand, PropertyDescriptor, Record, SelectQuery,
13    UpdateCommand, Value,
14};
15use teaql_runtime::{
16    GraphNode, InternalIdGenerator, RawAuditEvent, RuntimeError, SchemaProvider, UserContext,
17};
18use teaql_sql::{
19    CompiledQuery, DatabaseKind, SqlCompileError, SqlDialect, SqlTransport,
20    quote_identifier_if_needed,
21};
22
23pub const DEFAULT_ID_SPACE_TABLE: &str = "teaql_id_space";
24
25const SQLITE_DECIMAL_PREFIX: &str = "__teaql_decimal__:";
26
27#[derive(Debug, Default, Clone, Copy)]
28pub struct SqliteDialect;
29
30impl SqlDialect for SqliteDialect {
31    fn kind(&self) -> DatabaseKind {
32        DatabaseKind::Sqlite
33    }
34
35    fn quote_ident(&self, ident: &str) -> String {
36        quote_ident(ident)
37    }
38
39    fn placeholder(&self, _index: usize) -> String {
40        "?".to_owned()
41    }
42
43    fn schema_type_sql(
44        &self,
45        data_type: DataType,
46        property: &PropertyDescriptor,
47    ) -> Result<&'static str, SqlCompileError> {
48        match data_type {
49            DataType::Bool => Ok("BOOLEAN"),
50            DataType::I64 | DataType::U64 if property.is_id => Ok("INTEGER"),
51            DataType::I64 | DataType::U64 => Ok("INTEGER"),
52            DataType::F64 => Ok("REAL"),
53            DataType::Decimal => Ok("NUMERIC"),
54            DataType::Text => Ok("VARCHAR(255)"),
55            DataType::LargeText => Ok("TEXT"),
56            DataType::Json => Ok("JSON"),
57            DataType::Date => Ok("DATE"),
58            DataType::Timestamp => Ok("TIMESTAMP"),
59        }
60    }
61
62    fn compile_add_column(
63        &self,
64        entity: &EntityDescriptor,
65        property: &PropertyDescriptor,
66    ) -> Result<String, SqlCompileError> {
67        // SQLite does not support adding NOT NULL columns without a DEFAULT.
68        // Since TeaQL enforces nullability at the application layer, we can safely
69        // strip the NOT NULL constraint when adding columns to existing tables.
70        let def = self.column_definition_sql(property)?;
71        let def_without_not_null = def.replace(" NOT NULL", "");
72
73        Ok(format!(
74            "ALTER TABLE {} ADD COLUMN {}",
75            self.quote_ident(&entity.table_name),
76            def_without_not_null
77        ))
78    }
79}
80
81#[derive(Debug)]
82pub enum MutationExecutorError {
83    Sqlite(rusqlite::Error),
84    SqlCompile(SqlCompileError),
85    UnsupportedValue(&'static str),
86    UnsupportedColumnType(String),
87    Bind(String),
88    Lock(String),
89}
90
91impl std::fmt::Display for MutationExecutorError {
92    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
93        match self {
94            Self::Sqlite(err) => err.fmt(f),
95            Self::SqlCompile(err) => err.fmt(f),
96            Self::UnsupportedValue(kind) => {
97                write!(
98                    f,
99                    "unsupported rusqlite bind value for mutation executor: {kind}"
100                )
101            }
102            Self::UnsupportedColumnType(kind) => {
103                write!(
104                    f,
105                    "unsupported rusqlite column type for record decoding: {kind}"
106                )
107            }
108            Self::Bind(message) => write!(f, "rusqlite bind error: {message}"),
109            Self::Lock(message) => write!(f, "rusqlite connection lock error: {message}"),
110        }
111    }
112}
113
114impl std::error::Error for MutationExecutorError {}
115
116impl From<rusqlite::Error> for MutationExecutorError {
117    fn from(value: rusqlite::Error) -> Self {
118        Self::Sqlite(value)
119    }
120}
121
122impl From<SqlCompileError> for MutationExecutorError {
123    fn from(value: SqlCompileError) -> Self {
124        Self::SqlCompile(value)
125    }
126}
127
128#[derive(Clone)]
129pub struct SqliteMutationExecutor {
130    connection: Arc<Mutex<Connection>>,
131}
132
133impl SqliteMutationExecutor {
134    pub fn new(connection: Arc<Mutex<Connection>>) -> Self {
135        Self { connection }
136    }
137
138    pub fn from_connection(connection: Connection) -> Self {
139        Self::new(Arc::new(Mutex::new(connection)))
140    }
141
142    pub fn connection(&self) -> Arc<Mutex<Connection>> {
143        Arc::clone(&self.connection)
144    }
145
146    pub fn ensure_schema(
147        &self,
148        dialect: &SqliteDialect,
149        entities: &[&EntityDescriptor],
150    ) -> Result<(), MutationExecutorError> {
151        self.ensure_id_space_table(DEFAULT_ID_SPACE_TABLE)?;
152
153        for entity in entities {
154            if !self.table_exists(&entity.table_name)? {
155                let sql = dialect.compile_create_table(entity)?;
156                self.lock()?.execute(&sql, [])?;
157                continue;
158            }
159
160            let existing_columns = self.table_columns(&entity.table_name)?;
161            for property in &entity.properties {
162                let bare_column = strip_identifier_quotes(&property.column_name).to_lowercase();
163                if existing_columns.contains(&bare_column) {
164                    continue;
165                }
166                let sql = dialect.compile_add_column(entity, property)?;
167                self.lock()?.execute(&sql, [])?;
168            }
169
170            for sql in dialect.schema_indexes_sqls(entity)? {
171                self.lock()?.execute(&sql, [])?;
172            }
173        }
174        Ok(())
175    }
176
177    pub fn ensure_id_space_table(&self, table_name: &str) -> Result<(), MutationExecutorError> {
178        let sql = format!(
179            "CREATE TABLE IF NOT EXISTS {} (type_name VARCHAR(100) PRIMARY KEY, current_level BIGINT NOT NULL)",
180            quote_ident(table_name)
181        );
182        self.lock()?.execute(&sql, [])?;
183        Ok(())
184    }
185
186    pub fn begin_transaction(&self) -> Result<(), MutationExecutorError> {
187        self.lock()?.execute("BEGIN IMMEDIATE", [])?;
188        Ok(())
189    }
190
191    pub fn commit_transaction(&self) -> Result<(), MutationExecutorError> {
192        self.lock()?.execute("COMMIT", [])?;
193        Ok(())
194    }
195
196    pub fn rollback_transaction(&self) -> Result<(), MutationExecutorError> {
197        self.lock()?.execute("ROLLBACK", [])?;
198        Ok(())
199    }
200
201    pub fn execute(&self, query: &CompiledQuery) -> Result<u64, MutationExecutorError> {
202        let params = bind_values(&query.params)?;
203        let rows = self
204            .lock()?
205            .execute(&query.sql_with_comment(), params_from_iter(params.iter()))?;
206        Ok(rows as u64)
207    }
208
209    pub fn fetch_all(&self, query: &CompiledQuery) -> Result<Vec<Record>, MutationExecutorError> {
210        let params = bind_values(&query.params)?;
211        let connection = self.lock()?;
212        let mut statement = connection.prepare(&query.sql_with_comment())?;
213        let columns = statement_columns(&statement);
214        let mut rows = statement.query(params_from_iter(params.iter()))?;
215        let mut records = Vec::new();
216        while let Some(row) = rows.next()? {
217            records.push(decode_sqlite_row(row, &columns)?);
218        }
219        Ok(records)
220    }
221
222    /// Fetch rows in streaming mode (chunked).
223    /// Returns a Vec of StreamChunk, each containing up to `chunk_size` rows.
224    pub fn fetch_stream(
225        &self,
226        query: &CompiledQuery,
227        chunk_size: usize,
228    ) -> Result<Vec<teaql_data_service::StreamChunk>, MutationExecutorError> {
229        let params = bind_values(&query.params)?;
230        let connection = self.lock()?;
231        let mut statement = connection.prepare(&query.sql_with_comment())?;
232        let columns = statement_columns(&statement);
233        let mut rows = statement.query(params_from_iter(params.iter()))?;
234
235        let mut chunks = Vec::new();
236        let mut current_chunk = Vec::new();
237        let mut chunk_index = 0;
238
239        while let Some(row) = rows.next()? {
240            current_chunk.push(decode_sqlite_row(row, &columns)?);
241            if current_chunk.len() >= chunk_size {
242                chunks.push(teaql_data_service::StreamChunk {
243                    rows: current_chunk,
244                    chunk_index,
245                    is_last: false,
246                });
247                current_chunk = Vec::new();
248                chunk_index += 1;
249            }
250        }
251
252        // Push the final chunk (may be empty if exactly aligned)
253        chunks.push(teaql_data_service::StreamChunk {
254            rows: current_chunk,
255            chunk_index,
256            is_last: true,
257        });
258
259        Ok(chunks)
260    }
261
262    pub fn table_exists(&self, table_name: &str) -> Result<bool, MutationExecutorError> {
263        let exists: i64 = self.lock()?.query_row(
264            "SELECT COUNT(1) FROM sqlite_master WHERE type = 'table' AND name = ?",
265            [table_name],
266            |row| row.get(0),
267        )?;
268        Ok(exists > 0)
269    }
270
271    pub fn table_columns(
272        &self,
273        table_name: &str,
274    ) -> Result<BTreeSet<String>, MutationExecutorError> {
275        let pragma_sql = format!("PRAGMA table_info({})", quote_ident(table_name));
276        let connection = self.lock()?;
277        let mut statement = connection.prepare(&pragma_sql)?;
278        let rows = statement.query_map([], |row| row.get::<_, String>("name"))?;
279        let mut columns = BTreeSet::new();
280        for row in rows {
281            columns.insert(row?.to_lowercase());
282        }
283        Ok(columns)
284    }
285
286    fn lock(&self) -> Result<MutexGuard<'_, Connection>, MutationExecutorError> {
287        self.connection
288            .lock()
289            .map_err(|err| MutationExecutorError::Lock(err.to_string()))
290    }
291}
292
293impl teaql_data_service::DataServiceExecutor for SqliteMutationExecutor {
294    type Error = MutationExecutorError;
295
296    fn capabilities(&self) -> teaql_data_service::DataServiceCapabilities {
297        teaql_data_service::DataServiceCapabilities {
298            query: true,
299            mutation: true,
300            transaction: true,
301            schema: true,
302            id_generation: true,
303            ..Default::default()
304        }
305    }
306}
307
308impl SqlTransport for SqliteMutationExecutor {
309    type Error = MutationExecutorError;
310
311    async fn fetch_all_sql(&self, query: &CompiledQuery) -> Result<Vec<Record>, Self::Error> {
312        SqliteMutationExecutor::fetch_all(self, query)
313    }
314
315    async fn execute_sql(&self, query: &CompiledQuery) -> Result<u64, Self::Error> {
316        SqliteMutationExecutor::execute(self, query)
317    }
318}
319
320impl teaql_data_service::StreamQueryExecutor for SqliteMutationExecutor {
321    async fn query_stream(
322        &self,
323        request: teaql_data_service::QueryRequest,
324        chunk_size: usize,
325    ) -> Result<Vec<teaql_data_service::StreamChunk>, Self::Error> {
326        let dialect = SqliteDialect;
327        // Use a dummy entity descriptor for compilation
328        let entity_desc = teaql_core::EntityDescriptor::new(&request.query.entity);
329        let compiled = dialect
330            .compile_select(&entity_desc, &request.query)
331            .map_err(MutationExecutorError::SqlCompile)?;
332        SqliteMutationExecutor::fetch_stream(self, &compiled, chunk_size)
333    }
334}
335
336impl teaql_sql::SqlTransaction for SqliteMutationExecutor {
337    type Error = MutationExecutorError;
338
339    async fn commit_sql(self) -> Result<(), Self::Error> {
340        self.commit_transaction()
341    }
342
343    async fn rollback_sql(self) -> Result<(), Self::Error> {
344        self.rollback_transaction()
345    }
346}
347
348impl teaql_sql::SqlTransactionTransport for SqliteMutationExecutor {
349    type Tx<'a>
350        = Self
351    where
352        Self: 'a;
353
354    async fn begin_sql(&self) -> Result<Self::Tx<'_>, Self::Error> {
355        self.begin_transaction()?;
356        Ok(self.clone())
357    }
358}
359
360fn initial_graph_exists_sqlite(
361    executor: &SqliteMutationExecutor,
362    dialect: &SqliteDialect,
363    entity: &EntityDescriptor,
364    graph: &GraphNode,
365) -> Result<bool, MutationExecutorError> {
366    let Some(id) = graph.values.get("id") else {
367        return Ok(false);
368    };
369    let query = dialect.compile_select(
370        entity,
371        &SelectQuery::new(&graph.entity)
372            .project("id")
373            .filter(Expr::eq("id", id.clone()))
374            .limit(1),
375    )?;
376    Ok(!executor.fetch_all(&query)?.is_empty())
377}
378
379fn compile_initial_graph_insert(
380    dialect: &impl SqlDialect,
381    entity: &EntityDescriptor,
382    graph: &GraphNode,
383) -> Result<CompiledQuery, MutationExecutorError> {
384    let mut command = InsertCommand::new(&graph.entity);
385    for (field, value) in &graph.values {
386        command = command.value(field.clone(), value.clone());
387    }
388    dialect.compile_insert(entity, &command).map_err(Into::into)
389}
390
391fn compile_initial_graph_update(
392    dialect: &impl SqlDialect,
393    entity: &EntityDescriptor,
394    graph: &GraphNode,
395) -> Result<Option<CompiledQuery>, MutationExecutorError> {
396    let Some(id) = graph.values.get("id") else {
397        return Ok(None);
398    };
399    let mut command = UpdateCommand::new(&graph.entity, id.clone());
400    for (field, value) in &graph.values {
401        if field == "id" {
402            continue;
403        }
404        command = command.value(field.clone(), value.clone());
405    }
406    match dialect.compile_update(entity, &command) {
407        Ok(query) => Ok(Some(query)),
408        Err(SqlCompileError::EmptyMutation(_)) => Ok(None),
409        Err(err) => Err(err.into()),
410    }
411}
412
413pub trait SqliteSchemaExt {
414    fn ensure_sqlite_schema(
415        &self,
416    ) -> Pin<Box<dyn Future<Output = Result<(), MutationExecutorError>> + Send + '_>>;
417}
418
419pub fn ensure_sqlite_schema_for(ctx: &UserContext) -> Result<(), MutationExecutorError> {
420    let dialect = ctx.get_resource::<SqliteDialect>().ok_or_else(|| {
421        MutationExecutorError::Bind("missing typed resource: SqliteDialect".to_owned())
422    })?;
423    let executor = ctx
424        .get_resource::<SqliteMutationExecutor>()
425        .ok_or_else(|| {
426            MutationExecutorError::Bind("missing typed resource: SqliteMutationExecutor".to_owned())
427        })?;
428
429    let entities = ctx.all_entities();
430
431    // Ensure id space table exists
432    executor.ensure_id_space_table(DEFAULT_ID_SPACE_TABLE)?;
433
434    // Process each entity table individually with granular events
435    for entity in &entities {
436        let field_count = entity.properties.len();
437        if !executor.table_exists(&entity.table_name)? {
438            // New table: create it
439            let sql = dialect.compile_create_table(entity)?;
440            executor.lock()?.execute(&sql, [])?;
441            let _ = ctx.send_event(RawAuditEvent::schema_created(
442                &entity.name,
443                &entity.table_name,
444                field_count,
445            ));
446            continue;
447        }
448        // Existing table: check for missing columns
449        let existing_columns = executor.table_columns(&entity.table_name)?;
450        let mut fields_added = 0;
451        for property in &entity.properties {
452            let bare_column = strip_identifier_quotes(&property.column_name).to_lowercase();
453            if existing_columns.contains(&bare_column) {
454                continue;
455            }
456            let sql = dialect.compile_add_column(entity, property)?;
457            executor.lock()?.execute(&sql, [])?;
458            let _ = ctx.send_event(RawAuditEvent::field_added(
459                &entity.name,
460                &entity.table_name,
461                &property.column_name,
462            ));
463            fields_added += 1;
464        }
465        let _ = ctx.send_event(RawAuditEvent::schema_verified(
466            &entity.name,
467            &entity.table_name,
468            field_count,
469        ));
470        let _ = fields_added; // used above for FieldAdded events
471    }
472
473    // Seed initial data, tracking insert vs update counts per entity
474    let mut seed_counts: BTreeMap<String, (usize, usize)> = BTreeMap::new(); // (inserted, updated)
475    for graph in ctx.initial_graphs() {
476        let entity = ctx.entity(&graph.entity).ok_or_else(|| {
477            MutationExecutorError::Bind(format!("missing entity: {}", graph.entity))
478        })?;
479        let counts = seed_counts.entry(graph.entity.clone()).or_insert((0, 0));
480        if initial_graph_exists_sqlite(executor, dialect, entity, graph)? {
481            if let Some(query) = compile_initial_graph_update(dialect, entity, graph)? {
482                executor.execute(&query)?;
483            }
484            counts.1 += 1; // updated
485            continue;
486        }
487        let query = compile_initial_graph_insert(dialect, entity, graph)?;
488        executor.execute(&query)?;
489        counts.0 += 1; // inserted
490    }
491
492    // Fire DataSeeded events per entity type
493    for (entity_name, (inserted, updated)) in &seed_counts {
494        let entity = ctx.entity(entity_name).ok_or_else(|| {
495            MutationExecutorError::Bind(format!("missing entity: {}", entity_name))
496        })?;
497        let _ = ctx.send_event(RawAuditEvent::data_seeded(
498            entity_name,
499            &entity.table_name,
500            *inserted,
501            *updated,
502        ));
503    }
504
505    Ok(())
506}
507
508impl SqliteSchemaExt for UserContext {
509    fn ensure_sqlite_schema(
510        &self,
511    ) -> Pin<Box<dyn Future<Output = Result<(), MutationExecutorError>> + Send + '_>> {
512        Box::pin(async move { ensure_sqlite_schema_for(self) })
513    }
514}
515
516#[derive(Debug, Default, Clone, Copy)]
517pub struct SqliteSchemaProvider;
518
519impl SchemaProvider for SqliteSchemaProvider {
520    fn ensure_schema<'a>(
521        &'a self,
522        ctx: &'a UserContext,
523    ) -> Pin<Box<dyn Future<Output = Result<(), RuntimeError>> + Send + 'a>> {
524        Box::pin(async move {
525            ensure_sqlite_schema_for(ctx).map_err(|err| RuntimeError::Schema(err.to_string()))
526        })
527    }
528}
529
530pub trait SqliteProviderExt {
531    fn use_sqlite_provider(&mut self, executor: SqliteMutationExecutor) -> &mut Self;
532}
533
534impl SqliteProviderExt for UserContext {
535    fn use_sqlite_provider(&mut self, executor: SqliteMutationExecutor) -> &mut Self {
536        self.insert_resource(SqliteDialect);
537        self.insert_resource(executor);
538        self.set_schema_provider(SqliteSchemaProvider);
539        self
540    }
541}
542
543#[derive(Clone)]
544pub struct SqliteIdSpaceGenerator {
545    executor: SqliteMutationExecutor,
546    table_name: String,
547}
548
549impl SqliteIdSpaceGenerator {
550    pub fn new(connection: Connection) -> Self {
551        Self::from_executor(SqliteMutationExecutor::from_connection(connection))
552    }
553
554    pub fn from_executor(executor: SqliteMutationExecutor) -> Self {
555        Self {
556            executor,
557            table_name: DEFAULT_ID_SPACE_TABLE.to_owned(),
558        }
559    }
560
561    pub fn with_table_name(mut self, table_name: impl Into<String>) -> Self {
562        self.table_name = table_name.into();
563        self
564    }
565
566    pub fn ensure_table(&self) -> Result<(), MutationExecutorError> {
567        self.executor.ensure_id_space_table(&self.table_name)
568    }
569
570    pub fn next_id(&self, entity: &str) -> Result<u64, MutationExecutorError> {
571        self.ensure_table()?;
572        let sql = format!(
573            "INSERT INTO {} (type_name, current_level) VALUES (?, 1) \
574             ON CONFLICT (type_name) DO UPDATE \
575             SET current_level = current_level + 1 \
576             RETURNING current_level",
577            quote_ident(&self.table_name)
578        );
579        let id: i64 = self
580            .executor
581            .lock()?
582            .query_row(&sql, [entity], |row| row.get(0))?;
583        u64::try_from(id).map_err(|_| {
584            MutationExecutorError::Bind(format!("generated id {id} cannot be represented as u64"))
585        })
586    }
587}
588
589impl InternalIdGenerator for SqliteIdSpaceGenerator {
590    fn generate_id(&self, entity: &str) -> Result<u64, RuntimeError> {
591        self.next_id(entity)
592            .map_err(|err| RuntimeError::IdGeneration(err.to_string()))
593    }
594}
595
596fn quote_ident(ident: &str) -> String {
597    quote_identifier_if_needed(ident, '"')
598}
599
600/// Strip wrapping identifier quotes from a SQL identifier.
601///
602/// SQLite `PRAGMA table_info` returns bare column names (e.g. `description`),
603/// but generated `PropertyDescriptor::column_name` may carry quotes
604/// (e.g. `"description"`) when the name is a reserved keyword.  This helper
605/// normalises the column name so the two can be compared correctly during
606/// schema migration.
607fn strip_identifier_quotes(ident: &str) -> &str {
608    let bytes = ident.as_bytes();
609    if bytes.len() >= 2 {
610        let (first, last) = (bytes[0], bytes[bytes.len() - 1]);
611        if (first == b'"' && last == b'"')
612            || (first == b'`' && last == b'`')
613            || (first == b'[' && last == b']')
614        {
615            return &ident[1..ident.len() - 1];
616        }
617    }
618    ident
619}
620
621fn bind_values(values: &[Value]) -> Result<Vec<SqliteValue>, MutationExecutorError> {
622    values.iter().map(bind_sqlite_value).collect()
623}
624
625fn bind_sqlite_value(value: &Value) -> Result<SqliteValue, MutationExecutorError> {
626    match value {
627        Value::Null => Ok(SqliteValue::Null),
628        Value::Bool(v) => Ok(SqliteValue::Integer(i64::from(*v))),
629        Value::I64(v) => Ok(SqliteValue::Integer(*v)),
630        Value::U64(v) => i64::try_from(*v)
631            .map(SqliteValue::Integer)
632            .map_err(|_| MutationExecutorError::Bind(format!("u64 value {v} exceeds i64 range"))),
633        Value::F64(v) => Ok(SqliteValue::Real(*v)),
634        Value::Decimal(v) => Ok(SqliteValue::Text(format!("{SQLITE_DECIMAL_PREFIX}{v}"))),
635        Value::Text(v) => Ok(SqliteValue::Text(v.clone())),
636        Value::Json(v) => Ok(SqliteValue::Text(v.to_string())),
637        Value::Date(v) => Ok(SqliteValue::Text(v.format("%Y-%m-%d").to_string())),
638        Value::Timestamp(v) => Ok(SqliteValue::Text(v.0.to_string())),
639        Value::Object(_) => Err(MutationExecutorError::UnsupportedValue("object")),
640        Value::List(_) => Err(MutationExecutorError::UnsupportedValue("list")),
641        Value::TypedNull(_) => Ok(SqliteValue::Null),
642    }
643}
644
645#[derive(Debug, Clone)]
646struct ColumnInfo {
647    name: String,
648    decl_type: Option<String>,
649}
650
651fn statement_columns(statement: &rusqlite::Statement<'_>) -> Vec<ColumnInfo> {
652    statement
653        .columns()
654        .into_iter()
655        .map(|column| ColumnInfo {
656            name: column.name().to_owned(),
657            decl_type: column.decl_type().map(|value| value.to_ascii_uppercase()),
658        })
659        .collect()
660}
661
662fn decode_sqlite_row(
663    row: &Row<'_>,
664    columns: &[ColumnInfo],
665) -> Result<Record, MutationExecutorError> {
666    let mut record = BTreeMap::new();
667    for (index, column) in columns.iter().enumerate() {
668        let value_ref = row.get_ref(index)?;
669        let value = match value_ref {
670            ValueRef::Null => Value::Null,
671            ValueRef::Integer(value) => decode_sqlite_integer(value, column),
672            ValueRef::Real(value) => Value::F64(value),
673            ValueRef::Text(value) => decode_sqlite_text(value, column)?,
674            ValueRef::Blob(_) => {
675                return Err(MutationExecutorError::UnsupportedColumnType(
676                    "BLOB".to_owned(),
677                ));
678            }
679        };
680        record.insert(column.name.clone(), value);
681    }
682    Ok(record)
683}
684
685fn decode_sqlite_integer(value: i64, column: &ColumnInfo) -> Value {
686    match column_decl_type(column).as_deref() {
687        Some("BOOLEAN") | Some("BOOL") => Value::Bool(value != 0),
688        _ => Value::I64(value),
689    }
690}
691
692fn decode_sqlite_text(value: &[u8], column: &ColumnInfo) -> Result<Value, MutationExecutorError> {
693    let value = std::str::from_utf8(value)
694        .map_err(|err| MutationExecutorError::Bind(format!("invalid sqlite text: {err}")))?;
695    if let Some(decimal) = value.strip_prefix(SQLITE_DECIMAL_PREFIX) {
696        return Decimal::from_str(decimal)
697            .map(Value::Decimal)
698            .map_err(|err| MutationExecutorError::Bind(format!("invalid sqlite decimal: {err}")));
699    }
700
701    match column_decl_type(column).as_deref() {
702        Some("NUMERIC") | Some("DECIMAL") => Decimal::from_str(value)
703            .map(Value::Decimal)
704            .map_err(|err| MutationExecutorError::Bind(format!("invalid sqlite decimal: {err}"))),
705        Some("JSON") => serde_json::from_str(value).map(Value::Json).map_err(|err| {
706            MutationExecutorError::Bind(format!("invalid sqlite json value: {err}"))
707        }),
708        Some("DATE") => NaiveDate::parse_from_str(value, "%Y-%m-%d")
709            .map(Value::Date)
710            .map_err(|err| MutationExecutorError::Bind(format!("invalid sqlite date: {err}"))),
711        Some("TIMESTAMP") | Some("DATETIME") => parse_sqlite_timestamp(value),
712        _ => infer_sqlite_text(value),
713    }
714}
715
716fn infer_sqlite_text(value: &str) -> Result<Value, MutationExecutorError> {
717    if let Ok(date) = NaiveDate::parse_from_str(value, "%Y-%m-%d") {
718        return Ok(Value::Date(date));
719    }
720    if let Ok(timestamp) = DateTime::parse_from_rfc3339(value) {
721        return Ok(Value::Timestamp(teaql_core::time::Timestamp(
722            timestamp.timestamp_millis(),
723        )));
724    }
725    if let Ok(timestamp) = NaiveDateTime::parse_from_str(value, "%Y-%m-%d %H:%M:%S") {
726        return Ok(Value::Timestamp(teaql_core::time::Timestamp(
727            timestamp.and_utc().timestamp_millis(),
728        )));
729    }
730    Ok(Value::Text(value.to_owned()))
731}
732
733fn parse_sqlite_timestamp(value: &str) -> Result<Value, MutationExecutorError> {
734    if let Ok(timestamp) = DateTime::parse_from_rfc3339(value) {
735        return Ok(Value::Timestamp(teaql_core::time::Timestamp(
736            timestamp.timestamp_millis(),
737        )));
738    }
739    if let Ok(date) = NaiveDate::parse_from_str(value, "%Y-%m-%d") {
740        return Ok(Value::Timestamp(teaql_core::time::Timestamp(
741            date.and_hms_opt(0, 0, 0)
742                .unwrap_or_default()
743                .and_utc()
744                .timestamp_millis(),
745        )));
746    }
747    NaiveDateTime::parse_from_str(value, "%Y-%m-%d %H:%M:%S")
748        .map(|timestamp| {
749            Value::Timestamp(teaql_core::time::Timestamp(
750                timestamp.and_utc().timestamp_millis(),
751            ))
752        })
753        .map_err(|err| MutationExecutorError::Bind(format!("invalid sqlite timestamp: {err}")))
754}
755
756fn column_decl_type(column: &ColumnInfo) -> Option<String> {
757    column
758        .decl_type
759        .as_ref()
760        .map(|value| value.split('(').next().unwrap_or(value).trim().to_owned())
761}
762
763#[cfg(test)]
764mod tests {
765    use super::*;
766    use teaql_core::{DeleteCommand, RecoverCommand};
767    use teaql_macros::TeaqlEntity;
768    use teaql_runtime::InMemoryMetadataStore;
769
770    fn entity() -> EntityDescriptor {
771        EntityDescriptor::new("Order")
772            .table_name("orders")
773            .property(
774                PropertyDescriptor::new("id", DataType::U64)
775                    .column_name("id")
776                    .id()
777                    .not_null(),
778            )
779            .property(
780                PropertyDescriptor::new("version", DataType::I64)
781                    .column_name("version")
782                    .version()
783                    .not_null(),
784            )
785            .property(PropertyDescriptor::new("name", DataType::Text).column_name("name"))
786    }
787
788    #[allow(dead_code)]
789    #[derive(Debug, PartialEq, TeaqlEntity)]
790    #[teaql(entity = "FeatureFlag", table = "feature_flags")]
791    struct FeatureFlagRow {
792        #[teaql(id)]
793        id: u64,
794        #[teaql(version)]
795        version: i64,
796        enabled: bool,
797        optional_enabled: Option<bool>,
798    }
799
800    fn feature_flag_record(enabled: Value, optional_enabled: Value) -> Record {
801        Record::from([
802            ("id".to_owned(), Value::U64(1)),
803            ("version".to_owned(), Value::I64(1)),
804            ("enabled".to_owned(), enabled),
805            ("optional_enabled".to_owned(), optional_enabled),
806        ])
807    }
808
809    #[test]
810    fn sqlite_dialect_compiles_mutations_and_schema() {
811        let insert = SqliteDialect
812            .compile_insert(
813                &entity(),
814                &InsertCommand::new("Order")
815                    .value("id", 1_u64)
816                    .value("name", "A"),
817            )
818            .unwrap();
819        assert_eq!(insert.sql, "INSERT INTO orders (id, name) VALUES (?, ?)");
820
821        let update = SqliteDialect
822            .compile_update(
823                &entity(),
824                &UpdateCommand::new("Order", 1_u64)
825                    .expected_version(3)
826                    .value("name", "B"),
827            )
828            .unwrap();
829        assert_eq!(
830            update.sql,
831            "UPDATE orders SET name = ?, version = ? WHERE id = ? AND version = ?"
832        );
833
834        let delete = SqliteDialect
835            .compile_delete(
836                &entity(),
837                &DeleteCommand::new("Order", 1_u64).expected_version(3),
838            )
839            .unwrap();
840        let recover = SqliteDialect
841            .compile_recover(&entity(), &RecoverCommand::new("Order", 1_u64, -4))
842            .unwrap();
843        assert_eq!(
844            delete.sql,
845            "UPDATE orders SET version = ? WHERE id = ? AND version = ?"
846        );
847        assert_eq!(
848            recover.sql,
849            "UPDATE orders SET version = ? WHERE id = ? AND version = ?"
850        );
851
852        let create = SqliteDialect.compile_create_table(&entity()).unwrap();
853        assert_eq!(
854            create,
855            "CREATE TABLE IF NOT EXISTS orders (id INTEGER PRIMARY KEY NOT NULL, version INTEGER NOT NULL, name VARCHAR(255))"
856        );
857    }
858
859    #[test]
860    fn sqlite_executor_ensures_schema_and_roundtrips_rows() {
861        let executor =
862            SqliteMutationExecutor::from_connection(Connection::open_in_memory().unwrap());
863        let entity = entity();
864        let mut ctx = UserContext::new()
865            .with_metadata(InMemoryMetadataStore::new().with_entity(entity.clone()));
866
867        ctx.use_sqlite_provider(executor.clone());
868        ensure_sqlite_schema_for(&ctx).unwrap();
869
870        let insert = SqliteDialect
871            .compile_insert(
872                &entity,
873                &InsertCommand::new("Order")
874                    .value("id", 1_u64)
875                    .value("version", 1_i64)
876                    .value("name", "draft"),
877            )
878            .unwrap();
879        assert_eq!(executor.execute(&insert).unwrap(), 1);
880
881        let select = SqliteDialect
882            .compile_select(
883                &entity,
884                &SelectQuery::new("Order")
885                    .filter(Expr::eq("id", 1_u64))
886                    .order_asc("id"),
887            )
888            .unwrap();
889        let rows = executor.fetch_all(&select).unwrap();
890        assert_eq!(rows.len(), 1);
891        assert_eq!(rows[0].get("id"), Some(&Value::I64(1)));
892        assert_eq!(rows[0].get("version"), Some(&Value::I64(1)));
893        assert_eq!(rows[0].get("name"), Some(&Value::Text("draft".to_owned())));
894    }
895
896    #[test]
897    fn sqlite_boolean_new_schema_roundtrips_as_bool() {
898        let executor =
899            SqliteMutationExecutor::from_connection(Connection::open_in_memory().unwrap());
900        let entity = <FeatureFlagRow as teaql_core::TeaqlEntity>::entity_descriptor();
901        let ddl = SqliteDialect.compile_create_table(&entity).unwrap();
902        assert!(ddl.contains("enabled BOOLEAN NOT NULL"), "{ddl}");
903        assert!(ddl.contains("optional_enabled BOOLEAN"), "{ddl}");
904        assert!(!ddl.contains("enabled INTEGER"), "{ddl}");
905
906        executor.ensure_schema(&SqliteDialect, &[&entity]).unwrap();
907        for (id, enabled, optional_enabled) in [(1_u64, false, true), (2_u64, true, false)] {
908            let insert = SqliteDialect
909                .compile_insert(
910                    &entity,
911                    &InsertCommand::new("FeatureFlag")
912                        .value("id", id)
913                        .value("version", 1_i64)
914                        .value("enabled", enabled)
915                        .value("optional_enabled", optional_enabled),
916                )
917                .unwrap();
918            assert_eq!(executor.execute(&insert).unwrap(), 1);
919        }
920
921        let select = SqliteDialect
922            .compile_select(&entity, &SelectQuery::new("FeatureFlag").order_asc("id"))
923            .unwrap();
924        let rows = executor.fetch_all(&select).unwrap();
925        assert_eq!(rows[0].get("enabled"), Some(&Value::Bool(false)));
926        assert_eq!(rows[0].get("optional_enabled"), Some(&Value::Bool(true)));
927        assert_eq!(rows[1].get("enabled"), Some(&Value::Bool(true)));
928        assert_eq!(rows[1].get("optional_enabled"), Some(&Value::Bool(false)));
929
930        let first = <FeatureFlagRow as teaql_core::Entity>::from_record(rows[0].clone()).unwrap();
931        let second = <FeatureFlagRow as teaql_core::Entity>::from_record(rows[1].clone()).unwrap();
932        assert!(!first.enabled);
933        assert_eq!(first.optional_enabled, Some(true));
934        assert!(second.enabled);
935        assert_eq!(second.optional_enabled, Some(false));
936    }
937
938    #[test]
939    fn sqlite_boolean_legacy_integer_schema_maps_only_binary_values() {
940        let executor =
941            SqliteMutationExecutor::from_connection(Connection::open_in_memory().unwrap());
942        let entity = <FeatureFlagRow as teaql_core::TeaqlEntity>::entity_descriptor();
943        executor
944            .execute(&CompiledQuery {
945                sql: "CREATE TABLE feature_flags (id INTEGER PRIMARY KEY, version INTEGER NOT NULL, enabled INTEGER NOT NULL, optional_enabled INTEGER)"
946                    .to_owned(),
947                params: Vec::new(),
948                comment: None,
949            })
950            .unwrap();
951
952        let insert = SqliteDialect
953            .compile_insert(
954                &entity,
955                &InsertCommand::new("FeatureFlag")
956                    .value("id", 1_u64)
957                    .value("version", 1_i64)
958                    .value("enabled", true)
959                    .value("optional_enabled", false),
960            )
961            .unwrap();
962        executor.execute(&insert).unwrap();
963        executor
964            .execute(&CompiledQuery {
965                sql: "INSERT INTO feature_flags (id, version, enabled, optional_enabled) VALUES (?, ?, ?, ?)"
966                    .to_owned(),
967                params: vec![
968                    Value::U64(2),
969                    Value::I64(1),
970                    Value::I64(2),
971                    Value::Null,
972                ],
973                comment: None,
974            })
975            .unwrap();
976        let select = SqliteDialect
977            .compile_select(&entity, &SelectQuery::new("FeatureFlag").order_asc("id"))
978            .unwrap();
979        let rows = executor.fetch_all(&select).unwrap();
980        assert_eq!(rows[0].get("version"), Some(&Value::I64(1)));
981        assert_eq!(rows[0].get("enabled"), Some(&Value::I64(1)));
982        assert_eq!(rows[0].get("optional_enabled"), Some(&Value::I64(0)));
983
984        let decoded = <FeatureFlagRow as teaql_core::Entity>::from_record(rows[0].clone()).unwrap();
985        assert!(decoded.enabled);
986        assert_eq!(decoded.optional_enabled, Some(false));
987        assert_eq!(rows[1].get("enabled"), Some(&Value::I64(2)));
988        let error =
989            <FeatureFlagRow as teaql_core::Entity>::from_record(rows[1].clone()).unwrap_err();
990        assert!(error.message.contains("invalid field enabled"));
991
992        for (value, expected) in [
993            (Value::I64(0), false),
994            (Value::I64(1), true),
995            (Value::U64(0), false),
996            (Value::U64(1), true),
997        ] {
998            let decoded = <FeatureFlagRow as teaql_core::Entity>::from_record(feature_flag_record(
999                value,
1000                Value::Null,
1001            ))
1002            .unwrap();
1003            assert_eq!(decoded.enabled, expected);
1004            assert_eq!(decoded.optional_enabled, None);
1005        }
1006
1007        for invalid in [Value::I64(-1), Value::I64(2), Value::U64(2)] {
1008            let error = <FeatureFlagRow as teaql_core::Entity>::from_record(feature_flag_record(
1009                invalid,
1010                Value::Null,
1011            ))
1012            .unwrap_err();
1013            assert!(error.message.contains("invalid field enabled"));
1014        }
1015        let error = <FeatureFlagRow as teaql_core::Entity>::from_record(feature_flag_record(
1016            Value::Bool(true),
1017            Value::U64(2),
1018        ))
1019        .unwrap_err();
1020        assert!(error.message.contains("invalid field optional_enabled"));
1021    }
1022
1023    #[test]
1024    fn sqlite_executor_parses_json_only_for_json_columns() {
1025        let executor =
1026            SqliteMutationExecutor::from_connection(Connection::open_in_memory().unwrap());
1027
1028        executor
1029            .execute(&CompiledQuery {
1030                sql: "CREATE TABLE payloads (text_payload TEXT, json_payload JSON)".to_owned(),
1031                params: Vec::new(),
1032                comment: None,
1033            })
1034            .unwrap();
1035        executor
1036            .execute(&CompiledQuery {
1037                sql: "INSERT INTO payloads (text_payload, json_payload) VALUES (?, ?)".to_owned(),
1038                params: vec![
1039                    Value::Text("{\"active\":true}".to_owned()),
1040                    Value::Json(serde_json::json!({"active": true})),
1041                ],
1042                comment: None,
1043            })
1044            .unwrap();
1045
1046        let rows = executor
1047            .fetch_all(&CompiledQuery {
1048                sql: "SELECT text_payload, json_payload FROM payloads".to_owned(),
1049                params: Vec::new(),
1050                comment: None,
1051            })
1052            .unwrap();
1053
1054        assert_eq!(
1055            rows[0].get("text_payload"),
1056            Some(&Value::Text("{\"active\":true}".to_owned()))
1057        );
1058        assert_eq!(
1059            rows[0].get("json_payload"),
1060            Some(&Value::Json(serde_json::json!({"active": true})))
1061        );
1062    }
1063
1064    #[test]
1065    fn sqlite_id_space_generator_increments_ids() {
1066        let executor =
1067            SqliteMutationExecutor::from_connection(Connection::open_in_memory().unwrap());
1068        let generator = SqliteIdSpaceGenerator::from_executor(executor);
1069        assert_eq!(generator.next_id("Order").unwrap(), 1);
1070        assert_eq!(generator.next_id("Order").unwrap(), 2);
1071    }
1072
1073    #[test]
1074    fn sqlite_fetch_stream_returns_chunked_rows() {
1075        let executor = SqliteMutationExecutor::new(Arc::new(Mutex::new(
1076            Connection::open_in_memory().unwrap(),
1077        )));
1078        let entity = entity();
1079
1080        // Create table and insert 25 rows
1081        executor
1082            .execute(&CompiledQuery {
1083                sql: "CREATE TABLE orders (id INTEGER PRIMARY KEY, version INTEGER, name VARCHAR(255))"
1084                    .to_owned(),
1085                params: Vec::new(),
1086                comment: None,
1087            })
1088            .unwrap();
1089
1090        for i in 1..=25 {
1091            let insert = SqliteDialect
1092                .compile_insert(
1093                    &entity,
1094                    &InsertCommand::new("Order")
1095                        .value("id", i as u64)
1096                        .value("version", 1_i64)
1097                        .value("name", format!("order-{i}")),
1098                )
1099                .unwrap();
1100            executor.execute(&insert).unwrap();
1101        }
1102
1103        // Stream with chunk_size = 10
1104        let query = SelectQuery::new("Order")
1105            .filter(Expr::gt("version", 0_i64))
1106            .order_asc("id")
1107            .stream(10);
1108
1109        let compiled = SqliteDialect.compile_select(&entity, &query).unwrap();
1110
1111        let chunks = executor.fetch_stream(&compiled, 10).unwrap();
1112
1113        // 25 rows / 10 per chunk = 3 chunks
1114        assert_eq!(chunks.len(), 3);
1115        assert_eq!(chunks[0].rows.len(), 10);
1116        assert_eq!(chunks[0].chunk_index, 0);
1117        assert!(!chunks[0].is_last);
1118
1119        assert_eq!(chunks[1].rows.len(), 10);
1120        assert_eq!(chunks[1].chunk_index, 1);
1121        assert!(!chunks[1].is_last);
1122
1123        assert_eq!(chunks[2].rows.len(), 5);
1124        assert_eq!(chunks[2].chunk_index, 2);
1125        assert!(chunks[2].is_last);
1126
1127        // Verify first and last row
1128        assert_eq!(
1129            chunks[0].rows[0].get("name"),
1130            Some(&Value::Text("order-1".to_owned()))
1131        );
1132        assert_eq!(
1133            chunks[2].rows[4].get("name"),
1134            Some(&Value::Text("order-25".to_owned()))
1135        );
1136    }
1137
1138    #[test]
1139    fn sqlite_fetch_stream_handles_empty_result() {
1140        let executor = SqliteMutationExecutor::new(Arc::new(Mutex::new(
1141            Connection::open_in_memory().unwrap(),
1142        )));
1143
1144        executor
1145            .execute(&CompiledQuery {
1146                sql: "CREATE TABLE orders (id INTEGER PRIMARY KEY, version INTEGER, name VARCHAR(255))"
1147                    .to_owned(),
1148                params: Vec::new(),
1149                comment: None,
1150            })
1151            .unwrap();
1152
1153        let entity = entity();
1154        let query = SelectQuery::new("Order")
1155            .filter(Expr::gt("version", 0_i64))
1156            .stream(10);
1157
1158        let compiled = SqliteDialect.compile_select(&entity, &query).unwrap();
1159
1160        let chunks = executor.fetch_stream(&compiled, 10).unwrap();
1161
1162        // Empty result = 1 chunk with 0 rows, marked as last
1163        assert_eq!(chunks.len(), 1);
1164        assert_eq!(chunks[0].rows.len(), 0);
1165        assert!(chunks[0].is_last);
1166    }
1167
1168    #[test]
1169    fn sqlite_fetch_stream_exact_chunk_boundary() {
1170        let executor = SqliteMutationExecutor::new(Arc::new(Mutex::new(
1171            Connection::open_in_memory().unwrap(),
1172        )));
1173        let entity = entity();
1174
1175        executor
1176            .execute(&CompiledQuery {
1177                sql: "CREATE TABLE orders (id INTEGER PRIMARY KEY, version INTEGER, name VARCHAR(255))"
1178                    .to_owned(),
1179                params: Vec::new(),
1180                comment: None,
1181            })
1182            .unwrap();
1183
1184        // Insert exactly 20 rows
1185        for i in 1..=20 {
1186            let insert = SqliteDialect
1187                .compile_insert(
1188                    &entity,
1189                    &InsertCommand::new("Order")
1190                        .value("id", i as u64)
1191                        .value("version", 1_i64)
1192                        .value("name", format!("order-{i}")),
1193                )
1194                .unwrap();
1195            executor.execute(&insert).unwrap();
1196        }
1197
1198        let query = SelectQuery::new("Order")
1199            .filter(Expr::gt("version", 0_i64))
1200            .order_asc("id")
1201            .stream(10);
1202
1203        let compiled = SqliteDialect.compile_select(&entity, &query).unwrap();
1204
1205        let chunks = executor.fetch_stream(&compiled, 10).unwrap();
1206
1207        // 20 rows / 10 per chunk = 2 full chunks + 1 empty final chunk
1208        assert_eq!(chunks.len(), 3);
1209        assert_eq!(chunks[0].rows.len(), 10);
1210        assert!(!chunks[0].is_last);
1211        assert_eq!(chunks[1].rows.len(), 10);
1212        assert!(!chunks[1].is_last);
1213        assert_eq!(chunks[2].rows.len(), 0);
1214        assert!(chunks[2].is_last);
1215    }
1216
1217    #[test]
1218    fn test_parse_sqlite_timestamp() {
1219        let ts1 = parse_sqlite_timestamp("2023-01-01 12:30:45").unwrap();
1220        assert!(matches!(ts1, Value::Timestamp(_)));
1221
1222        let ts2 = parse_sqlite_timestamp("2023-01-01").unwrap();
1223        assert!(matches!(ts2, Value::Timestamp(_)));
1224
1225        let ts3 = parse_sqlite_timestamp("2023-01-01T12:30:45Z").unwrap();
1226        assert!(matches!(ts3, Value::Timestamp(_)));
1227
1228        assert!(parse_sqlite_timestamp("invalid").is_err());
1229    }
1230}
teaql_provider_sqlite/lib.rs

teaql_provider_sqlite/
lib.rs
