Extremely basic support for raw tables

Author: simolus3

crates/core/src/sync/storage_adapter.rs

@@ -184,7 +184,11 @@ impl StorageAdapter {
         }
 
         let sync_result = match priority {
-            None => SyncOperation::new(self.db, None).apply(),
+            None => {
+                let mut sync = SyncOperation::new(self.db, None);
+                sync.use_schema(schema);
+                sync.apply()
+            }
             Some(priority) => {
                 let args = PartialArgs {
                     priority,

crates/core/src/sync_local.rs

@@ -136,7 +136,8 @@ impl<'a> SyncOperation<'a> {
                     match data {
                         Ok(data) => {
                             let stmt = raw.put_statement(self.db)?;
-                            stmt.bind_for_put(id, data)?;
+                            let parsed: serde_json::Value = serde_json::from_str(data)?;
+                            stmt.bind_for_put(id, &parsed)?;
                             stmt.stmt.exec()?;
                         }
                         Err(_) => {
@@ -493,11 +494,8 @@ impl<'a> PreparedPendingStatement<'a> {
         })
     }
 
-    pub fn bind_for_put(&self, id: &str, json_data: &str) -> Result<(), SQLiteError> {
+    pub fn bind_for_put(&self, id: &str, json_data: &serde_json::Value) -> Result<(), SQLiteError> {
         use serde_json::Value;
-
-        let parsed: Value = serde_json::from_str(json_data)?;
-
         for (i, source) in self.params.iter().enumerate() {
             let i = (i + 1) as i32;
 
@@ -506,7 +504,7 @@ impl<'a> PreparedPendingStatement<'a> {
                     self.stmt.bind_text(i, id, Destructor::STATIC)?;
                 }
                 PendingStatementValue::Column(column) => {
-                    let parsed = parsed.as_object().ok_or_else(|| {
+                    let parsed = json_data.as_object().ok_or_else(|| {
                         SQLiteError(
                             ResultCode::CONSTRAINT_DATATYPE,
                             Some("expected oplog data to be an object".to_string()),

dart/test/sync_test.dart

@@ -92,7 +92,7 @@ void _syncTests<T>({
 
   List<Object?> pushSyncData(
       String bucket, String opId, String rowId, Object op, Object? data,
-      {int checksum = 0}) {
+      {int checksum = 0, String objectType = 'items'}) {
     return syncLine({
       'data': {
         'bucket': bucket,
@@ -103,7 +103,7 @@ void _syncTests<T>({
           {
             'op_id': opId,
             'op': op,
-            'object_type': 'items',
+            'object_type': objectType,
             'object_id': rowId,
             'checksum': checksum,
             'data': json.encode(data),
@@ -676,6 +676,72 @@ void _syncTests<T>({
       expect(db.select('SELECT * FROM ps_buckets'), isEmpty);
     });
   });
+
+  group('raw tables', () {
+    syncTest('smoke test', (_) {
+      db.execute(
+          'CREATE TABLE users (id TEXT NOT NULL PRIMARY KEY, name TEXT NOT NULL) STRICT;');
+
+      invokeControl(
+        'start',
+        json.encode({
+          'schema': {
+            'tables': [
+              {
+                'name': 'users',
+                'raw': {
+                  'put': {
+                    'sql':
+                        'INSERT OR REPLACE INTO users (id, name) VALUES (?, ?);',
+                    'params': [
+                      'Id',
+                      {'Column': 'name'}
+                    ],
+                  },
+                  'delete': {
+                    'sql': 'DELETE FROM users WHERE id = ?',
+                    'params': ['Id'],
+                  },
+                },
+                'columns': [],
+              }
+            ],
+          },
+        }),
+      );
+
+      // Insert
+      pushCheckpoint(buckets: [bucketDescription('a')]);
+      pushSyncData(
+        'a',
+        '1',
+        'my_user',
+        'PUT',
+        {'name': 'First user'},
+        objectType: 'users',
+      );
+      pushCheckpointComplete();
+
+      final users = db.select('SELECT * FROM users;');
+      expect(users, [
+        {'id': 'my_user', 'name': 'First user'}
+      ]);
+
+      // Delete
+      pushCheckpoint(buckets: [bucketDescription('a')]);
+      pushSyncData(
+        'a',
+        '1',
+        'my_user',
+        'REMOVE',
+        null,
+        objectType: 'users',
+      );
+      pushCheckpointComplete();
+
+      expect(db.select('SELECT * FROM users'), isEmpty);
+    });
+  });
 }
 
 const _schema = {