Start tracking checkpoints

Author: simolus3

Cargo.lock

@@ -9,7 +9,7 @@ default-members = ["crates/shell", "crates/sqlite"]
 
 [profile.dev]
 panic = "abort"
-strip = true
+#strip = true
 
 [profile.release]
 panic = "abort"

Cargo.toml

@@ -18,10 +18,11 @@ bytes = { version = "1.4", default-features = false }
 num-traits = { version = "0.2.15", default-features = false }
 num-derive = "0.3"
 serde_json = { version = "1.0", default-features = false, features = ["alloc"] }
-serde = { version = "1.0", default-features = false, features = ["alloc", "derive"] }
+serde = { version = "1.0", default-features = false, features = ["alloc", "derive", "rc"] }
 streaming-iterator = { version = "0.1.9", default-features = false, features = ["alloc"] }
 lock_api = { version = "0.4.12", default-features = false }
-futures-lite = { version = "2.6.0", default-features = false }
+futures-lite = { version = "2.6.0", default-features = false, features = ["alloc"] }
+rustc-hash = { version = "2.1", default-features = false }
 
 [dependencies.uuid]
 version = "1.4.1"

crates/core/Cargo.toml

@@ -2,7 +2,6 @@ use core::fmt::Display;
 
 use alloc::{
     boxed::Box,
-    format,
     string::{String, ToString},
 };
 use serde::de::{self, StdError};

crates/core/src/bson/error.rs

@@ -8,10 +8,10 @@ use sqlite::ResultCode;
 use sqlite_nostd as sqlite;
 use sqlite_nostd::{Connection, Context};
 
-use crate::bucket_priority::BucketPriority;
 use crate::create_sqlite_optional_text_fn;
 use crate::create_sqlite_text_fn;
 use crate::error::SQLiteError;
