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

teaql_provider_sqlite/
lib.rs
