Managed raw tables

crates/core/src/json_util.rs

@@ -6,8 +6,8 @@ use core::ffi::c_int;
 use crate::constants::SUBTYPE_JSON;
 use crate::create_sqlite_text_fn;
 use crate::error::{PowerSyncError, Result};
-use powersync_sqlite_nostd as sqlite;
 use powersync_sqlite_nostd::bindings::{SQLITE_RESULT_SUBTYPE, SQLITE_SUBTYPE};
+use powersync_sqlite_nostd::{self as sqlite, ColumnType};
 use powersync_sqlite_nostd::{Connection, Context, Value};
 use sqlite::ResultCode;
 
@@ -38,6 +38,10 @@ fn powersync_json_merge_impl(
     }
     let mut result = String::from("{");
     for arg in args {
+        if arg.value_type() == ColumnType::Null {
+            continue;
+        }
+
         let chunk = arg.text();
         if chunk.is_empty() || !chunk.starts_with('{') || !chunk.ends_with('}') {
             return Err(PowerSyncError::argument_error("Expected json object"));

crates/core/src/migrations.rs

@@ -4,7 +4,6 @@ use alloc::format;
 use alloc::string::{String, ToString};
 use alloc::vec::Vec;
 
-use powersync_sqlite_nostd::Context;
 use powersync_sqlite_nostd::{self as sqlite, Destructor};
 use serde::Serialize;
 use serde_json::json;
@@ -15,12 +14,16 @@ use crate::fix_data::apply_v035_fix;
 use crate::schema::inspection::ExistingView;
 use crate::sync::BucketPriority;
 use crate::utils::database::Database;
+use crate::utils::verify_in_transaction;
 
 pub const LATEST_VERSION: i32 = 14;
 
-pub fn powersync_migrate(ctx: *mut sqlite::context, target_version: i32) -> Result<()> {
-    let local_db = Database::from(ctx.db_handle());
+pub fn initialize_database(db: Database) -> Result<()> {
+    verify_in_transaction(db)?;
+    powersync_migrate(db, LATEST_VERSION)
+}
 
+pub fn powersync_migrate(local_db: Database, target_version: i32) -> Result<()> {
     // language=SQLite
     local_db.exec_safe(
         c"\

crates/core/src/schema/common.rs

@@ -1,10 +1,18 @@
-use core::slice;
+use core::fmt::Write;
 
-use alloc::{string::String, vec::Vec};
+use alloc::{
+    string::{String, ToString},
+    vec,
+    vec::Vec,
+};
 use serde::Deserialize;
 
-use crate::schema::{
-    Column, CommonTableOptions, RawTable, Table, raw_table::InferredTableStructure,
+use crate::{
+    schema::{
+        Column, CommonTableOptions, PendingStatement, PendingStatementValue, RawTable, Table,
+        raw_table::InferredTableStructure, table_info::RestColumnIndex,
+    },
+    utils::SqlBuffer,
 };
 
 /// Utility to wrap both PowerSync-managed JSON tables and raw tables (with their schema snapshot
@@ -18,6 +26,17 @@ pub enum SchemaTable<'a> {
 }
 
 impl<'a> SchemaTable<'a> {
+    /// The type name used for the table when referenced in `ps_crud`, `ps_oplog` and other tables.
+    pub fn name(&self) -> &str {
+        match self {
+            SchemaTable::Json(table) => &table.name,
+            SchemaTable::Raw {
+                definition,
+                schema: _,
+            } => &definition.name,
+        }
+    }
+
     pub fn common_options(&self) -> &CommonTableOptions {
         match self {
             Self::Json(table) => &table.options,
@@ -28,37 +47,101 @@ impl<'a> SchemaTable<'a> {
         }
     }
 
-    /// Iterates over defined column names in this table (not including the `id` column).
-    pub fn column_names(&self) -> impl Iterator<Item = &'a str> {
+    pub fn columns(&self) -> &'a [Column] {
         match self {
-            Self::Json(table) => SchemaTableColumnIterator::Json(table.columns.iter()),
+            Self::Json(table) => &table.columns,
             Self::Raw {
                 definition: _,
                 schema,
-            } => SchemaTableColumnIterator::Raw(schema.columns.iter()),
+            } => &schema.columns,
         }
     }
-}
 
-impl<'a> From<&'a Table> for SchemaTable<'a> {
-    fn from(value: &'a Table) -> Self {
-        Self::Json(value)
+    /// Iterates over defined column names in this table (not including the `id` column).
+    pub fn column_names(&self) -> impl Iterator<Item = &'a str> {
+        self.columns().iter().map(|c| &*c.name)
     }
-}
 
-enum SchemaTableColumnIterator<'a> {
-    Json(slice::Iter<'a, Column>),
-    Raw(slice::Iter<'a, String>),
-}
+    /// Generates a statement of the form `INSERT INTO $tbl ($cols) VALUES (?, ...) ON CONFLICT (id)
+    /// DO UPDATE SET ...` for the sync client.
+    pub fn infer_put_stmt(&self, table_name: &str) -> PendingStatement {
+        let mut buffer = SqlBuffer::new();
+        let mut params = vec![];
+        let mut rest = match self {
+            SchemaTable::Json(_) => Some(("_rest", RestColumnIndex::default())),
+            SchemaTable::Raw { .. } => None,
+        };
+
+        buffer.push_str("INSERT INTO ");
+        let _ = buffer.identifier().write_str(table_name);
+        buffer.push_str(" (id");
+        if let Some((column, _)) = rest {
+            let _ = write!(&mut buffer, ", {column}");
+        }
+
+        for column in self.column_names() {
+            buffer.comma();
+            let _ = buffer.identifier().write_str(column);
+        }
+        buffer.push_str(") VALUES (?1");
+        params.push(PendingStatementValue::Id);
+        if let Some((_, ref mut rest_index)) = rest {
+            params.push(PendingStatementValue::Rest);
+            buffer.push_str(", ?2");
+            rest_index.rest_parameter_positions.push(1); // this is zero-indexed
+        }
 
-impl<'a> Iterator for SchemaTableColumnIterator<'a> {
-    type Item = &'a str;
+        let data_start_index = if rest.is_some() { 3 } else { 2 };
+        for (i, column) in self.column_names().enumerate() {
+            buffer.comma();
+            let _ = write!(&mut buffer, "?{}", i + data_start_index);
+            params.push(PendingStatementValue::Column(column.to_string()));
 
-    fn next(&mut self) -> Option<Self::Item> {
-        Some(match self {
-            Self::Json(iter) => &iter.next()?.name,
-            Self::Raw(iter) => iter.next()?.as_ref(),
-        })
+            if let Some((_, ref mut index)) = rest {
+                index.named_parameters.insert(column.to_string());
+            }
+        }
+        buffer.push_str(") ON CONFLICT (id) DO UPDATE SET ");
+        let mut do_update = buffer.comma_separated();
+
+        if let Some((column, _)) = rest {
+            let entry = do_update.element();
+            let _ = write!(entry, "{column} = ?2");
+        }
+
+        // Generate an "x" = ? for all synced columns to update them without affecting local-only
+        // columns.
+        for (i, column) in self.column_names().enumerate() {
+            let entry = do_update.element();
+            let _ = entry.identifier().write_str(column);
+            let _ = write!(entry, " = ?{}", i + data_start_index);
+        }
+
+        PendingStatement {
+            sql: buffer.sql,
+            params,
+            named_parameters_index: rest.map(|e| e.1),
+        }
+    }
+
+    /// Generates a statement of the form `DELETE FROM $tbl WHERE id = ?` for the sync client.
+    pub fn infer_delete_stmt(&self, table_name: &str) -> PendingStatement {
+        let mut buffer = SqlBuffer::new();
+        buffer.push_str("DELETE FROM ");
+        let _ = buffer.identifier().write_str(table_name);
+        buffer.push_str(" WHERE id = ?");
+
+        PendingStatement {
+            sql: buffer.sql,
+            params: vec![PendingStatementValue::Id],
+            named_parameters_index: None,
+        }
+    }
+}
+
+impl<'a> From<&'a Table> for SchemaTable<'a> {
+    fn from(value: &'a Table) -> Self {
+        Self::Json(value)
     }
 }
 
@@ -99,3 +182,61 @@ impl<'de> Deserialize<'de> for ColumnFilter {
         Ok(Self::from(Vec::<String>::deserialize(deserializer)?))
     }
 }
+#[cfg(test)]
+mod test {
+    use alloc::{string::ToString, vec};
+    use core::assert_matches;
+
+    use crate::schema::{
+        Column, PendingStatementValue, RawTable, SchemaTable, raw_table::InferredTableStructure,
+        table_info::RawTableSchema,
+    };
+
+    #[test]
+    fn infer_sync_statements() {
+        let raw_table = RawTable {
+            name: "users".to_string(),
+            schema: RawTableSchema::default(),
+            put: None,
+            delete: None,
+            clear: None,
+        };
+        let structure = InferredTableStructure {
+            columns: vec![
+                Column {
+                    name: "foo".to_string(),
+                    type_name: "TEXT".to_string(),
+                },
+                Column {
+                    name: "bar".to_string(),
+                    type_name: "TEXT".to_string(),
+                },
+            ],
+        };
+        let schema_table = SchemaTable::Raw {
+            definition: &raw_table,
+            schema: &structure,
+        };
+
+        let put = schema_table.infer_put_stmt("tbl");
+        assert_eq!(
+            put.sql,
+            r#"INSERT INTO "tbl" (id, "foo", "bar") VALUES (?1, ?2, ?3) ON CONFLICT (id) DO UPDATE SET "foo" = ?2, "bar" = ?3"#
+        );
+        assert_eq!(put.params.len(), 3);
+        assert_matches!(put.params[0], PendingStatementValue::Id);
+        assert_matches!(
+            put.params[1],
+            PendingStatementValue::Column(ref name) if name == "foo"
+        );
+        assert_matches!(
+            put.params[2],
+            PendingStatementValue::Column(ref name) if name == "bar"
+        );
+
+        let delete = schema_table.infer_delete_stmt("tbl");
+        assert_eq!(delete.sql, r#"DELETE FROM "tbl" WHERE id = ?"#);
+        assert_eq!(delete.params.len(), 1);
+        assert_matches!(delete.params[0], PendingStatementValue::Id);
+    }
+}