Persist sync lines

Author: simolus3

crates/core/src/bson/de.rs

@@ -154,7 +154,7 @@ impl<'de, 'a> de::Deserializer<'de> for &'a mut Deserializer<'de> {
     fn deserialize_enum<V>(
         self,
         name: &'static str,
-        variants: &'static [&'static str],
+        _variants: &'static [&'static str],
         visitor: V,
     ) -> Result<V::Value, Self::Error>
     where
@@ -283,7 +283,7 @@ impl<'a, 'de> VariantAccess<'de> for &'a mut Deserializer<'de> {
         seed.deserialize(self)
     }
 
-    fn tuple_variant<V>(self, len: usize, visitor: V) -> Result<V::Value, Self::Error>
+    fn tuple_variant<V>(self, _len: usize, visitor: V) -> Result<V::Value, Self::Error>
     where
         V: Visitor<'de>,
     {
@@ -292,8 +292,8 @@ impl<'a, 'de> VariantAccess<'de> for &'a mut Deserializer<'de> {
     }
 
     fn struct_variant<V>(
-        mut self,
-        fields: &'static [&'static str],
+        self,
+        _fields: &'static [&'static str],
         visitor: V,
     ) -> Result<V::Value, Self::Error>
     where

crates/core/src/bson/error.rs

@@ -6,8 +6,6 @@ use alloc::{
 };
 use serde::de::{self, StdError};
 
-use crate::error::SQLiteError;
-
 use super::parser::ElementType;
 
 #[derive(Debug)]

crates/core/src/bson/writer.rs

