Merge pull request #21 from powersync-ja/client-id

powersync_client_id() / powersync_last_synced_at() / powersync_clear()

Author: rkistner

crates/core/src/error.rs

@@ -1,12 +1,10 @@
 use alloc::string::{String, ToString};
 use core::error::Error;
-use sqlite_nostd::{Connection, ResultCode, sqlite3};
-
+use sqlite_nostd::{sqlite3, Connection, ResultCode};
 
 #[derive(Debug)]
 pub struct SQLiteError(pub ResultCode, pub Option<String>);
 
-
 impl core::fmt::Display for SQLiteError {
     fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
         write!(f, "{:?}", self)
@@ -16,21 +14,10 @@ impl core::fmt::Display for SQLiteError {
 impl Error for SQLiteError {}
 
 pub trait PSResult<T> {
-    fn into_result(self) -> Result<T, SQLiteError>;
     fn into_db_result(self, db: *mut sqlite3) -> Result<T, SQLiteError>;
 }
 
 impl<T> PSResult<T> for Result<T, ResultCode> {
-    fn into_result(self) -> Result<T, SQLiteError> {
-        if let Err(code) = self {
-            Err(SQLiteError(code, None))
-        } else if let Ok(r) = self {
-            Ok(r)
-        } else {
-            Err(SQLiteError(ResultCode::ABORT, None))
-        }
-    }
-
     fn into_db_result(self, db: *mut sqlite3) -> Result<T, SQLiteError> {
         if let Err(code) = self {
             let message = db.errmsg().unwrap_or(String::from("Conversion error"));

crates/core/src/kv.rs

@@ -0,0 +1,88 @@
+extern crate alloc;
+
+use alloc::format;
+use alloc::string::{String, ToString};
+use core::ffi::c_int;
+use core::slice;
+
+use sqlite::ResultCode;
+use sqlite_nostd as sqlite;
+use sqlite_nostd::{Connection, Context};
+
+use crate::create_sqlite_optional_text_fn;
+use crate::create_sqlite_text_fn;
+use crate::error::SQLiteError;
+
+fn powersync_client_id_impl(
+    ctx: *mut sqlite::context,
+    _args: &[*mut sqlite::value],
+) -> Result<String, SQLiteError> {
+    let db = ctx.db_handle();
+
+    // language=SQLite
+    let statement = db.prepare_v2("select value from ps_kv where key = 'client_id'")?;
+
+    if statement.step()? == ResultCode::ROW {
+        let client_id = statement.column_text(0)?;
+        return Ok(client_id.to_string());
+    } else {
+        return Err(SQLiteError(
+            ResultCode::ABORT,
+            Some(format!("No client_id found in ps_kv")),
+        ));
+    }
+}
+
+create_sqlite_text_fn!(
+    powersync_client_id,
+    powersync_client_id_impl,
+    "powersync_last_synced_at"
+);
+
+fn powersync_last_synced_at_impl(
+    ctx: *mut sqlite::context,
+    _args: &[*mut sqlite::value],
+) -> Result<Option<String>, SQLiteError> {
+    let db = ctx.db_handle();
+
+    // language=SQLite
+    let statement = db.prepare_v2("select value from ps_kv where key = 'last_synced_at'")?;
+
+    if statement.step()? == ResultCode::ROW {
+        let client_id = statement.column_text(0)?;
+        return Ok(Some(client_id.to_string()));
+    } else {
+        return Ok(None);
+    }
+}
+
+create_sqlite_optional_text_fn!(
+    powersync_last_synced_at,
+    powersync_last_synced_at_impl,
+    "powersync_last_synced_at"
+);
+
+pub fn register(db: *mut sqlite::sqlite3) -> Result<(), ResultCode> {
+    db.create_function_v2(
+        "powersync_client_id",
+        0,
+        sqlite::UTF8 | sqlite::DETERMINISTIC,
+        None,
+        Some(powersync_client_id),
+        None,
+        None,
+        None,
+    )?;
+    db.create_function_v2(
+        "powersync_last_synced_at",
+        0,
+        sqlite::UTF8 | sqlite::DETERMINISTIC,
+        None,
+        Some(powersync_last_synced_at),
+        None,
+        None,
+        None,
+    )?;
+
+    Ok(())
+}

crates/core/src/lib.rs

@@ -1,5 +1,6 @@
 #![no_std]
 #![feature(vec_into_raw_parts)]
+#![allow(internal_features)]
 #![feature(core_intrinsics)]
 #![feature(error_in_core)]
 #![feature(assert_matches)]
@@ -11,23 +12,24 @@ use core::ffi::{c_char, c_int};
 use sqlite::ResultCode;
 use sqlite_nostd as sqlite;
 
-mod util;
-mod uuid;
-mod views;
-mod view_admin;
-mod macros;
+mod checkpoint;
+mod crud_vtab;
 mod diff;
-mod schema_management;
-mod operations_vtab;
-mod operations;
-mod ext;
 mod error;
-mod crud_vtab;
-mod vtab_util;
+mod ext;
+mod kv;
+mod macros;
+mod operations;
+mod operations_vtab;
+mod schema_management;
 mod sync_local;
-mod checkpoint;
-mod version;
 mod sync_types;
+mod util;
+mod uuid;
+mod version;
+mod view_admin;
+mod views;
+mod vtab_util;
 
 #[no_mangle]
 pub extern "C" fn sqlite3_powersync_init(
@@ -43,7 +45,7 @@ pub extern "C" fn sqlite3_powersync_init(
         code as c_int
     } else {
         ResultCode::OK as c_int
-    }
+    };
 }
 
 fn init_extension(db: *mut sqlite::sqlite3) -> Result<(), ResultCode> {
@@ -53,6 +55,7 @@ fn init_extension(db: *mut sqlite::sqlite3) -> Result<(), ResultCode> {
     crate::diff::register(db)?;
     crate::view_admin::register(db)?;
     crate::checkpoint::register(db)?;
+    crate::kv::register(db)?;
 
     crate::schema_management::register(db)?;
     crate::operations_vtab::register(db)?;
@@ -61,12 +64,17 @@ fn init_extension(db: *mut sqlite::sqlite3) -> Result<(), ResultCode> {
     Ok(())
 }
 
-
 extern "C" {
     #[cfg(feature = "static")]
     #[allow(non_snake_case)]
     pub fn sqlite3_auto_extension(
-        xEntryPoint: Option<extern "C" fn(*mut sqlite::sqlite3, *mut *mut c_char, *mut sqlite::api_routines) -> c_int>,
+        xEntryPoint: Option<
+            extern "C" fn(
+                *mut sqlite::sqlite3,
+                *mut *mut c_char,
+                *mut sqlite::api_routines,
+            ) -> c_int,
+        >,
     ) -> ::core::ffi::c_int;
 }
 

crates/core/src/macros.rs

@@ -1,4 +1,3 @@
-
 #[macro_export]
 macro_rules! create_sqlite_text_fn {
     ($fn_name:ident, $fn_impl_name:ident, $description:literal) => {
@@ -31,6 +30,41 @@ macro_rules! create_sqlite_text_fn {
     };
 }
 
+#[macro_export]
+macro_rules! create_sqlite_optional_text_fn {
+    ($fn_name:ident, $fn_impl_name:ident, $description:literal) => {
+        extern "C" fn $fn_name(
+            ctx: *mut sqlite::context,
+            argc: c_int,
+            argv: *mut *mut sqlite::value,
+        ) {
+            let args = sqlite::args!(argc, argv);
+
+            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);
+            } else if let Ok(r) = result {
+                if let Some(s) = r {
+                    ctx.result_text_transient(&s);
+                } else {
+                    ctx.result_null();
+                }
+            }
+        }
+    };
+}
 
 // Wrap a function in an auto-transaction.
 // Gives the equivalent of SQLite's auto-commit behaviour, except that applies to all statements

crates/core/src/operations.rs

@@ -1,15 +1,11 @@
 use alloc::format;
 use alloc::string::{String, ToString};
-use alloc::vec::Vec;
-use serde::{Deserialize, Deserializer, Serialize};
-use serde_json as json;
 
 use crate::error::{PSResult, SQLiteError};
 use sqlite_nostd as sqlite;
 use sqlite_nostd::{Connection, ResultCode};
 
 use crate::ext::SafeManagedStmt;
-use crate::sync_types::{BucketChecksum, Checkpoint, StreamingSyncLine};
 use crate::util::*;
 
 // Run inside a transaction
@@ -142,7 +138,7 @@ INSERT INTO ps_oplog(bucket, op_id, op, key, row_type, row_id, data, hash, super
             supersede_statement.reset()?;
 
             let should_skip_remove = !superseded && op == "REMOVE";
-            if (should_skip_remove) {
+            if should_skip_remove {
                 // If a REMOVE statement did not replace (supersede) any previous
                 // operations, we do not need to persist it.
                 // The same applies if the bucket was not synced to the local db yet,
@@ -307,9 +303,3 @@ pub fn delete_bucket(db: *mut sqlite::sqlite3, name: &str) -> Result<(), SQLiteE
 
     Ok(())
 }
-
-pub fn stream_operation(db: *mut sqlite::sqlite3, data: &str) -> Result<(), SQLiteError> {
-    let line: StreamingSyncLine = serde_json::from_str(data)?;
-
-    Ok(())
-}

crates/core/src/operations_vtab.rs

@@ -7,22 +7,21 @@ use core::slice;
 use sqlite::{Connection, ResultCode, Value};
 use sqlite_nostd as sqlite;
 
-use crate::operations::{clear_remove_ops, delete_bucket, delete_pending_buckets, insert_operation, stream_operation};
+use crate::operations::{
+    clear_remove_ops, delete_bucket, delete_pending_buckets, insert_operation,
+};
 use crate::sync_local::sync_local;
-use crate::sync_types::Checkpoint;
 use crate::vtab_util::*;
 
 #[repr(C)]
 struct VirtualTable {
     base: sqlite::vtab,
     db: *mut sqlite::sqlite3,
 
-    target_checkpoint: Option<Checkpoint>,
     target_applied: bool,
-    target_validated: bool
+    target_validated: bool,
 }
 
-
 extern "C" fn connect(
     db: *mut sqlite::sqlite3,
     _aux: *mut c_void,
@@ -31,7 +30,8 @@ extern "C" fn connect(
     vtab: *mut *mut sqlite::vtab,
     _err: *mut *mut c_char,
 ) -> c_int {
-    if let Err(rc) = sqlite::declare_vtab(db, "CREATE TABLE powersync_operations(op TEXT, data TEXT);")
+    if let Err(rc) =
+        sqlite::declare_vtab(db, "CREATE TABLE powersync_operations(op TEXT, data TEXT);")
     {
         return rc as c_int;
     }
@@ -44,9 +44,8 @@ extern "C" fn connect(
                 zErrMsg: core::ptr::null_mut(),
             },
             db,
-            target_checkpoint: None,
             target_validated: false,
-            target_applied: false
+            target_applied: false,
         }));
         *vtab = tab.cast::<sqlite::vtab>();
         let _ = sqlite::vtab_config(db, 0);
@@ -102,10 +101,7 @@ extern "C" fn update(
         } else if op == "delete_bucket" {
             let result = delete_bucket(db, data);
             vtab_result(vtab, result)
-        } else if op == "stream" {
-            let result = stream_operation(db, data);
-            vtab_result(vtab, result)
-        }  else {
+        } else {
             ResultCode::MISUSE as c_int
         }
     } else {