update sentinels for pk only rows too

Signed-off-by: Somtochi Onyekwere <[email protected]>

Author: somtochiama

core/rs/core/src/local_writes/after_insert.rs

@@ -46,20 +46,27 @@ fn after_insert(
     let (create_record_existed, key_new) = tbl_info
         .get_or_create_key_for_insert(db, pks_new)
         .map_err(|_| "failed getting or creating lookaside key")?;
-    if tbl_info.non_pks.is_empty() {
+
+    let cl = if tbl_info.non_pks.is_empty() {
         let seq = bump_seq(ext_data);
         // just a sentinel record
-        return super::mark_new_pk_row_created(db, tbl_info, key_new, db_version, seq, &ts);
-    } else if create_record_existed {
-        // update the create record since it already exists.
-        let seq = bump_seq(ext_data);
-        let cl = update_create_record(db, tbl_info, key_new, db_version, seq, &ts)?;
-        if let Some(cl) = cl {
-            tbl_info.set_cl(key_new, cl);
-        }
-    }
+        let cl = super::mark_new_pk_row_created(db, tbl_info, key_new, db_version, seq, &ts)?;
+        Some(cl)
+    } else {
+        let cl = if create_record_existed {
+            // update the create record since it already exists.
+            let seq = bump_seq(ext_data);
+            update_create_record(db, tbl_info, key_new, db_version, seq, &ts)?
+        } else {
+            None
+        };
+        super::mark_locally_inserted(db, ext_data, tbl_info, key_new, db_version, &ts)?;
+        cl
+    };
 
-    super::mark_locally_inserted(db, ext_data, tbl_info, key_new, db_version, &ts)?;
+    if let Some(cl) = cl {
+        tbl_info.set_cl(key_new, cl);
+    }
 
     Ok(ResultCode::OK)
 }
@@ -94,22 +101,15 @@ fn update_create_record(
         .map_err(|_e| "failed binding to update_create_record_stmt")?;
 
     let res = update_create_record_stmt.step();
-    match res {
+    let result = match res {
         Ok(ResultCode::ROW) => {
             let col_version = update_create_record_stmt.column_int64(0);
-            super::reset_cached_stmt(update_create_record_stmt.stmt)
-                .map_err(|_e| "failed to reset cached stmt")?;
             Ok(Some(col_version))
         }
-        Ok(ResultCode::DONE) => {
-            super::reset_cached_stmt(update_create_record_stmt.stmt)
-                .map_err(|_e| "failed to reset cached stmt")?;
-            Ok(None)
-        }
-        _ => {
-            super::reset_cached_stmt(update_create_record_stmt.stmt)
-                .map_err(|_e| "failed to reset cached stmt")?;
-            Err("failed to step update_create_record_stmt".to_string())
-        }
-    }
+        Ok(ResultCode::DONE) => Ok(None),
+        _ => Err("failed to step update_create_record_stmt".to_string()),
+    };
+    super::reset_cached_stmt(update_create_record_stmt.stmt)
+        .map_err(|_e| "failed to reset cached stmt")?;
+    result
 }

core/rs/core/src/local_writes/after_update.rs

@@ -85,7 +85,7 @@ fn after_update(
         if crate::compare_values::any_value_changed(pks_new, pks_old)? {
             let old_key = tbl_info
                 .get_or_create_key_via_raw_values(db, pks_old)
-                .map_err(|_| "failed geteting or creating lookaside key")?;
+                .map_err(|_| "failed getting or creating lookaside key")?;
             let next_seq = super::bump_seq(ext_data);
             changed = true;
             // Record the delete of the row identified by the old primary keys

core/rs/core/src/local_writes/mod.rs

@@ -84,7 +84,7 @@ fn mark_new_pk_row_created(
     db_version: i64,
     seq: i32,
     ts: &str,
-) -> Result<ResultCode, String> {
+) -> Result<i64, String> {
     let mark_locally_created_stmt_ref = tbl_info
         .get_mark_locally_created_stmt(db)
         .map_err(|_e| "failed to get mark_locally_created_stmt")?;
@@ -98,7 +98,17 @@ fn mark_new_pk_row_created(
         .and_then(|_| mark_locally_created_stmt.bind_int(3, seq))
         .and_then(|_| mark_locally_created_stmt.bind_text(4, ts, sqlite::Destructor::STATIC))
         .map_err(|_| "failed binding to mark_locally_created_stmt")?;
-    step_trigger_stmt(mark_locally_created_stmt)
+    let stmt_res = mark_locally_created_stmt.step();
+    let result = match stmt_res {
+        Ok(ResultCode::ROW) => {
+            let cl = mark_locally_created_stmt.column_int64(0);
+            Ok(cl)
+        }
+        _ => Err("failed to step mark locally created stmt".to_string()),
+    };
+    reset_cached_stmt(mark_locally_created_stmt.stmt)
+        .map_err(|_e| "failed to reset cached stmt")?;
+    result
 }
 
 fn bump_seq(ext_data: *mut crsql_ExtData) -> c_int {
@@ -291,19 +301,18 @@ fn mark_locally_deleted(
         .and_then(|_| mark_locally_deleted_stmt.bind_text(4, &ts, sqlite::Destructor::STATIC))
         .map_err(|_| "failed binding to mark locally deleted stmt")?;
 
-    let res = mark_locally_deleted_stmt.step();
+    let stmt_res = mark_locally_deleted_stmt.step();
 
-    match res {
+    let result = match stmt_res {
         Ok(ResultCode::ROW) => {
             let cl = mark_locally_deleted_stmt.column_int64(0);
-            reset_cached_stmt(mark_locally_deleted_stmt.stmt)
-                .map_err(|_e| "failed to reset cached stmt")?;
             Ok(cl)
         }
-        _ => {
-            reset_cached_stmt(mark_locally_deleted_stmt.stmt)
-                .map_err(|_e| "failed to reset cached stmt")?;
-            Err("failed to step mark locally deleted stmt".to_string())
-        }
-    }
+        _ => Err("failed to step mark locally deleted stmt".to_string()),
+    };
+
+    reset_cached_stmt(mark_locally_deleted_stmt.stmt)
+        .map_err(|_e| "failed to reset cached stmt")?;
+
+    result
 }

core/rs/core/src/tableinfo.rs

@@ -28,6 +28,7 @@ use sqlite_nostd::ResultCode;
 use sqlite_nostd::Stmt;
 use sqlite_nostd::StrRef;
 
+// TODO: make this configurable with a crsql_config_set.
 const MAX_CL_CACHE_SIZE: usize = 1500;
 pub struct TableInfo {
     pub tbl_name: String,
@@ -561,7 +562,8 @@ impl TableInfo {
                   db_version = excluded.db_version,
                   seq = excluded.seq,
                   site_id = 0,
-                  ts = excluded.ts",
+                  ts = excluded.ts
+                  RETURNING col_version",
               table_name = crate::util::escape_ident(&self.tbl_name),
               sentinel = crate::c::INSERT_SENTINEL,
             );