@@ -25,7 +25,7 @@ impl BsonWriter {
     pub fn put_str(&mut self, name: &str, value: &str) {
         self.put_entry(ElementType::String, name);
 
-        let bytes = name.as_bytes();
+        let bytes = value.as_bytes();
         self.output.put_i32_le(bytes.len() as i32);
         self.output.put_slice(bytes);
         self.output.push(0);

crates/core/src/diff.rs

@@ -1,6 +1,5 @@
 extern crate alloc;
 
-use alloc::format;
 use alloc::string::{String, ToString};
 use core::ffi::c_int;
 

crates/core/src/json_merge.rs

@@ -1,6 +1,5 @@
 extern crate alloc;
 
-use alloc::format;
 use alloc::string::{String, ToString};
 use core::ffi::c_int;
 

crates/core/src/operations.rs

@@ -1,255 +1,19 @@
 use alloc::format;
-use alloc::string::String;
 
-use crate::error::{PSResult, SQLiteError};
+use crate::error::SQLiteError;
+use crate::sync::line::DataLine;
+use crate::sync::operations::insert_bucket_operations;
+use crate::sync::storage_adapter::StorageAdapter;
 use sqlite_nostd as sqlite;
 use sqlite_nostd::{Connection, ResultCode};
 
 use crate::ext::SafeManagedStmt;
 
 // Run inside a transaction
 pub fn insert_operation(db: *mut sqlite::sqlite3, data: &str) -> Result<(), SQLiteError> {
-    // language=SQLite
-    let statement = db.prepare_v2(
-        "\
-SELECT
-    json_extract(e.value, '$.bucket') as bucket,
-    json_extract(e.value, '$.data') as data,
-    json_extract(e.value, '$.has_more') as has_more,
-    json_extract(e.value, '$.after') as after,
-    json_extract(e.value, '$.next_after') as next_after
-FROM json_each(json_extract(?1, '$.buckets')) e",
-    )?;
-    statement.bind_text(1, data, sqlite::Destructor::STATIC)?;
-
-    while statement.step()? == ResultCode::ROW {
-        let bucket = statement.column_text(0)?;
-        let data = statement.column_text(1)?;
-        // let _has_more = statement.column_int(2)? != 0;
-        // let _after = statement.column_text(3)?;
-        // let _next_after = statement.column_text(4)?;
-
-        insert_bucket_operations(db, bucket, data)?;
-    }
-
-    Ok(())
-}
-
-pub fn insert_bucket_operations(
-    db: *mut sqlite::sqlite3,
-    bucket: &str,
-    data: &str,
-) -> Result<(), SQLiteError> {
-    // Statement to insert new operations (only for PUT and REMOVE).
-    // language=SQLite
-    let iterate_statement = db.prepare_v2(
-        "\
-SELECT
-    json_extract(e.value, '$.op_id') as op_id,
-    json_extract(e.value, '$.op') as op,
-    json_extract(e.value, '$.object_type') as object_type,
-    json_extract(e.value, '$.object_id') as object_id,
-    json_extract(e.value, '$.checksum') as checksum,
-    json_extract(e.value, '$.data') as data,
-    json_extract(e.value, '$.subkey') as subkey
-FROM json_each(?) e",
-    )?;
-    iterate_statement.bind_text(1, data, sqlite::Destructor::STATIC)?;
-
-    // We do an ON CONFLICT UPDATE simply so that the RETURNING bit works for existing rows.
-    // We can consider splitting this into separate SELECT and INSERT statements.
-    // language=SQLite
-    let bucket_statement = db.prepare_v2(
-        "INSERT INTO ps_buckets(name)
-                            VALUES(?)
-                        ON CONFLICT DO UPDATE
-                            SET last_applied_op = last_applied_op
-                        RETURNING id, last_applied_op",
-    )?;
-    bucket_statement.bind_text(1, bucket, sqlite::Destructor::STATIC)?;
-    bucket_statement.step()?;
-
-    let bucket_id = bucket_statement.column_int64(0);
-
-    // This is an optimization for initial sync - we can avoid persisting individual REMOVE
-    // operations when last_applied_op = 0.
-    // We do still need to do the "supersede_statement" step for this case, since a REMOVE
-    // operation can supersede another PUT operation we're syncing at the same time.
-    let mut is_empty = bucket_statement.column_int64(1) == 0;
-
-    // Statement to supersede (replace) operations with the same key.
-    // language=SQLite
-    let supersede_statement = db.prepare_v2(
-        "\
-DELETE FROM ps_oplog
-    WHERE unlikely(ps_oplog.bucket = ?1)
-    AND ps_oplog.key = ?2
-RETURNING op_id, hash",
-    )?;
-    supersede_statement.bind_int64(1, bucket_id)?;
-
-    // language=SQLite
-    let insert_statement = db.prepare_v2("\
-INSERT INTO ps_oplog(bucket, op_id, key, row_type, row_id, data, hash) VALUES (?, ?, ?, ?, ?, ?, ?)")?;
-    insert_statement.bind_int64(1, bucket_id)?;
-
-    let updated_row_statement = db.prepare_v2(
-        "\
-INSERT OR IGNORE INTO ps_updated_rows(row_type, row_id) VALUES(?1, ?2)",
-    )?;
-
-    bucket_statement.reset()?;
-
-    let mut last_op: Option<i64> = None;
-    let mut add_checksum: i32 = 0;
-    let mut op_checksum: i32 = 0;
-    let mut added_ops: i32 = 0;
-
-    while iterate_statement.step()? == ResultCode::ROW {
-        let op_id = iterate_statement.column_int64(0);
-        let op = iterate_statement.column_text(1)?;
-        let object_type = iterate_statement.column_text(2);
-        let object_id = iterate_statement.column_text(3);
-        let checksum = iterate_statement.column_int(4);
-        let op_data = iterate_statement.column_text(5);
-
-        last_op = Some(op_id);
-        added_ops += 1;
-
-        if op == "PUT" || op == "REMOVE" {
-            let key: String;
-            if let (Ok(object_type), Ok(object_id)) = (object_type.as_ref(), object_id.as_ref()) {
-                let subkey = iterate_statement.column_text(6).unwrap_or("null");
-                key = format!("{}/{}/{}", &object_type, &object_id, subkey);
-            } else {
-                key = String::from("");
-            }
-
-            supersede_statement.bind_text(2, &key, sqlite::Destructor::STATIC)?;
-
-            let mut superseded = false;
-
-            while supersede_statement.step()? == ResultCode::ROW {
-                // Superseded (deleted) a previous operation, add the checksum
-                let supersede_checksum = supersede_statement.column_int(1);
-                add_checksum = add_checksum.wrapping_add(supersede_checksum);
-                op_checksum = op_checksum.wrapping_sub(supersede_checksum);
-
-                // Superseded an operation, only skip if the bucket was empty
-                // Previously this checked "superseded_op <= last_applied_op".
-                // However, that would not account for a case where a previous
-                // PUT operation superseded the original PUT operation in this
-                // same batch, in which case superseded_op is not accurate for this.
-                if !is_empty {
-                    superseded = true;
-                }
-            }
-            supersede_statement.reset()?;
-
-            if op == "REMOVE" {
-                let should_skip_remove = !superseded;
-
-                add_checksum = add_checksum.wrapping_add(checksum);
-
-                if !should_skip_remove {
-                    if let (Ok(object_type), Ok(object_id)) = (object_type, object_id) {
-                        updated_row_statement.bind_text(
-                            1,
-                            object_type,
-                            sqlite::Destructor::STATIC,
-                        )?;
-                        updated_row_statement.bind_text(
-                            2,
-                            object_id,
-                            sqlite::Destructor::STATIC,
-                        )?;
-                        updated_row_statement.exec()?;
-                    }
-                }
-
-                continue;
-            }
-
-            insert_statement.bind_int64(2, op_id)?;
-            if key != "" {
-                insert_statement.bind_text(3, &key, sqlite::Destructor::STATIC)?;
-            } else {
-                insert_statement.bind_null(3)?;
-            }
-
-            if let (Ok(object_type), Ok(object_id)) = (object_type, object_id) {
-                insert_statement.bind_text(4, object_type, sqlite::Destructor::STATIC)?;
-                insert_statement.bind_text(5, object_id, sqlite::Destructor::STATIC)?;
-            } else {
-                insert_statement.bind_null(4)?;
-                insert_statement.bind_null(5)?;
-            }
-            if let Ok(data) = op_data {
-                insert_statement.bind_text(6, data, sqlite::Destructor::STATIC)?;
-            } else {
-                insert_statement.bind_null(6)?;
-            }
-
-            insert_statement.bind_int(7, checksum)?;
-            insert_statement.exec()?;
-
-            op_checksum = op_checksum.wrapping_add(checksum);
-        } else if op == "MOVE" {
-            add_checksum = add_checksum.wrapping_add(checksum);
-        } else if op == "CLEAR" {
-            // Any remaining PUT operations should get an implicit REMOVE
-            // language=SQLite
-            let clear_statement1 = db
-                .prepare_v2(
-                    "INSERT OR IGNORE INTO ps_updated_rows(row_type, row_id)
-SELECT row_type, row_id
-FROM ps_oplog
-WHERE bucket = ?1",
-                )
-                .into_db_result(db)?;
-            clear_statement1.bind_int64(1, bucket_id)?;
-            clear_statement1.exec()?;
-
-            let clear_statement2 = db
-                .prepare_v2("DELETE FROM ps_oplog WHERE bucket = ?1")
-                .into_db_result(db)?;
-            clear_statement2.bind_int64(1, bucket_id)?;
-            clear_statement2.exec()?;
-
-            // And we need to re-apply all of those.
-            // We also replace the checksum with the checksum of the CLEAR op.
-            // language=SQLite
-            let clear_statement2 = db.prepare_v2(
-                "UPDATE ps_buckets SET last_applied_op = 0, add_checksum = ?1, op_checksum = 0 WHERE id = ?2",
-            )?;
-            clear_statement2.bind_int64(2, bucket_id)?;
-            clear_statement2.bind_int(1, checksum)?;
-            clear_statement2.exec()?;
-
-            add_checksum = 0;
-            is_empty = true;
-            op_checksum = 0;
-        }
-    }
-
-    if let Some(last_op) = &last_op {
-        // language=SQLite
-        let statement = db.prepare_v2(
-            "UPDATE ps_buckets
-                SET last_op = ?2,
-                    add_checksum = (add_checksum + ?3) & 0xffffffff,
-                    op_checksum = (op_checksum + ?4) & 0xffffffff,
-                    count_since_last = count_since_last + ?5
-            WHERE id = ?1",
-        )?;
-        statement.bind_int64(1, bucket_id)?;
-        statement.bind_int64(2, *last_op)?;
-        statement.bind_int(3, add_checksum)?;
-        statement.bind_int(4, op_checksum)?;
-        statement.bind_int(5, added_ops)?;
-
-        statement.exec()?;
-    }
+    let line = serde_json::from_str::<DataLine>(data)?;
+    let adapter = StorageAdapter::new(db)?;
+    insert_bucket_operations(&adapter, &line)?;
 
     Ok(())
 }

crates/core/src/sync/line.rs

@@ -1,8 +1,6 @@
-use core::any::TypeId;
-
+use alloc::borrow::Cow;
 use alloc::string::ToString;
 use alloc::vec::Vec;
-use alloc::{borrow::Cow, string::String};
 use serde::de::DeserializeSeed;
 use serde::{de::Visitor, Deserialize};
 
@@ -108,12 +106,12 @@ pub struct OplogEntry<'a> {
 }
 
 #[derive(Debug)]
-enum OplogData<'a> {
+pub enum OplogData<'a> {
     JsonString { data: Cow<'a, str> },
     BsonDocument { data: Cow<'a, [u8]> },
 }
 
-#[derive(Deserialize, Debug, Clone, Copy)]
+#[derive(Deserialize, Debug, Clone, Copy, PartialEq, Eq)]
 pub enum OpType {
     CLEAR,
     MOVE,

crates/core/src/sync/mod.rs

@@ -3,7 +3,8 @@ use sqlite_nostd::{self as sqlite, ResultCode};
 pub mod bucket_priority;
 mod interface;
 pub mod line;
-mod storage_adapter;
+pub mod operations;
+pub mod storage_adapter;
 mod streaming_sync;
 mod sync_status;