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,23 +4,26 @@ 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;
 use sqlite::ResultCode;
 
 use crate::error::{PowerSyncError, Result};
 use crate::fix_data::apply_v035_fix;
-use crate::schema::inspection::ExistingView;
+use crate::schema::inspection::{ExistingTable, 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"\
@@ -170,7 +173,8 @@ VALUES(4,
         // Down migrations are less common, so we're okay about that breaking
         // in some cases.
 
-        for mut view in ExistingView::list(local_db)? {
+        let tables = ExistingTable::list(local_db)?;
+        for mut view in ExistingView::list(local_db, &tables)? {
             view.delete_trigger_sql = String::default();
             view.update_trigger_sql = String::default();
             view.insert_trigger_sql = String::default();

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,
+    },
+    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,85 @@ 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![];
 
-impl<'a> Iterator for SchemaTableColumnIterator<'a> {
-    type Item = &'a str;
+        buffer.push_str("INSERT INTO ");
+        let _ = buffer.identifier().write_str(table_name);
+        buffer.push_str(" (id");
 
-    fn next(&mut self) -> Option<Self::Item> {
-        Some(match self {
-            Self::Json(iter) => &iter.next()?.name,
-            Self::Raw(iter) => iter.next()?.as_ref(),
-        })
+        for column in self.column_names() {
+            buffer.comma();
+            let _ = buffer.identifier().write_str(column);
+        }
+        buffer.push_str(") VALUES (?1");
+        params.push(PendingStatementValue::Id);
+
+        let data_start_index = 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()));
+        }
+        buffer.push_str(") ON CONFLICT (id) DO ");
+        if self.columns().is_empty() {
+            buffer.push_str("NOTHING");
+        } else {
+            buffer.push_str("UPDATE SET ");
+            let mut do_update = buffer.comma_separated();
+
+            // 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: None,
+        }
+    }
+
+    /// 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 +166,90 @@ 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);
+    }
+
+    #[test]
+    fn no_columns_statements() {
+        let raw_table = RawTable {
+            name: "users".to_string(),
+            schema: RawTableSchema::default(),
+            put: None,
+            delete: None,
+            clear: None,
+        };
+        let structure = InferredTableStructure { columns: vec![] };
+        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) VALUES (?1) ON CONFLICT (id) DO NOTHING"#
+        );
+        assert_eq!(put.params.len(), 1);
+        assert_matches!(put.params[0], PendingStatementValue::Id);
+
+        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);
+    }
+}