+use crate::sync::bucket_priority::BucketPriority;
 
 fn powersync_client_id_impl(
     ctx: *mut sqlite::context,

crates/core/src/kv.rs

@@ -13,7 +13,6 @@ use sqlite::ResultCode;
 use sqlite_nostd as sqlite;
 
 mod bson;
-mod bucket_priority;
 mod checkpoint;
 mod crud_vtab;
 mod diff;
@@ -62,6 +61,7 @@ fn init_extension(db: *mut sqlite::sqlite3) -> Result<(), ResultCode> {
     crate::view_admin::register(db)?;
     crate::checkpoint::register(db)?;
     crate::kv::register(db)?;
+    sync::register(db)?;
 
     crate::schema::register(db)?;
     crate::operations_vtab::register(db)?;

crates/core/src/lib.rs

@@ -11,18 +11,7 @@ macro_rules! create_sqlite_text_fn {
             let result = $fn_impl_name(ctx, args);
 
             if let Err(err) = result {
-                let SQLiteError(code, message) = SQLiteError::from(err);
-                if message.is_some() {
-                    ctx.result_error(&format!("{:} {:}", $description, message.unwrap()));
-                } else {
-                    let error = ctx.db_handle().errmsg().unwrap();
-                    if error == "not an error" {
-                        ctx.result_error(&format!("{:}", $description));
-                    } else {
-                        ctx.result_error(&format!("{:} {:}", $description, error));
-                    }
-                }
-                ctx.result_error_code(code);
+                crate::util::context_set_error(ctx, SQLiteError::from(err), $description);
             } else if let Ok(r) = result {
                 ctx.result_text_transient(&r);
             }
@@ -43,18 +32,7 @@ macro_rules! create_sqlite_optional_text_fn {
             let result = $fn_impl_name(ctx, args);
 
             if let Err(err) = result {
-                let SQLiteError(code, message) = SQLiteError::from(err);
-                if message.is_some() {
-                    ctx.result_error(&format!("{:} {:}", $description, message.unwrap()));
-                } else {
-                    let error = ctx.db_handle().errmsg().unwrap();
-                    if error == "not an error" {
-                        ctx.result_error(&format!("{:}", $description));
-                    } else {
-                        ctx.result_error(&format!("{:} {:}", $description, error));
-                    }
-                }
-                ctx.result_error_code(code);
+                crate::util::context_set_error(ctx, SQLiteError::from(err), $description);
             } else if let Ok(r) = result {
                 if let Some(s) = r {
                     ctx.result_text_transient(&s);

crates/core/src/macros.rs

@@ -8,9 +8,9 @@ use sqlite::ResultCode;
 use sqlite_nostd as sqlite;
 use sqlite_nostd::{Connection, Context};
 
-use crate::bucket_priority::BucketPriority;
 use crate::error::{PSResult, SQLiteError};
 use crate::fix035::apply_v035_fix;
+use crate::sync::bucket_priority::BucketPriority;
 
 pub fn powersync_migrate(
     ctx: *mut sqlite::context,

crates/core/src/migrations.rs

@@ -1,10 +1,10 @@
-use serde::{de::Visitor, Deserialize};
+use serde::{de::Visitor, Deserialize, Serialize};
 use sqlite_nostd::ResultCode;
 
 use crate::error::SQLiteError;
 
 #[repr(transparent)]
-#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
 pub struct BucketPriority {
     pub number: i32,
 }
@@ -14,6 +14,9 @@ impl BucketPriority {
         self == BucketPriority::HIGHEST
     }
 
+    /// The priority to use when the sync service doesn't attach priorities in checkpoints.
+    pub const FALLBACK: BucketPriority = BucketPriority { number: 3 };
+
     pub const HIGHEST: BucketPriority = BucketPriority { number: 0 };
 
     /// A low priority used to represent fully-completed sync operations across all priorities.
@@ -87,3 +90,12 @@ impl<'de> Deserialize<'de> for BucketPriority {
         deserializer.deserialize_i32(PriorityVisitor)
     }
 }
+
+impl Serialize for BucketPriority {
+    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
+    where
+        S: serde::Serializer,
+    {
+        serializer.serialize_i32(self.number)
+    }
+}

crates/core/src/bucket_priority.rs renamed to crates/core/src/sync/bucket_priority.rs

@@ -1,18 +1,50 @@
+use core::cell::RefCell;
+use core::ffi::{c_int, c_void};
+
+use alloc::boxed::Box;
+use alloc::rc::Rc;
+use alloc::string::ToString;
 use alloc::{string::String, vec::Vec};
 use serde::Serialize;
+use sqlite::{ResultCode, Value};
+use sqlite_nostd::{self as sqlite, ColumnType};
+use sqlite_nostd::{Connection, Context};
+
+use crate::error::SQLiteError;
+use crate::util::context_set_error;
+
+use super::streaming_sync::SyncClient;
+use super::sync_status::DownloadSyncStatus;
+
+pub enum SyncControlRequest<'a> {
+    StartSyncStream {
+        parameters: Option<serde_json::Map<String, serde_json::Value>>,
+    },
+    StopSyncStream,
+    SyncEvent(SyncEvent<'a>),
+}
 
 pub enum SyncEvent<'a> {
-    StartSyncStream,
-    SyncStreamClosed { error: bool },
+    Initialize,
+    TearDown,
     TextLine { data: &'a str },
     BinaryLine { data: &'a [u8] },
 }
 
 /// An instruction sent by the core extension to the SDK.
 #[derive(Serialize)]
 pub enum Instruction {
-    LogLine { severity: LogSeverity, line: String },
-    EstablishSyncStream { request: StreamingSyncRequest },
+    LogLine {
+        severity: LogSeverity,
+        line: String,
+    },
+    UpdateSyncStatus {
+        status: Rc<RefCell<DownloadSyncStatus>>,
+    },
+    EstablishSyncStream {
+        request: StreamingSyncRequest,
+    },
+    FlushFileSystem,
     CloseSyncStream,
 }
 
@@ -37,3 +69,81 @@ pub struct BucketRequest {
     pub name: String,
     pub after: String,
 }
+
+struct SqlController {
+    client: SyncClient,
+}
+
+pub fn register(db: *mut sqlite::sqlite3) -> Result<(), ResultCode> {
+    extern "C" fn control(
+        ctx: *mut sqlite::context,
+        argc: c_int,
+        argv: *mut *mut sqlite::value,
+    ) -> () {
+        let result = (|| -> Result<(), SQLiteError> {
+            let controller = unsafe { ctx.user_data().cast::<SqlController>().as_ref() }
+                .ok_or_else(|| SQLiteError::from(ResultCode::INTERNAL))?;
+
+            let args = sqlite::args!(argc, argv);
+            let [op, payload] = args else {
+                return Err(ResultCode::MISUSE.into());
+            };
+
+            if op.value_type() != ColumnType::Text {
+                return Err(SQLiteError(
+                    ResultCode::MISUSE,
+                    Some("First argument must be a string".to_string()),
+                ));
+            }
+
+            let op = op.text();
+            let event = match op {
+                "start" => SyncControlRequest::StartSyncStream {
+                    parameters: if payload.value_type() == ColumnType::Text {
+                        Some(serde_json::from_str(payload.text())?)
+                    } else {
+                        None
+                    },
+                },
+                "stop" => SyncControlRequest::StopSyncStream,
+                _ => {
+                    return Err(SQLiteError(
+                        ResultCode::MISUSE,
+                        Some("Unknown operation".to_string()),
+                    ))
+                }
+            };
+
+            let instructions = controller.client.push_event(event)?;
+            let formatted = serde_json::to_string(&instructions)?;
+            ctx.result_text_transient(&formatted);
+
+            Ok(())
+        })();
+
+        if let Err(e) = result {
+            context_set_error(ctx, e, "powersync_control");
+        }
+    }
+
+    unsafe extern "C" fn destroy(ptr: *mut c_void) {
+        drop(Box::from_raw(ptr.cast::<SqlController>()));
+    }
+
+    let controller = Box::new(SqlController {
+        client: SyncClient::new(db),
+    });
+
+    db.create_function_v2(
+        "powersync_control",
+        2,
+        sqlite::UTF8 | sqlite::DIRECTONLY,
+        Some(Box::into_raw(controller).cast()),
+        Some(control),
+        None,
+        None,
+        Some(destroy),
+    )?;
+
+    Ok(())
+}