Add last_synced_at.

rkistner · rkistner · commit e259a2e82eb2 · 2024-08-20T14:37:55.000+02:00
diff --git a/crates/core/src/kv.rs b/crates/core/src/kv.rs
@@ -12,6 +12,7 @@ use sqlite::ResultCode;
 use sqlite_nostd as sqlite;
 use sqlite_nostd::{Connection, Context, Value};
 
+use crate::create_sqlite_optional_text_fn;
 use crate::create_sqlite_text_fn;
 use crate::error::SQLiteError;
 use crate::sync_types::Checkpoint;
@@ -39,7 +40,30 @@ fn powersync_client_id_impl(
 create_sqlite_text_fn!(
     powersync_client_id,
     powersync_client_id_impl,
-    "powersync_client_id"
+    "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> {
@@ -53,6 +77,16 @@ pub fn register(db: *mut sqlite::sqlite3) -> Result<(), ResultCode> {
         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(())
 }
diff --git a/crates/core/src/lib.rs b/crates/core/src/lib.rs
@@ -12,11 +12,11 @@ use sqlite::ResultCode;
 use sqlite_nostd as sqlite;
 
 mod checkpoint;
-mod client_id;
 mod crud_vtab;
 mod diff;
 mod error;
 mod ext;
+mod kv;
 mod macros;
 mod operations;
 mod operations_vtab;
@@ -54,7 +54,7 @@ fn init_extension(db: *mut sqlite::sqlite3) -> Result<(), ResultCode> {
     crate::diff::register(db)?;
     crate::view_admin::register(db)?;
     crate::checkpoint::register(db)?;
-    crate::client_id::register(db)?;
+    crate::kv::register(db)?;
 
     crate::schema_management::register(db)?;
     crate::operations_vtab::register(db)?;
diff --git a/crates/core/src/macros.rs b/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
diff --git a/crates/core/src/sync_local.rs b/crates/core/src/sync_local.rs
@@ -2,20 +2,21 @@ use alloc::collections::BTreeSet;
 use alloc::format;
 use alloc::string::String;
 
+use crate::error::{PSResult, SQLiteError};
 use sqlite_nostd as sqlite;
 use sqlite_nostd::{ColumnType, Connection, ResultCode};
-use crate::error::{SQLiteError, PSResult};
 
 use crate::ext::SafeManagedStmt;
 use crate::util::{internal_table_name, quote_internal_name};
 
-
 pub fn can_update_local(db: *mut sqlite::sqlite3) -> Result<bool, SQLiteError> {
     // language=SQLite
-    let statement = db.prepare_v2("\
+    let statement = db.prepare_v2(
+        "\
 SELECT group_concat(name)
 FROM ps_buckets
-WHERE target_op > last_op AND (name = '$local' OR pending_delete = 0)")?;
+WHERE target_op > last_op AND (name = '$local' OR pending_delete = 0)",
+    )?;
 
     if statement.step()? != ResultCode::ROW {
         return Err(SQLiteError::from(ResultCode::ABORT));
@@ -36,15 +37,15 @@ WHERE target_op > last_op AND (name = '$local' OR pending_delete = 0)")?;
     Ok(true)
 }
 
-pub fn sync_local(
-    db: *mut sqlite::sqlite3, _data: &str) -> Result<i64, SQLiteError> {
-
+pub fn sync_local(db: *mut sqlite::sqlite3, _data: &str) -> Result<i64, SQLiteError> {
     if !can_update_local(db)? {
         return Ok(0);
     }
 
     // language=SQLite
-    let statement = db.prepare_v2("SELECT name FROM sqlite_master WHERE type='table' AND name GLOB 'ps_data_*'").into_db_result(db)?;
+    let statement = db
+        .prepare_v2("SELECT name FROM sqlite_master WHERE type='table' AND name GLOB 'ps_data_*'")
+        .into_db_result(db)?;
     let mut tables: BTreeSet<String> = BTreeSet::new();
 
     while statement.step()? == ResultCode::ROW {
@@ -60,7 +61,9 @@ pub fn sync_local(
     // |--SEARCH r USING INDEX ps_oplog_by_row (row_type=? AND row_id=?)
     // `--USE TEMP B-TREE FOR GROUP BY
     // language=SQLite
-    let statement = db.prepare_v2("\
+    let statement = db
+        .prepare_v2(
+            "\
 -- 3. Group the objects from different buckets together into a single one (ops).
 SELECT r.row_type as type,
     r.row_id as id,
@@ -79,7 +82,9 @@ FROM ps_buckets AS buckets
 WHERE r.superseded = 0
 AND b.superseded = 0
 -- Group for (3)
-GROUP BY r.row_type, r.row_id").into_db_result(db)?;
+GROUP BY r.row_type, r.row_id",
+        )
+        .into_db_result(db)?;
 
     // TODO: cache statements
 
@@ -96,12 +101,16 @@ GROUP BY r.row_type, r.row_id").into_db_result(db)?;
 
             if buckets == "[]" {
                 // DELETE
-                let delete_statement = db.prepare_v2(&format!("DELETE FROM {} WHERE id = ?", quoted)).into_db_result(db)?;
+                let delete_statement = db
+                    .prepare_v2(&format!("DELETE FROM {} WHERE id = ?", quoted))
+                    .into_db_result(db)?;
                 delete_statement.bind_text(1, id, sqlite::Destructor::STATIC)?;
                 delete_statement.exec()?;
             } else {
                 // INSERT/UPDATE
-                let insert_statement = db.prepare_v2(&format!("REPLACE INTO {}(id, data) VALUES(?, ?)", quoted)).into_db_result(db)?;
+                let insert_statement = db
+                    .prepare_v2(&format!("REPLACE INTO {}(id, data) VALUES(?, ?)", quoted))
+                    .into_db_result(db)?;
                 insert_statement.bind_text(1, id, sqlite::Destructor::STATIC)?;
                 insert_statement.bind_text(2, data?, sqlite::Destructor::STATIC)?;
                 insert_statement.exec()?;
@@ -110,14 +119,18 @@ GROUP BY r.row_type, r.row_id").into_db_result(db)?;
             if buckets == "[]" {
                 // DELETE
                 // language=SQLite
-                let delete_statement = db.prepare_v2("DELETE FROM ps_untyped WHERE type = ? AND id = ?").into_db_result(db)?;
+                let delete_statement = db
+                    .prepare_v2("DELETE FROM ps_untyped WHERE type = ? AND id = ?")
+                    .into_db_result(db)?;
                 delete_statement.bind_text(1, type_name, sqlite::Destructor::STATIC)?;
                 delete_statement.bind_text(2, id, sqlite::Destructor::STATIC)?;
                 delete_statement.exec()?;
             } else {
                 // INSERT/UPDATE
                 // language=SQLite
-                let insert_statement = db.prepare_v2("REPLACE INTO ps_untyped(type, id, data) VALUES(?, ?, ?)").into_db_result(db)?;
+                let insert_statement = db
+                    .prepare_v2("REPLACE INTO ps_untyped(type, id, data) VALUES(?, ?, ?)")
+                    .into_db_result(db)?;
                 insert_statement.bind_text(1, type_name, sqlite::Destructor::STATIC)?;
                 insert_statement.bind_text(2, id, sqlite::Destructor::STATIC)?;
                 insert_statement.bind_text(3, data?, sqlite::Destructor::STATIC)?;
@@ -127,9 +140,16 @@ GROUP BY r.row_type, r.row_id").into_db_result(db)?;
     }
 
     // language=SQLite
-    db.exec_safe("UPDATE ps_buckets
+    db.exec_safe(
+        "UPDATE ps_buckets
                  SET last_applied_op = last_op
-                 WHERE last_applied_op != last_op").into_db_result(db)?;
+                 WHERE last_applied_op != last_op",
+    )
+    .into_db_result(db)?;
+
+    // language=SQLite
+    db.exec_safe("insert or replace into ps_kv(key, value) values('last_synced_at', datetime())")
+        .into_db_result(db)?;
 
     Ok(1)
 }
