Generate triggers for raw tables

Author: simolus3

crates/core/src/schema/common.rs

@@ -0,0 +1,81 @@
+use core::slice;
+
+use alloc::{string::String, vec::Vec};
+
+use crate::schema::{
+    Column, CommonTableOptions, RawTable, Table, raw_table::InferredTableStructure,
+};
+
+pub enum SchemaTable<'a> {
+    Json(&'a Table),
+    Raw {
+        definition: &'a RawTable,
+        schema: &'a InferredTableStructure,
+    },
+}
+
+impl<'a> SchemaTable<'a> {
+    pub fn common_options(&self) -> &CommonTableOptions {
+        match self {
+            Self::Json(table) => &table.options,
+            Self::Raw {
+                definition,
+                schema: _,
+            } => &definition.schema.options,
+        }
+    }
+
+    pub fn column_names(&self) -> impl Iterator<Item = &'a str> {
+        match self {
+            Self::Json(table) => SchemaTableColumnIterator::Json(table.columns.iter()),
+            Self::Raw {
+                definition: _,
+                schema,
+            } => SchemaTableColumnIterator::Raw(schema.columns.iter()),
+        }
+    }
+}
+
+impl<'a> From<&'a Table> for SchemaTable<'a> {
+    fn from(value: &'a Table) -> Self {
+        Self::Json(value)
+    }
+}
+
+enum SchemaTableColumnIterator<'a> {
+    Json(slice::Iter<'a, Column>),
+    Raw(slice::Iter<'a, String>),
+}
+
+impl<'a> Iterator for SchemaTableColumnIterator<'a> {
+    type Item = &'a str;
+
+    fn next(&mut self) -> Option<Self::Item> {
+        Some(match self {
+            Self::Json(iter) => &iter.next()?.name,
+            Self::Raw(iter) => iter.next()?.as_ref(),
+        })
+    }
+}
+
+pub struct ColumnFilter {
+    sorted_names: Vec<String>,
+}
+
+impl From<Vec<String>> for ColumnFilter {
+    fn from(mut value: Vec<String>) -> Self {
+        value.sort();
+        Self {
+            sorted_names: value,
+        }
+    }
+}
+
+impl ColumnFilter {
+    /// Whether this filter matches the given column name.
+    pub fn matches(&self, column: &str) -> bool {
+        self.sorted_names
+            .binary_search_by(|item| item.as_str().cmp(column))
+            .is_ok()
+    }
+}

crates/core/src/schema/mod.rs

@@ -1,18 +1,23 @@
+mod common;
 pub mod inspection;
 mod management;
 mod raw_table;
 mod table_info;
 
 use alloc::{rc::Rc, vec::Vec};
-use powersync_sqlite_nostd as sqlite;
+pub use common::{ColumnFilter, SchemaTable};
+use powersync_sqlite_nostd::{self as sqlite, Connection, Context, Value, args};
 use serde::Deserialize;
 use sqlite::ResultCode;
 pub use table_info::{
-    Column, DiffIncludeOld, PendingStatement, PendingStatementValue, RawTable, Table,
+    Column, CommonTableOptions, PendingStatement, PendingStatementValue, RawTable, Table,
     TableInfoFlags,
 };
 
-use crate::state::DatabaseState;
+use crate::{
+    error::PowerSyncError, schema::raw_table::generate_raw_table_trigger, state::DatabaseState,
+    utils::WriteType,
+};
 
 #[derive(Deserialize, Default)]
 pub struct Schema {
@@ -22,5 +27,47 @@ pub struct Schema {
 }
 
 pub fn register(db: *mut sqlite::sqlite3, state: Rc<DatabaseState>) -> Result<(), ResultCode> {
-    management::register(db, state)
+    management::register(db, state)?;
+
+    {
+        fn create_trigger(
+            context: *mut sqlite::context,
+            args: &[*mut sqlite::value],
+        ) -> Result<(), PowerSyncError> {
+            // Args: Table (JSON), trigger_name, write_type
+            let table: RawTable =
+                serde_json::from_str(args[0].text()).map_err(PowerSyncError::as_argument_error)?;
+            let trigger_name = args[1].text();
+            let write_type: WriteType = args[2].text().parse()?;
+
+            let db = context.db_handle();
+            let create_trigger_stmt =
+                generate_raw_table_trigger(db, &table, trigger_name, write_type)?;
+            db.exec_safe(&create_trigger_stmt)?;
+            Ok(())
+        }
+
+        extern "C" fn create_raw_trigger_sqlite(
+            context: *mut sqlite::context,
+            argc: i32,
+            args: *mut *mut sqlite::value,
+        ) {
+            let args = args!(argc, args);
+            if let Err(e) = create_trigger(context, args) {
+                e.apply_to_ctx("powersync_create_raw_table_crud_trigger", context);
+            }
+        }
+
+        db.create_function_v2(
+            "powersync_create_raw_table_crud_trigger",
+            3,
+            sqlite::UTF8,
+            None,
+            Some(create_raw_trigger_sqlite),
+            None,
+            None,
+            Some(DatabaseState::destroy_rc),
+        )?;
+    }
+    Ok(())
 }

crates/core/src/schema/raw_table.rs

@@ -1,15 +1,21 @@
+use core::fmt::from_fn;
+
 use alloc::{
+    format,
     string::{String, ToString},
     vec,
     vec::Vec,
 };
 use powersync_sqlite_nostd::{Connection, Destructor, ResultCode};
 
-use crate::error::PowerSyncError;
+use crate::{
+    error::PowerSyncError,
+    schema::{RawTable, SchemaTable},
+    utils::{InsertIntoCrud, SqlBuffer, WriteType},
+    views::table_columns_to_json_object,
+};
 
 pub struct InferredTableStructure {
-    pub table_name: String,
-    pub has_id_column: bool,
     pub columns: Vec<String>,
 }
 
@@ -35,12 +41,112 @@ impl InferredTableStructure {
 
         if !has_id_column && columns.is_empty() {
             Ok(None)
+        } else if !has_id_column {
+            Err(PowerSyncError::argument_error(format!(
+                "Table {table_name} has no id column."
+            )))
+        } else {
+            Ok(Some(Self { columns }))
+        }
+    }
+}
+
+/// Generates a `CREATE TRIGGER` statement to capture writes on raw tables and to forward them to
+/// ps-crud.
+pub fn generate_raw_table_trigger(
+    db: impl Connection,
+    table: &RawTable,
+    trigger_name: &str,
+    write: WriteType,
+) -> Result<String, PowerSyncError> {
+    let Some(local_table_name) = table.schema.table_name.as_ref() else {
+        return Err(PowerSyncError::argument_error("Table has no local name"));
+    };
+
+    let Some(resolved_table) = InferredTableStructure::read_from_database(local_table_name, db)?
+    else {
+        return Err(PowerSyncError::argument_error(format!(
+            "Could not find {} in local schema",
+            local_table_name
+        )));
+    };
+
+    let as_schema_table = SchemaTable::Raw {
+        definition: table,
+        schema: &resolved_table,
+    };
+
+    let mut buffer = SqlBuffer::new();
+    buffer.create_trigger("", trigger_name);
+    buffer.trigger_after(write, local_table_name);
+    // Skip the trigger for writes during sync_local, these aren't crud writes.
+    buffer.push_str("WHEN NOT powersync_in_sync_operation() BEGIN\n");
+
+    if table.schema.options.flags.insert_only() {
+        if write != WriteType::Insert {
+            // Prevent illegal writes to a table marked as insert-only by raising errors here.
+            buffer.push_str("SELECT RAISE(FAIL, 'Unexpected update on insert-only table');\n");
         } else {
-            Ok(Some(Self {
-                table_name: table_name.to_string(),
-                has_id_column,
-                columns,
-            }))
+            // Write directly to powersync_crud_ to skip writing the $local bucket for insert-only
+            // tables.
+            let fragment = table_columns_to_json_object("NEW", &as_schema_table)?;
+            buffer.powersync_crud_manual_put(&table.name, &fragment);
+        }
+    } else {
+        if write == WriteType::Update {
+            // Updates must not change the id.
+            buffer.check_id_not_changed();
         }
+
+        let json_fragment_new = table_columns_to_json_object("NEW", &as_schema_table)?;
+        let json_fragment_old = if write == WriteType::Update {
+            Some(table_columns_to_json_object("OLD", &as_schema_table)?)
+        } else {
+            None
+        };
+
+        buffer.insert_into_powersync_crud(InsertIntoCrud {
+            op: write,
+            table: &as_schema_table,
+            id_expr: from_fn(|f| {
+                if write == WriteType::Delete {
+                    f.write_str("OLD.")
+                } else {
+                    f.write_str("NEW.")
+                }?;
+                f.write_str(".id")
+            }),
+            type_name: &table.name,
+            data: Some(from_fn(|f| {
+                match write {
+                    WriteType::Insert => {}
+                    WriteType::Update => todo!(),
+                    WriteType::Delete => {
+                        // There is no data for deleted rows, don't emit anything.
+                    }
+                }
+
+                if write == WriteType::Delete {
+                    // There is no data for deleted rows.
+                    return Ok(());
+                }
+
+                write!(f, "json(powersync_diff(")?;
+
+                if let Some(ref old) = json_fragment_old {
+                    f.write_str(old)?;
+                } else {
+                    // We don't have OLD values for inserts, we diff from an empty JSON object
+                    // instead.
+                    f.write_str("'{}'")?;
+                };
+
+                write!(f, ", {json_fragment_new}))")
+            })),
+            metadata: None::<&'static str>,
+        })?;
     }
+
+    buffer.trigger_end();
+    Ok(buffer.sql)
 }