diff --git a/crates/core/src/crud_vtab.rs b/crates/core/src/crud_vtab.rs index 80a79ebd..0f91db6b 100644 --- a/crates/core/src/crud_vtab.rs +++ b/crates/core/src/crud_vtab.rs @@ -7,12 +7,10 @@ use core::ffi::{CStr, c_char, c_int, c_void}; use serde::Serialize; use serde_json::value::RawValue; -use powersync_sqlite_nostd::ManagedStmt; use powersync_sqlite_nostd::{self as sqlite, ColumnType}; -use sqlite::{Connection, ResultCode, Value}; +use sqlite::{ResultCode, Value}; -use crate::error::PowerSyncError; -use crate::ext::SafeManagedStmt; +use crate::error::{PowerSyncError, Result}; use crate::schema::TableInfoFlags; use crate::state::DatabaseState; use crate::sync::storage_adapter::{ @@ -20,6 +18,7 @@ use crate::sync::storage_adapter::{ TARGET_CHECKPOINT_REQUEST_ID_KEY, }; use crate::utils::MAX_OP_ID; +use crate::utils::database::{Database, Statement}; use crate::vtab_util::*; const MANUAL_NAME: &CStr = c"powersync_crud_"; @@ -42,7 +41,7 @@ const SIMPLE_NAME: &CStr = c"powersync_crud"; #[repr(C)] struct VirtualTable { base: sqlite::vtab, - db: *mut sqlite::sqlite3, + db: Database, current_tx: Option, is_simple: bool, state: Rc, @@ -60,13 +59,13 @@ enum CrudTransactionMode { #[derive(Default)] struct ManualCrudTransactionMode { - stmt: Option, + stmt: Option, } #[derive(Default)] struct SimpleCrudTransactionMode { - stmt: Option, - set_updated_rows: Option, + stmt: Option, + set_updated_rows: Option, had_writes: bool, } @@ -84,7 +83,7 @@ impl VirtualTable { } } - fn handle_insert(&mut self, args: &[*mut sqlite::value]) -> Result<(), PowerSyncError> { + fn handle_insert(&mut self, args: &[*mut sqlite::value]) -> Result<()> { let current_tx = self .current_tx .as_mut() @@ -182,13 +181,13 @@ impl VirtualTable { Ok(()) } - fn begin(&mut self) -> Result<(), PowerSyncError> { + fn begin(&mut self) -> Result<()> { let db = self.db; // language=SQLite let statement = db.prepare_v2("UPDATE ps_tx SET next_tx = next_tx + 1 WHERE id = 1 RETURNING next_tx")?; - let tx_id = if statement.step()? == ResultCode::ROW { + let tx_id = if statement.step()? { statement.column_int64(0) - 1 } else { return Err(PowerSyncError::unknown_internal()); @@ -212,7 +211,7 @@ impl VirtualTable { } impl ManualCrudTransactionMode { - fn raw_crud_statement(&mut self, db: *mut sqlite::sqlite3) -> Result<&ManagedStmt, ResultCode> { + fn raw_crud_statement(&mut self, db: Database) -> Result<&Statement> { prepare_lazy(&mut self.stmt, || { const SQL: &str = formatcp!( "\ @@ -223,39 +222,33 @@ SELECT * FROM insertion WHERE (NOT (?3 & {})) OR data->>'op' != 'PATCH' OR data- TableInfoFlags::IGNORE_EMPTY_UPDATE ); - db.prepare_v3(SQL, 0) + db.prepare_v2(SQL) }) } } impl SimpleCrudTransactionMode { - fn raw_crud_statement(&mut self, db: *mut sqlite::sqlite3) -> Result<&ManagedStmt, ResultCode> { + fn raw_crud_statement(&mut self, db: Database) -> Result<&Statement> { prepare_lazy(&mut self.stmt, || { // language=SQLite - db.prepare_v3("INSERT INTO ps_crud(tx_id, data) VALUES (?, ?)", 0) + db.prepare_v2("INSERT INTO ps_crud(tx_id, data) VALUES (?, ?)") }) } - fn set_updated_rows_statement( - &mut self, - db: *mut sqlite::sqlite3, - ) -> Result<&ManagedStmt, ResultCode> { + fn set_updated_rows_statement(&mut self, db: Database) -> Result<&Statement> { prepare_lazy(&mut self.set_updated_rows, || { // language=SQLite - db.prepare_v3( - "INSERT OR IGNORE INTO ps_updated_rows(row_type, row_id) VALUES(?, ?)", - 0, - ) + db.prepare_v2("INSERT OR IGNORE INTO ps_updated_rows(row_type, row_id) VALUES(?, ?)") }) } - fn record_local_write(&mut self, db: *mut sqlite::sqlite3) -> Result<(), ResultCode> { + fn record_local_write(&mut self, db: Database) -> Result<()> { if !self.had_writes { // Also clear the seen/applied high-water marks: checkpoint request ids observed before // this write can't acknowledge it, and stale values may predate a request counter // restart. Keeping them around could open the apply gate for a newly allocated target // id that compares below a stale seen value. - db.exec_safe(formatcp!( + db.exec_safe_str(formatcp!( "INSERT OR REPLACE INTO ps_kv(key, value) VALUES('{TARGET_CHECKPOINT_REQUEST_ID_KEY}', {MAX_OP_ID}); DELETE FROM ps_kv WHERE key IN ('{LAST_SEEN_CHECKPOINT_REQUEST_ID_KEY}', '{LAST_APPLIED_CHECKPOINT_REQUEST_ID_KEY}')" ))?; @@ -268,9 +261,9 @@ DELETE FROM ps_kv WHERE key IN ('{LAST_SEEN_CHECKPOINT_REQUEST_ID_KEY}', '{LAST_ /// A variant of `Option.get_or_insert` that handles insertions returning errors. fn prepare_lazy( - stmt: &mut Option, - prepare: impl FnOnce() -> Result, -) -> Result<&ManagedStmt, ResultCode> { + stmt: &mut Option, + prepare: impl FnOnce() -> Result, +) -> Result<&Statement> { if let None = stmt { *stmt = Some(prepare()?); } @@ -312,7 +305,7 @@ extern "C" fn connect( zErrMsg: core::ptr::null_mut(), }, state: DatabaseState::clone_from(aux), - db, + db: db.into(), current_tx: None, is_simple, })); @@ -403,7 +396,10 @@ static MODULE: sqlite::module = sqlite::module { xIntegrity: None, }; -pub fn register(db: *mut sqlite::sqlite3, state: Rc) -> Result<(), ResultCode> { +pub fn register( + db: *mut sqlite::sqlite3, + state: Rc, +) -> core::result::Result<(), ResultCode> { sqlite::convert_rc(sqlite::create_module_v2( db, SIMPLE_NAME.as_ptr(), diff --git a/crates/core/src/diff.rs b/crates/core/src/diff.rs index 49fc652b..28235625 100644 --- a/crates/core/src/diff.rs +++ b/crates/core/src/diff.rs @@ -9,14 +9,11 @@ use sqlite::ResultCode; use crate::constants::SUBTYPE_JSON; use crate::create_sqlite_text_fn; -use crate::error::PowerSyncError; +use crate::error::{PowerSyncError, Result}; use powersync_sqlite_nostd::bindings::SQLITE_RESULT_SUBTYPE; use serde_json as json; -fn powersync_diff_impl( - ctx: *mut sqlite::context, - args: &[*mut sqlite::value], -) -> Result { +fn powersync_diff_impl(ctx: *mut sqlite::context, args: &[*mut sqlite::value]) -> Result { let data_old = args[0].text(); let data_new = args[1].text(); @@ -24,7 +21,7 @@ fn powersync_diff_impl( diff_objects(data_old, data_new) } -pub fn diff_objects(data_old: &str, data_new: &str) -> Result { +pub fn diff_objects(data_old: &str, data_new: &str) -> Result { let v_new: json::Value = json::from_str(data_new).map_err(PowerSyncError::as_argument_error)?; let v_old: json::Value = json::from_str(data_old).map_err(PowerSyncError::as_argument_error)?; @@ -64,7 +61,7 @@ pub fn diff_objects(data_old: &str, data_new: &str) -> Result Result<(), ResultCode> { +pub fn register(db: *mut sqlite::sqlite3) -> core::result::Result<(), ResultCode> { db.create_function_v2( "powersync_diff", 2, diff --git a/crates/core/src/error.rs b/crates/core/src/error.rs index e9a8856d..d1ace5f5 100644 --- a/crates/core/src/error.rs +++ b/crates/core/src/error.rs @@ -1,8 +1,13 @@ -use core::{error::Error, ffi::c_int, fmt::Display}; +use core::{ + error::Error, + ffi::{CStr, c_int}, + fmt::Display, +}; use alloc::{ - borrow::Cow, + borrow::{Cow, ToOwned}, boxed::Box, + ffi::{CString, NulError}, string::{String, ToString}, }; use num_traits::FromPrimitive; @@ -14,6 +19,8 @@ use crate::{ constants::{CORE_PKG_VERSION, MIN_SQLITE_VERSION_NUMBER}, }; +pub type Result = core::result::Result; + /// A [RawPowerSyncError], but boxed. /// /// We allocate errors in boxes to avoid large [Result] types (given the large size of the @@ -36,12 +43,12 @@ impl PowerSyncError { pub fn from_sqlite( db: *mut sqlite3, code: ResultCode, - context: impl Into>, + stmt: Option>, ) -> Self { RawPowerSyncError::Sqlite(SqliteError { code, errstr: Self::errstr(db), - context: Some(context.into()), + statement: stmt.map(|e| e.into()), }) .into() } @@ -108,6 +115,14 @@ impl PowerSyncError { return RawPowerSyncError::DownMigrationDidNotUpdateVersion { current_version }.into(); } + pub fn context(self, context: String) -> Self { + RawPowerSyncError::Context { + inner: self, + context, + } + .into() + } + /// Applies this error to a function result context, setting the error code and a descriptive /// text. pub fn apply_to_ctx(self, description: &str, ctx: *mut context) { @@ -132,6 +147,8 @@ impl PowerSyncError { | SqliteVersionMismatch { .. } => ResultCode::ABORT, LocalDataError { .. } => ResultCode::CORRUPT, Internal { .. } => ResultCode::INTERNAL, + CString { .. } => ResultCode::FORMAT, + Context { inner, context: _ } => inner.sqlite_error_code(), } } @@ -149,7 +166,7 @@ impl PowerSyncError { } } - pub fn check_sqlite3_version() -> Result<(), PowerSyncError> { + pub fn check_sqlite3_version() -> Result<()> { let actual_version = sqlite::libversion_number(); if actual_version < MIN_SQLITE_VERSION_NUMBER { @@ -180,14 +197,9 @@ impl From for PowerSyncError { } } -impl From for PowerSyncError { - fn from(value: ResultCode) -> Self { - return RawPowerSyncError::Sqlite(SqliteError { - code: value, - errstr: None, - context: None, - }) - .into(); +impl From for PowerSyncError { + fn from(value: NulError) -> Self { + RawPowerSyncError::CString { inner: value }.into() } } @@ -250,19 +262,47 @@ pub enum RawPowerSyncError { }, #[error("This function may only be called in transactions.")] MustBeCalledInTransaction, + #[error("Allocating c string: {inner}")] + CString { + #[from] + inner: NulError, + }, + #[error("{inner} (context: {context})")] + Context { + inner: PowerSyncError, + context: String, + }, } #[derive(Debug)] pub struct SqliteError { code: ResultCode, errstr: Option, - context: Option>, + statement: Option, +} + +#[derive(Debug)] +pub enum SqlText { + Rust(String), + C(CString), +} + +impl From<&str> for SqlText { + fn from(value: &str) -> Self { + Self::Rust(value.to_string()) + } +} + +impl From<&CStr> for SqlText { + fn from(value: &CStr) -> Self { + Self::C(value.to_owned()) + } } impl Display for SqliteError { fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result { - if let Some(context) = &self.context { - write!(f, "{}: ", context)?; + if let Some(stmt) = &self.statement { + write!(f, "statement {}: ", stmt)?; } write!(f, "internal SQLite call returned {}", self.code)?; @@ -274,20 +314,12 @@ impl Display for SqliteError { } } -pub trait PSResult { - fn into_db_result(self, db: *mut sqlite3) -> Result; -} - -impl PSResult for Result { - fn into_db_result(self, db: *mut sqlite3) -> Result { - self.map_err(|code| { - RawPowerSyncError::Sqlite(SqliteError { - code, - errstr: PowerSyncError::errstr(db), - context: None, - }) - .into() - }) +impl Display for SqlText { + fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result { + match self { + SqlText::Rust(s) => f.write_str(s), + SqlText::C(cstring) => f.write_str(&cstring.to_string_lossy()), + } } } diff --git a/crates/core/src/ext.rs b/crates/core/src/ext.rs deleted file mode 100644 index be759d79..00000000 --- a/crates/core/src/ext.rs +++ /dev/null @@ -1,38 +0,0 @@ -use powersync_sqlite_nostd::{Connection, Destructor, ManagedStmt, ResultCode, sqlite3}; - -pub trait SafeManagedStmt { - fn exec(&self) -> Result<(), ResultCode>; -} - -impl SafeManagedStmt for ManagedStmt { - fn exec(&self) -> Result<(), ResultCode> { - loop { - let rs = self.step()?; - if rs == ResultCode::ROW { - continue; - } - - self.reset()?; - if rs == ResultCode::DONE { - break; - } else { - return Err(rs); - } - } - Ok(()) - } -} - -pub trait ExtendedDatabase { - fn exec_text(&self, sql: &str, param: &str) -> Result<(), ResultCode>; -} - -impl ExtendedDatabase for *mut sqlite3 { - fn exec_text(&self, sql: &str, param: &str) -> Result<(), ResultCode> { - let statement = self.prepare_v2(sql)?; - statement.bind_text(1, param, Destructor::STATIC)?; - - statement.exec()?; - Ok(()) - } -} diff --git a/crates/core/src/fix_data.rs b/crates/core/src/fix_data.rs index 31c31b1d..73f15e16 100644 --- a/crates/core/src/fix_data.rs +++ b/crates/core/src/fix_data.rs @@ -4,14 +4,13 @@ use alloc::format; use alloc::string::String; use crate::create_sqlite_optional_text_fn; -use crate::error::{PSResult, PowerSyncError}; +use crate::error::{PowerSyncError, Result}; use crate::schema::inspection::ExistingTable; use crate::utils::SqlBuffer; +use crate::utils::database::Database; use powersync_sqlite_nostd::{self as sqlite, ColumnType, Value}; use powersync_sqlite_nostd::{Connection, Context, ResultCode}; -use crate::ext::SafeManagedStmt; - // Apply a data migration to fix any existing data affected by the issue // fixed in v0.3.5. // @@ -21,13 +20,13 @@ use crate::ext::SafeManagedStmt; // // The fix here is to find these dangling rows, and add them to ps_updated_rows. // The next time the sync_local operation is run, these rows will be removed. -pub fn apply_v035_fix(db: *mut sqlite::sqlite3) -> Result { +pub fn apply_v035_fix(db: Database) -> Result { // 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__*'", + )?; - while statement.step()? == ResultCode::ROW { + while statement.step()? { let full_name = statement.column_text(0)?; let Some((short_name, _)) = ExistingTable::external_name(full_name) else { continue; @@ -123,11 +122,15 @@ fn remove_duplicate_key_encoding(key: &str) -> Option { fn powersync_remove_duplicate_key_encoding_impl( _ctx: *mut sqlite::context, args: &[*mut sqlite::value], -) -> Result, PowerSyncError> { - let arg = args.get(0).ok_or(ResultCode::MISUSE)?; +) -> Result> { + fn unexpected_argument() -> PowerSyncError { + PowerSyncError::argument_error("Expected a text argument") + } + + let arg = args.get(0).ok_or_else(unexpected_argument)?; if arg.value_type() != ColumnType::Text { - return Err(ResultCode::MISMATCH.into()); + return Err(unexpected_argument()); } return Ok(remove_duplicate_key_encoding(arg.text())); @@ -139,7 +142,7 @@ create_sqlite_optional_text_fn!( "powersync_remove_duplicate_key_encoding" ); -pub fn register(db: *mut sqlite::sqlite3) -> Result<(), ResultCode> { +pub fn register(db: *mut sqlite::sqlite3) -> core::result::Result<(), ResultCode> { db.create_function_v2( "powersync_remove_duplicate_key_encoding", 1, diff --git a/crates/core/src/json_util.rs b/crates/core/src/json_util.rs index 734eb8f5..2d3fa049 100644 --- a/crates/core/src/json_util.rs +++ b/crates/core/src/json_util.rs @@ -5,7 +5,7 @@ use core::ffi::c_int; use crate::constants::SUBTYPE_JSON; use crate::create_sqlite_text_fn; -use crate::error::PowerSyncError; +use crate::error::{PowerSyncError, Result}; use powersync_sqlite_nostd as sqlite; use powersync_sqlite_nostd::bindings::{SQLITE_RESULT_SUBTYPE, SQLITE_SUBTYPE}; use powersync_sqlite_nostd::{Connection, Context, Value}; @@ -32,7 +32,7 @@ extern "C" fn powersync_strip_subtype( fn powersync_json_merge_impl( ctx: *mut sqlite::context, args: &[*mut sqlite::value], -) -> Result { +) -> Result { if args.is_empty() { return Ok("{}".to_string()); } @@ -67,7 +67,7 @@ create_sqlite_text_fn!( "powersync_json_merge" ); -pub fn register(db: *mut sqlite::sqlite3) -> Result<(), ResultCode> { +pub fn register(db: *mut sqlite::sqlite3) -> core::result::Result<(), ResultCode> { db.create_function_v2( "powersync_json_merge", -1, diff --git a/crates/core/src/kv.rs b/crates/core/src/kv.rs index c926c466..81226bdb 100644 --- a/crates/core/src/kv.rs +++ b/crates/core/src/kv.rs @@ -8,22 +8,23 @@ use powersync_sqlite_nostd::{Connection, Context}; use sqlite::ResultCode; use crate::create_sqlite_text_fn; -use crate::error::PowerSyncError; +use crate::error::{PowerSyncError, Result}; +use crate::utils::database::Database; fn powersync_client_id_impl( ctx: *mut sqlite::context, _args: &[*mut sqlite::value], -) -> Result { +) -> Result { let db = ctx.db_handle(); - client_id(db) + client_id(db.into()) } -pub fn client_id(db: *mut sqlite::sqlite3) -> Result { +pub fn client_id(db: Database) -> Result { // language=SQLite let statement = db.prepare_v2("select value from ps_kv where key = 'client_id'")?; - if statement.step()? == ResultCode::ROW { + if statement.step()? { let client_id = statement.column_text(0)?; Ok(client_id.to_string()) } else { @@ -37,7 +38,7 @@ create_sqlite_text_fn!( "powersync_client_id" ); -pub fn register(db: *mut sqlite::sqlite3) -> Result<(), ResultCode> { +pub fn register(db: *mut sqlite::sqlite3) -> core::result::Result<(), ResultCode> { db.create_function_v2( "powersync_client_id", 0, diff --git a/crates/core/src/lib.rs b/crates/core/src/lib.rs index 655f3356..b12f2249 100644 --- a/crates/core/src/lib.rs +++ b/crates/core/src/lib.rs @@ -7,14 +7,17 @@ use alloc::{ffi::CString, format, rc::Rc}; use powersync_sqlite_nostd as sqlite; use sqlite::ResultCode; -use crate::{error::PowerSyncError, state::DatabaseState}; +use crate::{ + error::{PowerSyncError, Result}, + state::DatabaseState, + utils::database::Database, +}; mod bson; mod constants; mod crud_vtab; mod diff; mod error; -mod ext; mod fix_data; mod json_util; mod kv; @@ -45,7 +48,7 @@ pub extern "C" fn sqlite3_powersync_init( debug_assert!(unsafe { *err_msg }.is_null()); sqlite::EXTENSION_INIT2(api); - let result = init_extension(db); + let result = init_extension(Database::from(db)); return if let Err(code) = result { if let Ok(desc) = CString::new(format!("Could not initialize PowerSync: {}", code)) { @@ -59,27 +62,29 @@ pub extern "C" fn sqlite3_powersync_init( }; } -fn init_extension(db: *mut sqlite::sqlite3) -> Result<(), PowerSyncError> { +fn init_extension(db: Database) -> Result<()> { PowerSyncError::check_sqlite3_version()?; let state = Rc::new(DatabaseState::new()); - crate::version::register(db)?; - crate::uuid::register(db)?; - crate::diff::register(db)?; - crate::fix_data::register(db)?; - crate::json_util::register(db)?; - crate::view_admin::register(db, state.clone())?; - crate::kv::register(db)?; - crate::state::register(db, state.clone())?; - sync::register(db, state.clone())?; - update_hooks::register(db, state.clone())?; + db.use_inner(|db| { + crate::version::register(db)?; + crate::uuid::register(db)?; + crate::diff::register(db)?; + crate::fix_data::register(db)?; + crate::json_util::register(db)?; + crate::view_admin::register(db, state.clone())?; + crate::kv::register(db)?; + crate::state::register(db, state.clone())?; + sync::register(db, state.clone())?; + update_hooks::register(db, state.clone())?; - crate::schema::register(db, state.clone())?; - crate::pre_close_vtab::register(db, state.clone())?; - crate::crud_vtab::register(db, state)?; + crate::schema::register(db, state.clone())?; + crate::pre_close_vtab::register(db, state.clone())?; + crate::crud_vtab::register(db, state)?; - Ok(()) + Ok(()) + }) } unsafe extern "C" { diff --git a/crates/core/src/macros.rs b/crates/core/src/macros.rs index 4e60c953..41495210 100644 --- a/crates/core/src/macros.rs +++ b/crates/core/src/macros.rs @@ -11,7 +11,7 @@ macro_rules! create_sqlite_text_fn { let result = $fn_impl_name(ctx, args); if let Err(err) = result { - PowerSyncError::from(err).apply_to_ctx($description, ctx); + err.apply_to_ctx($description, ctx); } else if let Ok(r) = result { ctx.result_text_transient(&r); } @@ -32,7 +32,7 @@ macro_rules! create_sqlite_optional_text_fn { let result = $fn_impl_name(ctx, args); if let Err(err) = result { - PowerSyncError::from(err).apply_to_ctx($description, ctx); + err.apply_to_ctx($description, ctx); } else if let Ok(r) = result { if let Some(s) = r { ctx.result_text_transient(&s); diff --git a/crates/core/src/migrations.rs b/crates/core/src/migrations.rs index 87c56cb9..3a8a58e1 100644 --- a/crates/core/src/migrations.rs +++ b/crates/core/src/migrations.rs @@ -4,38 +4,33 @@ use alloc::format; use alloc::string::{String, ToString}; use alloc::vec::Vec; +use powersync_sqlite_nostd::Context; use powersync_sqlite_nostd::{self as sqlite, Destructor}; -use powersync_sqlite_nostd::{Connection, Context}; use serde::Serialize; use serde_json::json; use sqlite::ResultCode; -use crate::error::{PSResult, PowerSyncError}; -use crate::ext::SafeManagedStmt; +use crate::error::{PowerSyncError, Result}; use crate::fix_data::apply_v035_fix; use crate::schema::inspection::ExistingView; use crate::sync::BucketPriority; +use crate::utils::database::Database; pub const LATEST_VERSION: i32 = 14; -pub fn powersync_migrate( - ctx: *mut sqlite::context, - target_version: i32, -) -> Result<(), PowerSyncError> { - let local_db = ctx.db_handle(); +pub fn powersync_migrate(ctx: *mut sqlite::context, target_version: i32) -> Result<()> { + let local_db = Database::from(ctx.db_handle()); // language=SQLite local_db.exec_safe( - "\ + c"\ CREATE TABLE IF NOT EXISTS ps_migration(id INTEGER PRIMARY KEY, down_migrations TEXT)", )?; // language=SQLite - let current_version_stmt = local_db - .prepare_v2("SELECT ifnull(max(id), 0) as version FROM ps_migration") - .into_db_result(local_db)?; - let rc = current_version_stmt.step()?; - if rc != ResultCode::ROW { + let current_version_stmt = + local_db.prepare_v2("SELECT ifnull(max(id), 0) as version FROM ps_migration")?; + if !current_version_stmt.step()? { return Err(PowerSyncError::unknown_internal()); } @@ -52,38 +47,23 @@ CREATE TABLE IF NOT EXISTS ps_migration(id INTEGER PRIMARY KEY, down_migrations let mut down_sql: Vec = alloc::vec![]; - while down_migrations_stmt.step()? == ResultCode::ROW { + while down_migrations_stmt.step()? { let sql = down_migrations_stmt.column_text(0)?; down_sql.push(sql.to_string()); } for sql in down_sql { - let rs = local_db.exec_safe(&sql); - if let Err(code) = rs { - return Err(PowerSyncError::from_sqlite( - local_db, - code, - format!( - "Down migration failed for {:} {:} {:}", - current_version, - sql, - local_db - .errmsg() - .unwrap_or(String::from("Conversion error")) - ), - )); - } + local_db + .exec_safe_str(&sql) + .map_err(|e| e.context(format!("Down migration failed from {current_version}")))?; } // Refresh the version current_version_stmt.reset()?; - let rc = current_version_stmt.step()?; - if rc != ResultCode::ROW { - return Err(PowerSyncError::from_sqlite( - local_db, - rc, - "Down migration failed - could not get version", - )); + if !current_version_stmt.step()? { + return Err(current_version_stmt + .map_error(ResultCode::DONE) + .context("Down migration failed - could not get version".to_string())); } let new_version = current_version_stmt.column_int(0); if new_version >= current_version { @@ -98,9 +78,8 @@ CREATE TABLE IF NOT EXISTS ps_migration(id INTEGER PRIMARY KEY, down_migrations if current_version < 1 { // language=SQLite - local_db - .exec_safe( - " + local_db.exec_safe( + c" CREATE TABLE ps_oplog( bucket TEXT NOT NULL, op_id INTEGER NOT NULL, @@ -131,35 +110,34 @@ CREATE TABLE ps_crud (id INTEGER PRIMARY KEY AUTOINCREMENT, data TEXT); INSERT INTO ps_migration(id, down_migrations) VALUES(1, NULL); ", - ) - .into_db_result(local_db)?; + )?; } if current_version < 2 && target_version >= 2 { // language=SQLite - local_db.exec_safe("\ + local_db.exec_safe(c"\ CREATE TABLE ps_tx(id INTEGER PRIMARY KEY NOT NULL, current_tx INTEGER, next_tx INTEGER); INSERT INTO ps_tx(id, current_tx, next_tx) VALUES(1, NULL, 1); ALTER TABLE ps_crud ADD COLUMN tx_id INTEGER; INSERT INTO ps_migration(id, down_migrations) VALUES(2, json_array(json_object('sql', 'DELETE FROM ps_migration WHERE id >= 2', 'params', json_array()), json_object('sql', 'DROP TABLE ps_tx', 'params', json_array()), json_object('sql', 'ALTER TABLE ps_crud DROP COLUMN tx_id', 'params', json_array()))); -").into_db_result(local_db)?; +")?; } if current_version < 3 && target_version >= 3 { // language=SQLite - local_db.exec_safe("\ + local_db.exec_safe(c"\ CREATE TABLE ps_kv(key TEXT PRIMARY KEY NOT NULL, value BLOB); INSERT INTO ps_kv(key, value) values('client_id', uuid()); INSERT INTO ps_migration(id, down_migrations) VALUES(3, json_array(json_object('sql', 'DELETE FROM ps_migration WHERE id >= 3'), json_object('sql', 'DROP TABLE ps_kv'))); - ").into_db_result(local_db)?; + ")?; } if current_version < 4 && target_version >= 4 { // language=SQLite - local_db.exec_safe("\ + local_db.exec_safe(c"\ ALTER TABLE ps_buckets ADD COLUMN op_checksum INTEGER NOT NULL DEFAULT 0; ALTER TABLE ps_buckets ADD COLUMN remove_operations INTEGER NOT NULL DEFAULT 0; @@ -174,7 +152,7 @@ VALUES(4, json_object('sql', 'ALTER TABLE ps_buckets DROP COLUMN op_checksum'), json_object('sql', 'ALTER TABLE ps_buckets DROP COLUMN remove_operations') )); - ").into_db_result(local_db)?; + ")?; } if current_version < 5 && target_version >= 5 { @@ -204,7 +182,7 @@ VALUES(4, // language=SQLite local_db .exec_safe( - "\ + c"\ ALTER TABLE ps_buckets RENAME TO ps_buckets_old; ALTER TABLE ps_oplog RENAME TO ps_oplog_old; @@ -297,8 +275,7 @@ VALUES(5, json_object('sql', 'DELETE FROM ps_migration WHERE id >= 5') )); ", - ) - .into_db_result(local_db)?; + )?; } if current_version < 6 && target_version >= 6 { @@ -307,17 +284,15 @@ VALUES(5, apply_v035_fix(local_db)?; } - local_db - .exec_safe( - "\ + local_db.exec_safe( + c"\ INSERT INTO ps_migration(id, down_migrations) VALUES(6, json_array( json_object('sql', 'DELETE FROM ps_migration WHERE id >= 6') )); ", - ) - .into_db_result(local_db)?; + )?; } if current_version < 7 && target_version >= 7 { @@ -339,11 +314,11 @@ json_object('sql', 'DELETE FROM ps_migration WHERE id >= 7') )); ", SENTINEL_PRIORITY, SENTINEL_PRIORITY); - local_db.exec_safe(&stmt).into_db_result(local_db)?; + local_db.exec_safe_str(&stmt)?; } if current_version < 8 && target_version >= 8 { - let stmt = "\ + let stmt = c"\ ALTER TABLE ps_sync_state RENAME TO ps_sync_state_old; CREATE TABLE ps_sync_state ( priority INTEGER NOT NULL PRIMARY KEY, @@ -360,11 +335,11 @@ json_object('sql', 'DROP TABLE ps_sync_state_new'), json_object('sql', 'DELETE FROM ps_migration WHERE id >= 8') )); "; - local_db.exec_safe(&stmt).into_db_result(local_db)?; + local_db.exec_safe(&stmt)?; } if current_version < 9 && target_version >= 9 { - let stmt = "\ + let stmt = c"\ ALTER TABLE ps_buckets ADD COLUMN count_at_last INTEGER NOT NULL DEFAULT 0; ALTER TABLE ps_buckets ADD COLUMN count_since_last INTEGER NOT NULL DEFAULT 0; INSERT INTO ps_migration(id, down_migrations) VALUES(9, json_array( @@ -374,7 +349,7 @@ json_object('sql', 'DELETE FROM ps_migration WHERE id >= 9') )); "; - local_db.exec_safe(stmt).into_db_result(local_db)?; + local_db.exec_safe(stmt)?; } if current_version < 10 && target_version >= 10 { @@ -383,18 +358,17 @@ json_object('sql', 'DELETE FROM ps_migration WHERE id >= 9') // by applying the PowerSync user schema after these internal migrations finish. local_db .exec_safe( - "\ + c"\ INSERT INTO ps_migration(id, down_migrations) VALUES (10, json_array( json_object('sql', 'SELECT powersync_drop_view(view.name)\n FROM sqlite_master view\n WHERE view.type = ''view''\n AND view.sql GLOB ''*-- powersync-auto-generated'''), json_object('sql', 'DELETE FROM ps_migration WHERE id >= 10') )); ", - ) - .into_db_result(local_db)?; + )?; } if current_version < 11 && target_version >= 11 { - let stmt = "\ + let stmt = c"\ CREATE TABLE ps_stream_subscriptions ( id INTEGER NOT NULL PRIMARY KEY, stream_name TEXT NOT NULL, @@ -413,22 +387,22 @@ json_object('sql', 'DROP TABLE ps_stream_subscriptions'), json_object('sql', 'DELETE FROM ps_migration WHERE id >= 11') )); "; - local_db.exec_safe(stmt).into_db_result(local_db)?; + local_db.exec_safe(stmt)?; } if current_version < 12 && target_version >= 12 { - let stmt = "\ + let stmt = c"\ ALTER TABLE ps_buckets ADD COLUMN downloaded_size INTEGER NOT NULL DEFAULT 0; INSERT INTO ps_migration(id, down_migrations) VALUES(12, json_array( json_object('sql', 'ALTER TABLE ps_buckets DROP COLUMN downloaded_size'), json_object('sql', 'DELETE FROM ps_migration WHERE id >= 12') )); "; - local_db.exec_safe(stmt).into_db_result(local_db)?; + local_db.exec_safe(stmt)?; } if current_version < 13 && target_version >= 13 { - let up = "\ + let up = c"\ UPDATE ps_stream_subscriptions SET expires_at = expires_at * 1000000, last_synced_at = last_synced_at * 1000000; ALTER TABLE ps_sync_state RENAME TO ps_sync_state_old; CREATE TABLE ps_sync_state ( @@ -439,7 +413,7 @@ INSERT INTO ps_sync_state (priority, last_synced_at) SELECT priority, unixepoch(last_synced_at) * 1000000 FROM ps_sync_state_old; DROP TABLE ps_sync_state_old; "; - local_db.exec_safe(up).into_db_result(local_db)?; + local_db.exec_safe(up)?; const DOWN_STATEMENTS: &[&str] = &[ "UPDATE ps_stream_subscriptions SET expires_at = expires_at / 1000000, last_synced_at = last_synced_at / 1000000", @@ -491,7 +465,7 @@ DROP TABLE ps_sync_state_old; // // DROP COLUMN requires SQLite 3.35+; the extension already refuses to load below // MIN_SQLITE_VERSION_NUMBER (3.44), so this is safe in the up path. - let up = "\ + let up = c"\ DELETE FROM ps_kv WHERE key IN ( 'last_applied_checkpoint_request_id', @@ -521,7 +495,7 @@ DELETE FROM ps_buckets WHERE name = '$local'; ALTER TABLE ps_buckets DROP COLUMN target_op; "; - local_db.exec_safe(up).into_db_result(local_db)?; + local_db.exec_safe(up)?; // Downgrading needs to rebuild the old `$local` row from the new ps_kv state so older SDKs // can keep using their target-op based blocking behavior. In that model, `$local.last_op` @@ -605,11 +579,11 @@ ON CONFLICT(name) DO UPDATE SET Ok(()) } -fn serialize_down_statements(statements: &[&'static str]) -> Result { +fn serialize_down_statements(statements: &[&'static str]) -> Result { struct DownStatements<'a>(&'a [&'static str]); impl<'a> Serialize for DownStatements<'a> { - fn serialize(&self, serializer: S) -> Result + fn serialize(&self, serializer: S) -> core::result::Result where S: serde::Serializer, { diff --git a/crates/core/src/pre_close_vtab.rs b/crates/core/src/pre_close_vtab.rs index 79dfa4ef..e37873ed 100644 --- a/crates/core/src/pre_close_vtab.rs +++ b/crates/core/src/pre_close_vtab.rs @@ -7,8 +7,10 @@ use core::ffi::{c_char, c_int, c_void}; use powersync_sqlite_nostd as sqlite; use sqlite::{Connection, ResultCode}; +use crate::error::Result; use crate::state::DatabaseState; use crate::update_hooks::uninstall_update_hooks; +use crate::utils::database::Database; use crate::vtab_util::*; /// A virtual table hack to implement a "pre-close hook" for databases. @@ -140,12 +142,15 @@ static MODULE: sqlite::module = sqlite::module { xIntegrity: None, }; -pub fn ensure_has_internal_close_vtab(db: *mut sqlite::sqlite3) -> Result<(), ResultCode> { - db.exec(c"SELECT 1 FROM powersync_internal_close;")?; +pub fn ensure_has_internal_close_vtab(db: Database) -> Result<()> { + db.exec_safe(c"SELECT 1 FROM powersync_internal_close;")?; Ok(()) } -pub fn register(db: *mut sqlite::sqlite3, state: Rc) -> Result<(), ResultCode> { +pub fn register( + db: *mut sqlite::sqlite3, + state: Rc, +) -> core::result::Result<(), ResultCode> { db.create_module_v2( "powersync_internal_close", &MODULE, diff --git a/crates/core/src/schema/inspection.rs b/crates/core/src/schema/inspection.rs index 34b1aaa6..09b72f72 100644 --- a/crates/core/src/schema/inspection.rs +++ b/crates/core/src/schema/inspection.rs @@ -1,11 +1,10 @@ use alloc::borrow::ToOwned; use alloc::{format, vec}; use alloc::{string::String, vec::Vec}; -use powersync_sqlite_nostd::Connection; -use powersync_sqlite_nostd::{self as sqlite, ResultCode}; -use crate::error::{PSResult, PowerSyncError}; +use crate::error::Result; use crate::utils::SqlBuffer; +use crate::utils::database::Database; /// An existing PowerSync-managed view that was found in the schema. #[derive(PartialEq)] @@ -24,7 +23,7 @@ pub struct ExistingView { } impl ExistingView { - pub fn list(db: *mut sqlite::sqlite3) -> Result, PowerSyncError> { + pub fn list(db: Database) -> Result> { let mut results = vec![]; let stmt = db.prepare_v2(" SELECT @@ -42,9 +41,9 @@ SELECT ON trigger3.tbl_name = view.name AND trigger3.type = 'trigger' AND trigger3.name GLOB 'ps_view_update*' WHERE view.type = 'view' AND view.sql GLOB '*-- powersync-auto-generated' GROUP BY view.name; - ").into_db_result(db)?; + ")?; - while stmt.step()? == ResultCode::ROW { + while stmt.step()? { let name = stmt.column_text(0)?.to_owned(); let sql = stmt.column_text(1)?.to_owned(); let delete = stmt.column_text(2)?.to_owned(); @@ -63,18 +62,18 @@ SELECT Ok(results) } - pub fn drop_by_name(db: *mut sqlite::sqlite3, name: &str) -> Result<(), PowerSyncError> { + pub fn drop_by_name(db: Database, name: &str) -> Result<()> { let q = format!("DROP VIEW IF EXISTS {:}", SqlBuffer::quote_identifier(name)); - db.exec_safe(&q)?; + db.exec_safe_str(&q)?; Ok(()) } - pub fn create(&self, db: *mut sqlite::sqlite3) -> Result<(), PowerSyncError> { + pub fn create(&self, db: Database) -> Result<()> { Self::drop_by_name(db, &self.name)?; - db.exec_safe(&self.sql).into_db_result(db)?; - db.exec_safe(&self.delete_trigger_sql).into_db_result(db)?; - db.exec_safe(&self.insert_trigger_sql).into_db_result(db)?; - db.exec_safe(&self.update_trigger_sql).into_db_result(db)?; + db.exec_safe_str(&self.sql)?; + db.exec_safe_str(&self.delete_trigger_sql)?; + db.exec_safe_str(&self.insert_trigger_sql)?; + db.exec_safe_str(&self.update_trigger_sql)?; Ok(()) } @@ -87,17 +86,15 @@ pub struct ExistingTable { } impl ExistingTable { - pub fn list(db: *mut sqlite::sqlite3) -> Result, PowerSyncError> { + pub fn list(db: Database) -> Result> { let mut results = vec![]; - let stmt = db - .prepare_v2( - " + let stmt = db.prepare_v2( + " SELECT name FROM sqlite_master WHERE type = 'table' AND name GLOB 'ps_data_*'; ", - ) - .into_db_result(db)?; + )?; - while stmt.step()? == ResultCode::ROW { + while stmt.step()? { let internal_name = stmt.column_text(0)?; let Some((name, local_only)) = Self::external_name(internal_name) else { continue; diff --git a/crates/core/src/schema/management.rs b/crates/core/src/schema/management.rs index 8b6007b3..90f5d8cf 100644 --- a/crates/core/src/schema/management.rs +++ b/crates/core/src/schema/management.rs @@ -14,11 +14,11 @@ use powersync_sqlite_nostd::Context; use sqlite::{Connection, ResultCode, Value}; use crate::create_sqlite_text_fn; -use crate::error::{PSResult, PowerSyncError}; -use crate::ext::ExtendedDatabase; +use crate::error::{PowerSyncError, Result}; use crate::schema::inspection::{ExistingTable, ExistingView}; use crate::schema::table_info::Index; use crate::state::DatabaseState; +use crate::utils::database::Database; use crate::utils::{SqlBuffer, verify_in_transaction}; use crate::views::{ powersync_trigger_delete_sql, powersync_trigger_insert_sql, powersync_trigger_update_sql, @@ -27,7 +27,7 @@ use crate::views::{ use super::Schema; -fn update_tables(db: *mut sqlite::sqlite3, schema: &Schema) -> Result<(), PowerSyncError> { +fn update_tables(db: Database, schema: &Schema) -> Result<()> { let existing_tables = ExistingTable::list(db)?; let mut existing_tables = { let mut map = BTreeMap::new(); @@ -56,11 +56,10 @@ fn update_tables(db: *mut sqlite::sqlite3, schema: &Schema) -> Result<(), PowerS // New table. let quoted_internal_name = SqlBuffer::quote_identifier(&table.internal_name()); - db.exec_safe(&format!( + db.exec_safe_str(&format!( "CREATE TABLE {:}(id TEXT PRIMARY KEY NOT NULL, data TEXT)", quoted_internal_name - )) - .into_db_result(db)?; + ))?; if !table.local_only() { // MOVE data if any @@ -73,8 +72,7 @@ fn update_tables(db: *mut sqlite::sqlite3, schema: &Schema) -> Result<(), PowerS quoted_internal_name ), &table.name, - ) - .into_db_result(db)?; + )?; // language=SQLite db.exec_text("DELETE FROM ps_untyped WHERE type = ?", &table.name)?; @@ -91,8 +89,7 @@ fn update_tables(db: *mut sqlite::sqlite3, schema: &Schema) -> Result<(), PowerS SqlBuffer::quote_identifier(&remaining.internal_name) ), &remaining.name, - ) - .into_db_result(db)?; + )?; } } @@ -103,7 +100,7 @@ fn update_tables(db: *mut sqlite::sqlite3, schema: &Schema) -> Result<(), PowerS "DROP TABLE {:}", SqlBuffer::quote_identifier(&remaining.internal_name) ); - db.exec_safe(&q).into_db_result(db)?; + db.exec_safe_str(&q)?; } Ok(()) @@ -132,7 +129,7 @@ fn create_index_stmt(table_name: &str, index_name: &str, index: &Index) -> Strin sql.sql } -fn update_indexes(db: *mut sqlite::sqlite3, schema: &Schema) -> Result<(), PowerSyncError> { +fn update_indexes(db: Database, schema: &Schema) -> Result<()> { let mut statements: Vec = alloc::vec![]; let mut expected_index_names: Vec = vec![]; @@ -152,7 +149,7 @@ fn update_indexes(db: *mut sqlite::sqlite3, schema: &Schema) -> Result<(), Power find_index.reset()?; find_index.bind_text(1, &index_name, sqlite::Destructor::STATIC)?; - let result = if let ResultCode::ROW = find_index.step()? { + let result = if find_index.step()? { Some(find_index.column_text(0)?) } else { None @@ -178,9 +175,8 @@ fn update_indexes(db: *mut sqlite::sqlite3, schema: &Schema) -> Result<(), Power // In a block so that the statement is finalized before dropping indexes // language=SQLite - let statement = db - .prepare_v2( - "\ + let statement = db.prepare_v2( + "\ SELECT sqlite_master.name as index_name FROM sqlite_master @@ -188,13 +184,12 @@ SELECT AND sqlite_master.name GLOB 'ps_data_*' AND sqlite_master.name NOT IN (SELECT value FROM json_each(?)) ", - ) - .into_db_result(db)?; + )?; let json_names = serde_json::to_string(&expected_index_names) .map_err(PowerSyncError::as_argument_error)?; statement.bind_text(1, &json_names, sqlite::Destructor::STATIC)?; - while statement.step()? == ResultCode::ROW { + while statement.step()? { let name = statement.column_text(0)?; statements.push(format!("DROP INDEX {}", SqlBuffer::quote_identifier(name))); @@ -203,14 +198,14 @@ SELECT // We cannot have any open queries on sqlite_master at the point that we drop indexes, otherwise // we get "database table is locked (code 6)" errors. - for statement in statements { - db.exec_safe(&statement).into_db_result(db)?; + for statement in &statements { + db.exec_safe_str(statement)?; } Ok(()) } -fn update_views(db: *mut sqlite::sqlite3, schema: &Schema) -> Result<(), PowerSyncError> { +fn update_views(db: Database, schema: &Schema) -> Result<()> { // First, find all existing views and index them by name. let existing = ExistingView::list(db)?; let mut existing = { @@ -261,8 +256,8 @@ fn update_views(db: *mut sqlite::sqlite3, schema: &Schema) -> Result<(), PowerSy fn powersync_replace_schema_impl( ctx: *mut sqlite::context, args: &[*mut sqlite::value], -) -> Result { - let db = ctx.db_handle(); +) -> Result { + let db = Database::from(ctx.db_handle()); verify_in_transaction(db)?; let schema = args[0].text(); @@ -271,7 +266,7 @@ fn powersync_replace_schema_impl( serde_json::from_str::(schema).map_err(PowerSyncError::as_argument_error)?; // language=SQLite - db.exec_safe("SELECT powersync_init()").into_db_result(db)?; + db.exec_safe(c"SELECT powersync_init()")?; update_tables(db, &parsed_schema)?; update_indexes(db, &parsed_schema)?; @@ -287,7 +282,10 @@ create_sqlite_text_fn!( "powersync_replace_schema" ); -pub fn register(db: *mut sqlite::sqlite3, state: Rc) -> Result<(), ResultCode> { +pub fn register( + db: *mut sqlite::sqlite3, + state: Rc, +) -> core::result::Result<(), ResultCode> { db.create_function_v2( "powersync_replace_schema", 1, diff --git a/crates/core/src/schema/mod.rs b/crates/core/src/schema/mod.rs index 50e8e5c9..5451606e 100644 --- a/crates/core/src/schema/mod.rs +++ b/crates/core/src/schema/mod.rs @@ -16,10 +16,10 @@ pub use table_info::{ }; use crate::{ - error::{PSResult, PowerSyncError}, + error::PowerSyncError, schema::raw_table::generate_raw_table_trigger, state::DatabaseState, - utils::WriteType, + utils::{WriteType, database::Database}, }; #[derive(Deserialize, Default)] @@ -43,10 +43,10 @@ pub fn register(db: *mut sqlite::sqlite3, state: Rc) -> Result<() let trigger_name = args[1].text(); let write_type: WriteType = args[2].text().parse()?; - let db = context.db_handle(); + let db = Database::from(context.db_handle()); let create_trigger_stmt = generate_raw_table_trigger(db, &table, trigger_name, write_type)?; - db.exec_safe(&create_trigger_stmt).into_db_result(db)?; + db.exec_safe_str(&create_trigger_stmt)?; Ok(()) } diff --git a/crates/core/src/schema/raw_table.rs b/crates/core/src/schema/raw_table.rs index 23144ca7..ad146aad 100644 --- a/crates/core/src/schema/raw_table.rs +++ b/crates/core/src/schema/raw_table.rs @@ -11,12 +11,12 @@ use alloc::{ vec, vec::Vec, }; -use powersync_sqlite_nostd::{self as sqlite, Connection, Destructor, ResultCode}; +use powersync_sqlite_nostd::Destructor; use crate::{ - error::PowerSyncError, + error::{PowerSyncError, Result}, schema::{ColumnFilter, PendingStatement, PendingStatementValue, RawTable, SchemaTable}, - utils::{InsertIntoCrud, SqlBuffer, WriteType}, + utils::{InsertIntoCrud, SqlBuffer, WriteType, database::Database}, views::table_columns_to_json_object, }; @@ -28,16 +28,16 @@ pub struct InferredTableStructure { impl InferredTableStructure { pub fn read_from_database( table_name: &str, - db: impl Connection, + db: Database, synced_columns: &Option, - ) -> Result { + ) -> Result { let stmt = db.prepare_v2("select name from pragma_table_info(?)")?; stmt.bind_text(1, table_name, Destructor::STATIC)?; let mut has_id_column = false; let mut columns = vec![]; - while let ResultCode::ROW = stmt.step()? { + while stmt.step()? { let name = stmt.column_text(0)?; if name == "id" { has_id_column = true; @@ -128,7 +128,7 @@ pub struct InferredSchemaCache { } impl InferredSchemaCache { - pub fn current_schema_version(db: *mut sqlite::sqlite3) -> Result { + pub fn current_schema_version(db: Database) -> Result { let version = db.prepare_v2("PRAGMA schema_version")?; version.step()?; let version = version.column_int64(0) as usize; @@ -137,29 +137,29 @@ impl InferredSchemaCache { pub fn infer_put_statement( &self, - db: *mut sqlite::sqlite3, + db: Database, schema_version: usize, tbl: &RawTable, - ) -> Result, PowerSyncError> { + ) -> Result> { self.with_entry(db, schema_version, tbl, SchemaCacheEntry::put) } pub fn infer_delete_statement( &self, - db: *mut sqlite::sqlite3, + db: Database, schema_version: usize, tbl: &RawTable, - ) -> Result, PowerSyncError> { + ) -> Result> { self.with_entry(db, schema_version, tbl, SchemaCacheEntry::delete) } fn with_entry( &self, - db: *mut sqlite::sqlite3, + db: Database, schema_version: usize, tbl: &RawTable, f: impl FnOnce(&mut SchemaCacheEntry) -> Rc, - ) -> Result, PowerSyncError> { + ) -> Result> { let mut entries = self.entries.borrow_mut(); if let Some(value) = entries.get_mut(&tbl.name) { if value.schema_version != schema_version { @@ -185,11 +185,7 @@ pub struct SchemaCacheEntry { } impl SchemaCacheEntry { - fn infer( - db: *mut sqlite::sqlite3, - schema_version: usize, - table: &RawTable, - ) -> Result { + fn infer(db: Database, schema_version: usize, table: &RawTable) -> Result { let local_table_name = table.require_table_name()?; let structure = InferredTableStructure::read_from_database( local_table_name, @@ -221,11 +217,11 @@ impl SchemaCacheEntry { /// Generates a `CREATE TRIGGER` statement to capture writes on raw tables and to forward them to /// ps-crud. pub fn generate_raw_table_trigger( - db: impl Connection, + db: Database, table: &RawTable, trigger_name: &str, write: WriteType, -) -> Result { +) -> Result { let local_table_name = table.require_table_name()?; let synced_columns = &table.schema.synced_columns; let resolved_table = diff --git a/crates/core/src/state.rs b/crates/core/src/state.rs index f9c2fa00..27b11070 100644 --- a/crates/core/src/state.rs +++ b/crates/core/src/state.rs @@ -12,9 +12,10 @@ use powersync_sqlite_nostd::{self as sqlite, Context}; use sqlite::{Connection, ResultCode}; use crate::{ - error::PowerSyncError, + error::Result, schema::{InferredSchemaCache, Schema}, sync::{SyncClient, storage_adapter::StorageAdapter}, + utils::database::Database, }; /// State that is shared for a SQLite database connection after the core extension has been @@ -101,10 +102,7 @@ impl DatabaseState { core::mem::replace(&mut *committed, Default::default()) } - pub fn storage_adapter( - &self, - db: *mut sqlite::sqlite3, - ) -> Result, PowerSyncError> { + pub fn storage_adapter(&self, db: Database) -> Result> { let mut adapter = self.storage_adapter.borrow_mut(); Ok(match *adapter { Some(ref adapter) => { @@ -161,7 +159,10 @@ impl DatabaseState { } } -pub fn register(db: *mut sqlite::sqlite3, state: Rc) -> Result<(), ResultCode> { +pub fn register( + db: *mut sqlite::sqlite3, + state: Rc, +) -> core::result::Result<(), ResultCode> { unsafe extern "C" fn func( ctx: *mut sqlite::context, _argc: c_int, diff --git a/crates/core/src/sync/checkpoint.rs b/crates/core/src/sync/checkpoint.rs index 641aca1c..dceaf038 100644 --- a/crates/core/src/sync/checkpoint.rs +++ b/crates/core/src/sync/checkpoint.rs @@ -1,9 +1,11 @@ use alloc::{rc::Rc, string::String, vec::Vec}; use num_traits::Zero; +use crate::error::Result; use crate::sync::line::{BucketChecksum, BucketSubscriptionReason}; use crate::sync::{BucketPriority, Checksum}; -use powersync_sqlite_nostd::{self as sqlite, Connection, ResultCode}; +use crate::utils::database::Database; +use powersync_sqlite_nostd::{self as sqlite}; /// A structure cloned from [BucketChecksum]s with an owned bucket name instead of one borrowed from /// a sync line. @@ -47,8 +49,8 @@ pub struct ChecksumMismatch { pub fn validate_checkpoint<'a>( buckets: impl Iterator, priority: Option, - db: *mut sqlite::sqlite3, -) -> Result, ResultCode> { + db: Database, +) -> Result> { // language=SQLite let statement = db.prepare_v2( " @@ -63,13 +65,12 @@ FROM ps_buckets WHERE name = ?;", if bucket.is_in_priority(priority) { statement.bind_text(1, &bucket.bucket, sqlite::Destructor::STATIC)?; - let (add_checksum, oplog_checksum) = match statement.step()? { - ResultCode::ROW => { - let add_checksum = Checksum::from_i32(statement.column_int(0)); - let oplog_checksum = Checksum::from_i32(statement.column_int(1)); - (add_checksum, oplog_checksum) - } - _ => (Checksum::zero(), Checksum::zero()), + let (add_checksum, oplog_checksum) = if statement.step()? { + let add_checksum = Checksum::from_i32(statement.column_int(0)); + let oplog_checksum = Checksum::from_i32(statement.column_int(1)); + (add_checksum, oplog_checksum) + } else { + (Checksum::zero(), Checksum::zero()) }; let actual = add_checksum + oplog_checksum; diff --git a/crates/core/src/sync/interface.rs b/crates/core/src/sync/interface.rs index 5bd0cb7c..8293dee1 100644 --- a/crates/core/src/sync/interface.rs +++ b/crates/core/src/sync/interface.rs @@ -6,11 +6,12 @@ use super::streaming_sync::SyncClient; use super::sync_status::DownloadSyncStatus; use crate::constants::SUBTYPE_JSON; use crate::create_sqlite_text_fn; -use crate::error::PowerSyncError; +use crate::error::{PowerSyncError, Result}; use crate::schema::Schema; use crate::state::DatabaseState; use crate::sync::diagnostics::{DiagnosticOptions, DiagnosticsEvent}; use crate::sync::subscriptions::{StreamKey, apply_subscriptions}; +use crate::utils::database::Database; use alloc::borrow::Cow; use alloc::boxed::Box; use alloc::format; @@ -244,14 +245,17 @@ pub struct BucketRequest { pub after: String, } -pub fn register(db: *mut sqlite::sqlite3, state: Rc) -> Result<(), ResultCode> { +pub fn register( + db: *mut sqlite::sqlite3, + state: Rc, +) -> core::result::Result<(), ResultCode> { extern "C" fn control( ctx: *mut sqlite::context, argc: c_int, argv: *mut *mut sqlite::value, ) -> () { - let result = (|| -> Result<(), PowerSyncError> { - let db = ctx.db_handle(); + let result = (|| -> Result<()> { + let db = Database::from(ctx.db_handle()); verify_in_transaction(db)?; let state = unsafe { DatabaseState::from_context(&ctx) }; @@ -427,9 +431,9 @@ pub fn register(db: *mut sqlite::sqlite3, state: Rc) -> Result<() fn powersync_offline_sync_status_impl( ctx: *mut sqlite::context, _args: &[*mut sqlite::value], -) -> Result { +) -> Result { let db_state = unsafe { DatabaseState::from_context(&ctx) }; - let adapter = db_state.storage_adapter(ctx.db_handle())?; + let adapter = db_state.storage_adapter(ctx.db_handle().into())?; let state = adapter.offline_sync_state()?; let serialized = serde_json::to_string(&state).map_err(PowerSyncError::internal)?; @@ -446,7 +450,7 @@ create_sqlite_text_fn!( /// Errors with a state error unless a sync iteration is currently active. /// /// Checkpoint request ids can only be seeded or allocated in the context of a running iteration. -fn require_active_sync_iteration(state: &DatabaseState) -> Result<(), PowerSyncError> { +fn require_active_sync_iteration(state: &DatabaseState) -> Result<()> { let has_sync_iteration = state .sync_client .borrow() @@ -465,7 +469,7 @@ fn parse_optional_i64_payload( payload: *mut sqlite::value, name: &'static str, type_error: &'static str, -) -> Result, PowerSyncError> { +) -> Result> { let value = match payload.value_type() { ColumnType::Null => return Ok(None), ColumnType::Integer => payload.int64(), @@ -491,7 +495,7 @@ fn parse_positive_i64_payload( payload: *mut sqlite::value, name: &'static str, type_error: &'static str, -) -> Result { +) -> Result { let Some(value) = parse_optional_i64_payload(payload, name, type_error)? else { return Err(PowerSyncError::argument_error(type_error)); }; diff --git a/crates/core/src/sync/operations.rs b/crates/core/src/sync/operations.rs index e3fa0d7b..1a491da1 100644 --- a/crates/core/src/sync/operations.rs +++ b/crates/core/src/sync/operations.rs @@ -1,13 +1,9 @@ use alloc::format; use alloc::string::String; use num_traits::Zero; -use powersync_sqlite_nostd::Connection; -use powersync_sqlite_nostd::{self as sqlite, ResultCode}; +use powersync_sqlite_nostd::{self as sqlite}; -use crate::{ - error::{PSResult, PowerSyncError}, - ext::SafeManagedStmt, -}; +use crate::error::Result; use super::Checksum; use super::line::OplogData; @@ -21,7 +17,7 @@ pub fn insert_bucket_operations( adapter: &StorageAdapter, data: &DataLine, size: usize, -) -> Result<(), PowerSyncError> { +) -> Result<()> { let db = adapter.db; let BucketInfo { id: bucket_id, @@ -84,7 +80,7 @@ INSERT OR IGNORE INTO ps_updated_rows(row_type, row_id) VALUES(?1, ?2)", let mut superseded = false; - while supersede_statement.step()? == ResultCode::ROW { + while supersede_statement.step()? { // Superseded (deleted) a previous operation, add the checksum let supersede_checksum = Checksum::from_i32(supersede_statement.column_int(1)); add_checksum += supersede_checksum; @@ -156,20 +152,16 @@ INSERT OR IGNORE INTO ps_updated_rows(row_type, row_id) VALUES(?1, ?2)", } else if op == OpType::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) + 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)?; + let clear_statement2 = db.prepare_v2("DELETE FROM ps_oplog WHERE bucket = ?1")?; clear_statement2.bind_int64(1, bucket_id)?; clear_statement2.exec()?; diff --git a/crates/core/src/sync/storage_adapter.rs b/crates/core/src/sync/storage_adapter.rs index 74ef4288..c79ccf92 100644 --- a/crates/core/src/sync/storage_adapter.rs +++ b/crates/core/src/sync/storage_adapter.rs @@ -1,12 +1,11 @@ use core::fmt::Display; use alloc::{rc::Rc, string::ToString, vec::Vec}; -use powersync_sqlite_nostd::{self as sqlite, Connection, ManagedStmt, ResultCode}; +use powersync_sqlite_nostd::{self as sqlite}; use serde::Serialize; use crate::{ - error::{PSResult, PowerSyncError}, - ext::SafeManagedStmt, + error::{PowerSyncError, Result}, pre_close_vtab::ensure_has_internal_close_vtab, schema::Schema, state::DatabaseState, @@ -20,7 +19,10 @@ use crate::{ ActiveStreamSubscription, DownloadSyncStatus, SyncPriorityStatus, TimestampMicros, }, }, - utils::{JsonString, column_nullable}, + utils::{ + JsonString, + database::{Database, Statement}, + }, }; use super::{ @@ -42,24 +44,23 @@ pub const TARGET_CHECKPOINT_REQUEST_ID_KEY: &str = "target_checkpoint_request_id /// in `streaming_sync.rs` easier to read. It also allows caching some prepared statements that are /// used frequently as an optimization, but we're not taking advantage of that yet. pub struct StorageAdapter { - pub db: *mut sqlite::sqlite3, - pub progress_stmt: ManagedStmt, - time_stmt: ManagedStmt, - delete_subscription: ManagedStmt, - update_subscription: ManagedStmt, + pub db: Database, + pub progress_stmt: Statement, + time_stmt: Statement, + delete_subscription: Statement, + update_subscription: Statement, } impl StorageAdapter { - pub fn new(db: *mut sqlite::sqlite3) -> Result { + pub fn new(db: Database) -> Result { // The cached statements here prevent sqlite3_close from completing. sqlite3_close invokes // the xDisconnect callback on attached virtual tables, which we use to implement a // "pre-close hook". See `pre_close_vtab.rs` for more details on how that works. ensure_has_internal_close_vtab(db)?; // language=SQLite - let progress = db - .prepare_v2("SELECT name, count_at_last, count_since_last FROM ps_buckets") - .into_db_result(db)?; + let progress = + db.prepare_v2("SELECT name, count_at_last, count_since_last FROM ps_buckets")?; // language=SQLite let time = db.prepare_v2("SELECT CAST(unixepoch('subsec') * 1000000 as integer)")?; @@ -81,16 +82,15 @@ impl StorageAdapter { }) } - pub fn collect_bucket_requests(&self) -> Result, PowerSyncError> { + pub fn collect_bucket_requests(&self) -> Result> { // language=SQLite let statement = self .db - .prepare_v2("SELECT name, last_op FROM ps_buckets WHERE pending_delete = 0") - .into_db_result(self.db)?; + .prepare_v2("SELECT name, last_op FROM ps_buckets WHERE pending_delete = 0")?; let mut requests = Vec::::new(); - while statement.step()? == ResultCode::ROW { + while statement.step()? { let bucket_name = statement.column_text(0)?.to_string(); let last_op = statement.column_int64(1); @@ -103,16 +103,15 @@ impl StorageAdapter { Ok(requests) } - pub fn offline_sync_state(&self) -> Result { + pub fn offline_sync_state(&self) -> Result { let priority_items = { // language=SQLite - let statement = self - .db - .prepare_v2("SELECT priority, last_synced_at FROM ps_sync_state ORDER BY priority") - .into_db_result(self.db)?; + let statement = self.db.prepare_v2( + "SELECT priority, last_synced_at FROM ps_sync_state ORDER BY priority", + )?; let mut items = Vec::::new(); - while statement.step()? == ResultCode::ROW { + while statement.step()? { let priority = BucketPriority { number: statement.column_int(0), }; @@ -144,14 +143,11 @@ impl StorageAdapter { }) } - pub fn delete_buckets<'a>( - &self, - buckets: impl IntoIterator, - ) -> Result<(), ResultCode> { + pub fn delete_buckets<'a>(&self, buckets: impl IntoIterator) -> Result<()> { // Prepare statements lazily, this method may be called without any buckets to delete. - let mut delete_bucket_returning_id = None::; - let mut mark_updated = None::; - let mut delete_oplog = None::; + let mut delete_bucket_returning_id = None::; + let mut mark_updated = None::; + let mut delete_oplog = None::; for bucket in buckets { let delete_bucket_returning_id = match delete_bucket_returning_id { @@ -163,7 +159,7 @@ impl StorageAdapter { }; delete_bucket_returning_id.bind_text(1, bucket, sqlite::Destructor::STATIC)?; - if let ResultCode::ROW = delete_bucket_returning_id.step()? { + if delete_bucket_returning_id.step()? { let bucket_id = delete_bucket_returning_id.column_int64(0); let mark_updated = match mark_updated { @@ -195,8 +191,8 @@ WHERE bucket = ?1", Ok(()) } - pub fn step_progress(&'_ self) -> Result>, ResultCode> { - if self.progress_stmt.step()? == ResultCode::ROW { + pub fn step_progress(&'_ self) -> Result>> { + if self.progress_stmt.step()? { let bucket = self.progress_stmt.column_text(0)?; let count_at_last = self.progress_stmt.column_int64(1); let count_since_last = self.progress_stmt.column_int64(2); @@ -213,30 +209,26 @@ WHERE bucket = ?1", } } - pub fn reset_progress(&self) -> Result<(), PowerSyncError> { + pub fn reset_progress(&self) -> Result<()> { self.db - .exec_safe("UPDATE ps_buckets SET count_since_last = 0, count_at_last = 0;") - .into_db_result(self.db)?; + .exec_safe(c"UPDATE ps_buckets SET count_since_last = 0, count_at_last = 0;")?; Ok(()) } - pub fn lookup_bucket(&self, bucket: &str) -> Result { + pub fn lookup_bucket(&self, bucket: &str) -> Result { // 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 = self - .db - .prepare_v2( - "INSERT INTO ps_buckets(name) + let bucket_statement = self.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", - ) - .into_db_result(self.db)?; + )?; bucket_statement.bind_text(1, bucket, sqlite::Destructor::STATIC)?; - let res = bucket_statement.step()?; - debug_assert_eq!(res, ResultCode::ROW); + let has_row = bucket_statement.step()?; + debug_assert!(has_row); let bucket_id = bucket_statement.column_int64(0); let last_applied_op = bucket_statement.column_int64(1); @@ -253,7 +245,7 @@ WHERE bucket = ?1", checkpoint: &OwnedCheckpoint, priority: Option, schema: &Schema, - ) -> Result { + ) -> Result { let mismatched_checksums = validate_checkpoint(checkpoint.buckets.values(), priority, self.db)?; @@ -267,8 +259,7 @@ WHERE bucket = ?1", let update_bucket = self .db - .prepare_v2("UPDATE ps_buckets SET last_op = ? WHERE name = ?") - .into_db_result(self.db)?; + .prepare_v2("UPDATE ps_buckets SET last_op = ? WHERE name = ?")?; for bucket in checkpoint.buckets.values() { if bucket.is_in_priority(priority) { @@ -335,7 +326,7 @@ WHERE bucket = ?1", // partial completions. let update = self.db.prepare_v2( "UPDATE ps_buckets SET count_since_last = 0, count_at_last = ? WHERE name = ?", - ).into_db_result(self.db)?; + )?; for bucket in checkpoint.buckets.values() { if let Some(count) = bucket.count { @@ -357,7 +348,7 @@ WHERE bucket = ?1", pub fn collect_subscription_requests( &self, include_defaults: bool, - ) -> Result { + ) -> Result { self.delete_outdated_subscriptions()?; let mut subscriptions: Vec = Vec::new(); @@ -369,7 +360,7 @@ WHERE bucket = ?1", .db .prepare_v2("SELECT * FROM ps_stream_subscriptions WHERE ttl IS NOT NULL;")?; - while let ResultCode::ROW = stmt.step()? { + while stmt.step()? { let subscription = Self::read_stream_subscription(&stmt)?; subscriptions.push(RequestedStreamSubscription { @@ -389,7 +380,7 @@ WHERE bucket = ?1", }) } - pub fn now(&self) -> Result { + pub fn now(&self) -> Result { self.time_stmt.step()?; let res = TimestampMicros(self.time_stmt.column_int64(0)); self.time_stmt.reset()?; @@ -397,9 +388,7 @@ WHERE bucket = ?1", Ok(res) } - fn read_stream_subscription( - stmt: &ManagedStmt, - ) -> Result { + fn read_stream_subscription(stmt: &Statement) -> Result { let raw_params = stmt.column_text(5)?; Ok(LocallyTrackedSubscription { @@ -407,21 +396,20 @@ WHERE bucket = ?1", stream_name: stmt.column_text(1)?.to_string(), active: stmt.column_int(2) != 0, is_default: stmt.column_int(3) != 0, - local_priority: column_nullable(&stmt, 4, || { - BucketPriority::try_from(stmt.column_int(4)) - })?, + local_priority: stmt + .column_nullable(4, || BucketPriority::try_from(stmt.column_int(4)))?, local_params: if raw_params == "null" { None } else { Some(JsonString::from_string(stmt.column_text(5)?.to_string())?) }, - ttl: column_nullable(&stmt, 6, || Ok(stmt.column_int64(6)))?, - expires_at: column_nullable(&stmt, 7, || Ok(stmt.column_int64(7)))?, - last_synced_at: column_nullable(&stmt, 8, || Ok(stmt.column_int64(8)))?, + ttl: stmt.column_nullable(6, || Ok(stmt.column_int64(6)))?, + expires_at: stmt.column_nullable(7, || Ok(stmt.column_int64(7)))?, + last_synced_at: stmt.column_nullable(8, || Ok(stmt.column_int64(8)))?, }) } - fn delete_outdated_subscriptions(&self) -> Result<(), PowerSyncError> { + fn delete_outdated_subscriptions(&self) -> Result<()> { let now = self.now()?; let stmt = self.db.prepare_v2("DELETE FROM ps_stream_subscriptions WHERE (expires_at < ?) OR (ttl IS NULL AND NOT active)")?; stmt.bind_int64(1, now.0)?; @@ -430,7 +418,7 @@ WHERE bucket = ?1", } /// Increases the TTL for explicit subscriptions that are currently marked as active. - pub fn increase_ttl(&self, streams: &[StreamKey]) -> Result<(), PowerSyncError> { + pub fn increase_ttl(&self, streams: &[StreamKey]) -> Result<()> { let now = self.now()?; let stmt = self.db.prepare_v2( "UPDATE ps_stream_subscriptions SET expires_at = ? + ttl * 1000000 WHERE stream_name = ? AND local_params = ? AND ttl IS NOT NULL", @@ -449,12 +437,12 @@ WHERE bucket = ?1", pub fn iterate_local_subscriptions ()>( &self, mut action: F, - ) -> Result<(), PowerSyncError> { + ) -> Result<()> { let stmt = self .db .prepare_v2("SELECT * FROM ps_stream_subscriptions ORDER BY id ASC")?; - while stmt.step()? == ResultCode::ROW { + while stmt.step()? { action(Self::read_stream_subscription(&stmt)?); } Ok(()) @@ -463,22 +451,19 @@ WHERE bucket = ?1", pub fn create_default_subscription( &self, stream: &OwnedStreamDescription, - ) -> Result { + ) -> Result { debug_assert!(stream.is_default); let stmt = self.db.prepare_v2("INSERT INTO ps_stream_subscriptions (stream_name, active, is_default) VALUES (?, TRUE, TRUE) RETURNING *;")?; stmt.bind_text(1, &stream.name, sqlite::Destructor::STATIC)?; - if stmt.step()? == ResultCode::ROW { + if stmt.step()? { Self::read_stream_subscription(&stmt) } else { Err(PowerSyncError::unknown_internal()) } } - pub fn update_subscription( - &self, - subscription: &LocallyTrackedSubscription, - ) -> Result<(), PowerSyncError> { + pub fn update_subscription(&self, subscription: &LocallyTrackedSubscription) -> Result<()> { let _ = self.update_subscription.reset(); self.update_subscription.bind_int64(1, subscription.id)?; @@ -508,14 +493,14 @@ WHERE bucket = ?1", Ok(()) } - pub fn delete_subscription(&self, id: i64) -> Result<(), PowerSyncError> { + pub fn delete_subscription(&self, id: i64) -> Result<()> { let _ = self.delete_subscription.reset(); self.delete_subscription.bind_int64(1, id)?; self.delete_subscription.exec()?; Ok(()) } - pub fn target_checkpoint_request_id(&self) -> Result, PowerSyncError> { + pub fn target_checkpoint_request_id(&self) -> Result> { self.read_i64_kv(TARGET_CHECKPOINT_REQUEST_ID_KEY) } @@ -540,10 +525,7 @@ WHERE bucket = ?1", /// /// Negative values are rejected when parsing the `powersync_control` payload, before this is /// called. - pub fn probe_target_checkpoint_request_id( - &self, - target: Option, - ) -> Result, PowerSyncError> { + pub fn probe_target_checkpoint_request_id(&self, target: Option) -> Result> { let previous_target = self.target_checkpoint_request_id()?; let Some(target) = target else { @@ -563,10 +545,7 @@ WHERE bucket = ?1", /// Persists the checkpoint request id observed in a complete sync checkpoint. /// /// This is used to decide whether downloaded data can be applied after local uploads complete. - pub fn persist_last_seen_checkpoint_request_id( - &self, - request_id: i64, - ) -> Result<(), PowerSyncError> { + pub fn persist_last_seen_checkpoint_request_id(&self, request_id: i64) -> Result<()> { self.write_i64_kv(LAST_SEEN_CHECKPOINT_REQUEST_ID_KEY, request_id) } @@ -576,15 +555,12 @@ WHERE bucket = ?1", /// overwrite with no monotonicity enforced by core. External code owns consistency of these /// ids; waiters should rely on `DidCompleteSync.applied_checkpoint_request_id` rather than /// comparing this value across reconnects. - pub fn persist_last_applied_checkpoint_request_id( - &self, - request_id: i64, - ) -> Result<(), PowerSyncError> { + pub fn persist_last_applied_checkpoint_request_id(&self, request_id: i64) -> Result<()> { self.write_i64_kv(LAST_APPLIED_CHECKPOINT_REQUEST_ID_KEY, request_id) } /// Increments, persists and returns the next client-created checkpoint request id. - pub fn next_checkpoint_request_id(&self) -> Result { + pub fn next_checkpoint_request_id(&self) -> Result { let statement = self.db.prepare_v2( "INSERT INTO ps_kv(key, value) VALUES(?1, 1) @@ -597,7 +573,7 @@ RETURNING value", sqlite::Destructor::STATIC, )?; - if statement.step()? == ResultCode::ROW { + if statement.step()? { Ok(statement.column_int64(0)) } else { Err(PowerSyncError::unknown_internal()) @@ -605,12 +581,12 @@ RETURNING value", } /// Returns whether the local checkpoint request counter has been initialized. - pub fn has_checkpoint_request_id(&self) -> Result { + pub fn has_checkpoint_request_id(&self) -> Result { Ok(self.last_checkpoint_request_id()?.is_some()) } /// Returns the latest checkpoint request id known locally. - pub fn last_checkpoint_request_id(&self) -> Result, PowerSyncError> { + pub fn last_checkpoint_request_id(&self) -> Result> { self.read_i64_kv(LAST_REQUESTED_CHECKPOINT_REQUEST_ID_KEY) } @@ -620,7 +596,7 @@ RETURNING value", /// legacy-to-request-mode transition or after migrating unusual state, so include it as a /// lower bound. The max-op sentinel represents pending local writes without a concrete request /// id and must not be sent to the service. - pub fn initial_checkpoint_request_id(&self) -> Result { + pub fn initial_checkpoint_request_id(&self) -> Result { let last_requested = self.last_checkpoint_request_id()?.unwrap_or(0); let concrete_target = self .target_checkpoint_request_id()? @@ -635,25 +611,24 @@ RETURNING value", /// The value is stored verbatim: core does not enforce monotonicity here. SDKs are /// responsible for posting the initial payload to the service and seeding its accepted /// response, and cannot allocate new checkpoint request ids until that seeding has completed. - pub fn seed_checkpoint_request_id(&self, request_id: i64) -> Result<(), PowerSyncError> { + pub fn seed_checkpoint_request_id(&self, request_id: i64) -> Result<()> { self.write_i64_kv(LAST_REQUESTED_CHECKPOINT_REQUEST_ID_KEY, request_id) } - fn read_i64_kv(&self, key: &'static str) -> Result, PowerSyncError> { + fn read_i64_kv(&self, key: &'static str) -> Result> { let statement = self .db - .prepare_v2("SELECT value FROM ps_kv WHERE key = ?1") - .into_db_result(self.db)?; + .prepare_v2("SELECT value FROM ps_kv WHERE key = ?1")?; statement.bind_text(1, key, sqlite::Destructor::STATIC)?; - Ok(if statement.step()? == ResultCode::ROW { + Ok(if statement.step()? { Some(statement.column_int64(0)) } else { None }) } - fn write_i64_kv(&self, key: &'static str, value: i64) -> Result<(), PowerSyncError> { + fn write_i64_kv(&self, key: &'static str, value: i64) -> Result<()> { let stmt = self.db.prepare_v2( "INSERT INTO ps_kv(key, value) VALUES(?1, ?2) @@ -665,7 +640,7 @@ ON CONFLICT(key) DO UPDATE SET value = excluded.value", Ok(()) } - fn delete_kv(&self, key: &'static str) -> Result<(), PowerSyncError> { + fn delete_kv(&self, key: &'static str) -> Result<()> { let stmt = self.db.prepare_v2("DELETE FROM ps_kv WHERE key = ?1")?; stmt.bind_text(1, key, sqlite::Destructor::STATIC)?; stmt.exec()?; diff --git a/crates/core/src/sync/streaming_sync.rs b/crates/core/src/sync/streaming_sync.rs index 067743ce..986d160f 100644 --- a/crates/core/src/sync/streaming_sync.rs +++ b/crates/core/src/sync/streaming_sync.rs @@ -18,8 +18,7 @@ use alloc::{ use futures_lite::FutureExt; use crate::{ - error::{PowerSyncError, PowerSyncErrorCause}, - ext::SafeManagedStmt, + error::{PowerSyncError, PowerSyncErrorCause, Result}, kv::client_id, state::DatabaseState, sync::{ @@ -37,8 +36,8 @@ use crate::{ subscriptions::LocallyTrackedSubscription, sync_status::{ActiveStreamSubscription, TimestampMicros}, }, + utils::database::Database, }; -use powersync_sqlite_nostd::{self as sqlite, Connection}; use super::{ interface::{Instruction, LogSeverity, StreamingSyncRequest, SyncControlRequest, SyncEvent}, @@ -54,7 +53,7 @@ use super::{ /// The client consumes no resources and prepares no statements until a sync iteration is /// initialized. pub struct SyncClient { - db: *mut sqlite::sqlite3, + db: Database, adapter: Rc, db_state: Weak, /// The current [ClientState] (essentially an optional [StreamingSyncIteration]). @@ -62,10 +61,7 @@ pub struct SyncClient { } impl SyncClient { - pub fn new( - db: *mut sqlite::sqlite3, - state: &Rc, - ) -> Result { + pub fn new(db: Database, state: &Rc) -> Result { let adapter = state.storage_adapter(db)?; Ok(Self { @@ -76,10 +72,7 @@ impl SyncClient { }) } - pub fn push_event<'a>( - &mut self, - event: SyncControlRequest<'a>, - ) -> Result, PowerSyncError> { + pub fn push_event<'a>(&mut self, event: SyncControlRequest<'a>) -> Result> { match event { SyncControlRequest::StartSyncStream(options) => { self.state.tear_down()?; @@ -138,7 +131,7 @@ enum ClientState { } impl ClientState { - fn tear_down(&mut self) -> Result, PowerSyncError> { + fn tear_down(&mut self) -> Result> { let mut event = ActiveEvent::new(SyncEvent::TearDown); if let ClientState::IterationActive(old) = self { @@ -157,14 +150,14 @@ impl ClientState { /// At each invocation, the future is polled once (and gets access to context that allows it to /// render [Instruction]s to return from the function). struct SyncIterationHandle { - future: Pin>>>, + future: Pin>>>, } impl SyncIterationHandle { /// Creates a new sync iteration in a pending state by preparing statements for /// [StorageAdapter] and setting up the initial downloading state for [StorageAdapter] . fn new( - db: *mut sqlite::sqlite3, + db: Database, options: StartSyncStream, adapter: Rc, state: Weak, @@ -184,7 +177,7 @@ impl SyncIterationHandle { /// Forwards a [SyncEvent::Initialize] to the current sync iteration, returning the initial /// instructions generated. - fn initialize(&mut self) -> Result, PowerSyncError> { + fn initialize(&mut self) -> Result> { let mut event = ActiveEvent::new(SyncEvent::Initialize); let result = self.run(&mut event)?; assert!(!result, "Stream client aborted initialization"); @@ -192,7 +185,7 @@ impl SyncIterationHandle { Ok(event.instructions) } - fn run(&mut self, active: &mut ActiveEvent) -> Result { + fn run(&mut self, active: &mut ActiveEvent) -> Result { // Using a noop waker because the only event thing StreamingSyncIteration::run polls on is // the next incoming sync event. let waker = unsafe { @@ -245,7 +238,7 @@ impl<'a> ActiveEvent<'a> { } struct StreamingSyncIteration { - db: *mut sqlite::sqlite3, + db: Database, state: Weak, adapter: Rc, options: StartSyncStream, @@ -292,7 +285,7 @@ impl StreamingSyncIteration { target: &SyncTarget, event: &mut ActiveEvent, line: &'a SyncLineWithSource<'a>, - ) -> Result, PowerSyncError> { + ) -> Result> { let SyncLineWithSource { source, line } = line; Ok(match line { @@ -542,14 +535,14 @@ impl StreamingSyncIteration { target: &mut SyncTarget, event: &mut ActiveEvent, line: &SyncLineWithSource, - ) -> Result, PowerSyncError> { + ) -> Result> { let transition = self.prepare_handling_sync_line(target, event, line)?; Ok(self.apply_transition(target, event, transition)) } /// Runs a full sync iteration, returning nothing when it completes regularly or an error when /// the sync iteration should be interrupted. - async fn run(mut self) -> Result { + async fn run(mut self) -> Result { let mut target = SyncTarget::BeforeCheckpoint(self.prepare_request().await?); let hide_disconnect = loop { @@ -625,10 +618,7 @@ impl StreamingSyncIteration { Ok(CloseSyncStream { hide_disconnect }) } - fn load_progress( - &self, - checkpoint: &OwnedCheckpoint, - ) -> Result { + fn load_progress(&self, checkpoint: &OwnedCheckpoint) -> Result { let SyncProgressFromCheckpoint { progress, needs_counter_reset, @@ -641,10 +631,7 @@ impl StreamingSyncIteration { Ok(progress) } - fn try_applying_write_after_completed_upload( - &mut self, - event: &mut ActiveEvent, - ) -> Result<(), PowerSyncError> { + fn try_applying_write_after_completed_upload(&mut self, event: &mut ActiveEvent) -> Result<()> { let Some(checkpoint) = self.validated_but_not_applied.take() else { return Ok(()); }; @@ -695,7 +682,7 @@ impl StreamingSyncIteration { &self, tracked: &TrackedCheckpoint, event: &mut ActiveEvent, - ) -> Result, PowerSyncError> { + ) -> Result> { struct LocalAndServerSubscription<'a, T> { local: T, /// If this subscription has an acknowledged stream included in the checkpoint, the @@ -865,7 +852,7 @@ impl StreamingSyncIteration { &self, target: &OwnedCheckpoint, priority: Option, - ) -> Result { + ) -> Result { let state = match self.state.upgrade() { Some(state) => state, None => return Err(PowerSyncError::unknown_internal()), @@ -906,7 +893,7 @@ impl StreamingSyncIteration { /// This returns local bucket names (used to delete buckets that don't appear in checkpoints /// anymore) and the [LocallyTrackedSubscription::id] of explicitly-requested stream /// subscriptions, used to associate [BucketSubscriptionReason::DerivedFromExplicitSubscription]. - async fn prepare_request(&mut self) -> Result { + async fn prepare_request(&mut self) -> Result { let event = Self::receive_event().await; let SyncEvent::Initialize = event.event else { return Err(PowerSyncError::argument_error( diff --git a/crates/core/src/sync/subscriptions.rs b/crates/core/src/sync/subscriptions.rs index 65d758bc..da456dc3 100644 --- a/crates/core/src/sync/subscriptions.rs +++ b/crates/core/src/sync/subscriptions.rs @@ -1,13 +1,12 @@ use core::time::Duration; use alloc::{boxed::Box, string::String}; -use powersync_sqlite_nostd::{self as sqlite, Connection}; +use powersync_sqlite_nostd::{self as sqlite}; use serde::Deserialize; use serde_with::{DurationSeconds, serde_as}; use crate::{ - error::{PSResult, PowerSyncError}, - ext::SafeManagedStmt, + error::Result, sync::{BucketPriority, storage_adapter::StorageAdapter}, utils::JsonString, }; @@ -81,16 +80,15 @@ pub struct SubscribeToStream { pub fn apply_subscriptions( adapter: &StorageAdapter, subscription: SubscriptionChangeRequest, -) -> Result<(), PowerSyncError> { +) -> Result<()> { let db = adapter.db; match subscription { SubscriptionChangeRequest::Subscribe(subscription) => { let now = adapter.now()?; - let stmt = db - .prepare_v2( - " + let stmt = db.prepare_v2( + " INSERT INTO ps_stream_subscriptions (stream_name, local_priority, local_params, ttl, expires_at) VALUES (?, ?2, ?, ?4, ?5) ON CONFLICT DO UPDATE SET @@ -99,8 +97,7 @@ INSERT INTO ps_stream_subscriptions (stream_name, local_priority, local_params, ttl = ?4, expires_at = ?5 ", - ) - .into_db_result(db)?; + )?; stmt.bind_text(1, &subscription.stream.name, sqlite::Destructor::STATIC)?; match &subscription.priority { @@ -124,8 +121,7 @@ INSERT INTO ps_stream_subscriptions (stream_name, local_priority, local_params, } SubscriptionChangeRequest::Unsubscribe(subscription) => { let stmt = db - .prepare_v2("UPDATE ps_stream_subscriptions SET ttl = NULL WHERE stream_name = ? AND local_params = ?") - .into_db_result(db)?; + .prepare_v2("UPDATE ps_stream_subscriptions SET ttl = NULL WHERE stream_name = ? AND local_params = ?")?; stmt.bind_text(1, &subscription.name, sqlite::Destructor::STATIC)?; stmt.bind_text( 2, diff --git a/crates/core/src/sync/sync_local.rs b/crates/core/src/sync/sync_local.rs index d8770b91..c9734120 100644 --- a/crates/core/src/sync/sync_local.rs +++ b/crates/core/src/sync/sync_local.rs @@ -5,7 +5,7 @@ use alloc::string::{String, ToString}; use serde::Serialize; use serde::ser::SerializeMap; -use crate::error::{PSResult, PowerSyncError}; +use crate::error::{PowerSyncError, Result}; use crate::schema::inspection::ExistingTable; use crate::schema::{ InferredSchemaCache, PendingStatement, PendingStatementValue, RawTable, Schema, @@ -17,11 +17,9 @@ use crate::sync::storage_adapter::{ }; use crate::sync::sync_status::TimestampMicros; use crate::utils::SqlBuffer; +use crate::utils::database::{Database, Statement}; use const_format::formatcp; -use powersync_sqlite_nostd::{self as sqlite, Destructor, ManagedStmt}; -use powersync_sqlite_nostd::{Connection, ResultCode}; - -use crate::ext::SafeManagedStmt; +use powersync_sqlite_nostd::{self as sqlite, Destructor}; pub struct PartialSyncOperation<'a> { /// The lowest priority part of the partial sync operation. @@ -33,7 +31,7 @@ pub struct PartialSyncOperation<'a> { pub struct SyncOperation<'a> { state: &'a DatabaseState, - db: *mut sqlite::sqlite3, + db: Database, schema: ParsedDatabaseSchema<'a>, partial: Option>, time: TimestampMicros, @@ -42,7 +40,7 @@ pub struct SyncOperation<'a> { impl<'a> SyncOperation<'a> { pub fn new( state: &'a DatabaseState, - db: *mut sqlite::sqlite3, + db: Database, partial: Option>, time: TimestampMicros, ) -> Self { @@ -59,7 +57,7 @@ impl<'a> SyncOperation<'a> { self.schema.add_from_schema(schema); } - fn can_apply_sync_changes(&self) -> Result { + fn can_apply_sync_changes(&self) -> Result { // Don't publish downloaded data until the upload queue is empty (except for downloaded data // in priority 0, which is published earlier). @@ -78,12 +76,12 @@ WHERE target.key = '{TARGET_CHECKPOINT_REQUEST_ID_KEY}' AND CAST(target.value AS INTEGER) > COALESCE(CAST(seen.value AS INTEGER), 0)" ))?; - if statement.step()? == ResultCode::ROW { + if statement.step()? { return Ok(false); } let statement = self.db.prepare_v2("SELECT 1 FROM ps_crud LIMIT 1")?; - if statement.step()? != ResultCode::DONE { + if statement.step()? { return Ok(false); } } @@ -91,7 +89,7 @@ WHERE target.key = '{TARGET_CHECKPOINT_REQUEST_ID_KEY}' Ok(true) } - pub fn apply(&mut self) -> Result { + pub fn apply(&mut self) -> Result { let guard = self.state.sync_local_guard(); if !self.can_apply_sync_changes()? { @@ -107,16 +105,16 @@ WHERE target.key = '{TARGET_CHECKPOINT_REQUEST_ID_KEY}' // We cache the last insert and delete statements for each row struct CachedStatement { table: String, - statement: ManagedStmt, + statement: Statement, } let mut last_insert = None::; let mut last_delete = None::; - let mut untyped_delete_statement: Option = None; - let mut untyped_insert_statement: Option = None; + let mut untyped_delete_statement: Option = None; + let mut untyped_insert_statement: Option = None; - while statement.step().into_db_result(self.db)? == ResultCode::ROW { + while statement.step()? { let type_name = statement.column_text(0)?; let id = statement.column_text(1)?; let data = statement.column_text(2); @@ -136,13 +134,13 @@ WHERE target.key = '{TARGET_CHECKPOINT_REQUEST_ID_KEY}' let rest = stmt.render_rest_object(json_object)?; stmt.bind_for_put(id, &json_object, &rest)?; - stmt.exec(self.db, type_name, id, Some(&parsed))?; + stmt.exec(type_name, id, Some(&parsed))?; } Err(_) => { let stmt = raw.delete_statement(self.db, schema_version, schema_cache)?; stmt.bind_for_delete(id)?; - stmt.exec(self.db, type_name, id, None)?; + stmt.exec(type_name, id, None)?; } } } else { @@ -159,8 +157,7 @@ WHERE target.key = '{TARGET_CHECKPOINT_REQUEST_ID_KEY}' statement.quote_internal_name(type_name, false); statement.push_str(" WHERE id = ?"); - let statement = - self.db.prepare_v2(&statement.sql).into_db_result(self.db)?; + let statement = self.db.prepare_v2(&statement.sql)?; &last_delete .insert(CachedStatement { @@ -185,8 +182,7 @@ WHERE target.key = '{TARGET_CHECKPOINT_REQUEST_ID_KEY}' statement.quote_internal_name(type_name, false); statement.push_str("(id, data) VALUES (?, ?)"); - let statement = - self.db.prepare_v2(&statement.sql).into_db_result(self.db)?; + let statement = self.db.prepare_v2(&statement.sql)?; &last_insert .insert(CachedStatement { @@ -206,17 +202,16 @@ WHERE target.key = '{TARGET_CHECKPOINT_REQUEST_ID_KEY}' } else { if data.is_err() { // DELETE - let delete_statement = match &untyped_delete_statement { - Some(stmt) => stmt, - None => { - // Prepare statement on first use - untyped_delete_statement.insert( - self.db - .prepare_v2("DELETE FROM ps_untyped WHERE type = ? AND id = ?") - .into_db_result(self.db)?, - ) - } - }; + let delete_statement = + match &untyped_delete_statement { + Some(stmt) => stmt, + None => { + // Prepare statement on first use + untyped_delete_statement.insert(self.db.prepare_v2( + "DELETE FROM ps_untyped WHERE type = ? AND id = ?", + )?) + } + }; delete_statement.reset()?; delete_statement.bind_text(1, type_name, sqlite::Destructor::STATIC)?; @@ -228,13 +223,9 @@ WHERE target.key = '{TARGET_CHECKPOINT_REQUEST_ID_KEY}' Some(stmt) => stmt, None => { // Prepare statement on first use - untyped_insert_statement.insert( - self.db - .prepare_v2( - "REPLACE INTO ps_untyped(type, id, data) VALUES(?, ?, ?)", - ) - .into_db_result(self.db)?, - ) + untyped_insert_statement.insert(self.db.prepare_v2( + "REPLACE INTO ps_untyped(type, id, data) VALUES(?, ?, ?)", + )?) } }; @@ -254,18 +245,17 @@ WHERE target.key = '{TARGET_CHECKPOINT_REQUEST_ID_KEY}' Ok(1) } - fn collect_tables(&mut self) -> Result<(), PowerSyncError> { + fn collect_tables(&mut self) -> Result<()> { self.schema.add_from_db(self.db) } - fn collect_full_operations(&self) -> Result { + fn collect_full_operations(&self) -> Result { Ok(match &self.partial { None => { // Complete sync // See dart/test/sync_local_performance_test.dart for an annotated version of this query. - self.db - .prepare_v2( - "\ + self.db.prepare_v2( + "\ WITH updated_rows AS ( SELECT b.row_type, b.row_id FROM ps_buckets AS buckets CROSS JOIN ps_oplog AS b ON b.bucket = buckets.id @@ -285,14 +275,11 @@ SELECT ) as data FROM updated_rows b GROUP BY b.row_type, b.row_id;", - ) - .into_db_result(self.db)? + )? } Some(partial) => { - let stmt = self - .db - .prepare_v2( - "\ + let stmt = self.db.prepare_v2( + "\ -- 1. Filter oplog by the ops added but not applied yet (oplog b). -- We do not do any DISTINCT operation here, since that introduces a temp b-tree. -- We filter out duplicates using the GROUP BY below. @@ -326,8 +313,7 @@ SELECT FROM updated_rows b -- Group for (2) GROUP BY b.row_type, b.row_id;", - ) - .into_db_result(self.db)?; + )?; stmt.bind_text(1, partial.args, Destructor::STATIC)?; stmt @@ -335,7 +321,7 @@ SELECT }) } - fn set_last_applied_op(&self) -> Result<(), PowerSyncError> { + fn set_last_applied_op(&self) -> Result<()> { match &self.partial { Some(partial) => { // language=SQLite @@ -346,32 +332,28 @@ SELECT SET last_applied_op = last_op WHERE last_applied_op != last_op AND name IN (SELECT value FROM json_each(json_extract(?1, '$.buckets')))", - ) .into_db_result(self.db)?; + )?; updated.bind_text(1, partial.args, Destructor::STATIC)?; updated.exec()?; } None => { // language=SQLite - self.db - .exec_safe( - "UPDATE ps_buckets + self.db.exec_safe( + c"UPDATE ps_buckets SET last_applied_op = last_op WHERE last_applied_op != last_op", - ) - .into_db_result(self.db)?; + )?; } } Ok(()) } - fn mark_completed(&self) -> Result<(), PowerSyncError> { + fn mark_completed(&self) -> Result<()> { let priority_code: i32 = match &self.partial { None => { // language=SQLite - self.db - .exec_safe("DELETE FROM ps_updated_rows") - .into_db_result(self.db)?; + self.db.exec_safe(c"DELETE FROM ps_updated_rows")?; BucketPriority::SENTINEL } Some(partial) => partial.priority, @@ -384,18 +366,14 @@ SELECT // language=SQLite let stmt = self .db - .prepare_v2("DELETE FROM ps_sync_state WHERE priority < ?1;") - .into_db_result(self.db)?; + .prepare_v2("DELETE FROM ps_sync_state WHERE priority < ?1;")?; stmt.bind_int(1, priority_code)?; stmt.exec()?; // language=SQLite - let stmt = self - .db - .prepare_v2( - "INSERT OR REPLACE INTO ps_sync_state (priority, last_synced_at) VALUES (?, ?);", - ) - .into_db_result(self.db)?; + let stmt = self.db.prepare_v2( + "INSERT OR REPLACE INTO ps_sync_state (priority, last_synced_at) VALUES (?, ?);", + )?; stmt.bind_int(1, priority_code)?; stmt.bind_int64(2, self.time.0)?; stmt.exec()?; @@ -422,7 +400,7 @@ impl<'a> ParsedDatabaseSchema<'a> { } } - fn add_from_db(&mut self, db: *mut sqlite::sqlite3) -> Result<(), PowerSyncError> { + fn add_from_db(&mut self, db: Database) -> Result<()> { let tables = ExistingTable::list(db)?; for table in tables { if !table.local_only { @@ -449,10 +427,10 @@ struct RawTableWithCachedStatements<'a> { impl<'a> RawTableWithCachedStatements<'a> { fn prepare_lazily( - db: *mut sqlite::sqlite3, + db: Database, slot: &mut Option, def: Rc, - ) -> Result<&PreparedPendingStatement, PowerSyncError> { + ) -> Result<&PreparedPendingStatement> { Ok(match slot { Some(stmt) => stmt, None => { @@ -464,10 +442,10 @@ impl<'a> RawTableWithCachedStatements<'a> { fn put_statement( &'_ mut self, - db: *mut sqlite::sqlite3, + db: Database, schema_version: usize, cache: &InferredSchemaCache, - ) -> Result<&'_ PreparedPendingStatement, PowerSyncError> { + ) -> Result<&'_ PreparedPendingStatement> { Self::prepare_lazily( db, &mut self.cached_put, @@ -480,10 +458,10 @@ impl<'a> RawTableWithCachedStatements<'a> { fn delete_statement( &'_ mut self, - db: *mut sqlite::sqlite3, + db: Database, schema_version: usize, cache: &InferredSchemaCache, - ) -> Result<&'_ PreparedPendingStatement, PowerSyncError> { + ) -> Result<&'_ PreparedPendingStatement> { Self::prepare_lazily( db, &mut self.cached_delete, @@ -512,17 +490,14 @@ impl<'a> ParsedSchemaTable<'a> { } struct PreparedPendingStatement { - stmt: ManagedStmt, + stmt: Statement, definition: Rc, } impl PreparedPendingStatement { - pub fn prepare( - db: *mut sqlite::sqlite3, - pending: Rc, - ) -> Result { - let stmt = db.prepare_v2(&pending.sql).into_db_result(db)?; - if stmt.bind_parameter_count() as usize != pending.params.len() { + pub fn prepare(db: Database, pending: Rc) -> Result { + let stmt = db.prepare_v2(&pending.sql)?; + if stmt.bind_parameter_count() != pending.params.len() { return Err(PowerSyncError::argument_error(format!( "Statement {} has {} parameters, but {} values were provided as sources.", &pending.sql, @@ -542,7 +517,7 @@ impl PreparedPendingStatement { pub fn render_rest_object( &self, json_data: &serde_json::Map, - ) -> Result, PowerSyncError> { + ) -> Result> { use serde_json::Value; let Some(ref index) = self.definition.named_parameters_index else { @@ -552,7 +527,7 @@ impl PreparedPendingStatement { struct UnmatchedValues<'a>(BTreeMap<&'a String, &'a Value>); impl<'a> Serialize for UnmatchedValues<'a> { - fn serialize(&self, serializer: S) -> Result + fn serialize(&self, serializer: S) -> core::result::Result where S: serde::Serializer, { @@ -589,7 +564,7 @@ impl PreparedPendingStatement { id: &str, json_data: &serde_json::Map, rest: &Option, - ) -> Result<(), PowerSyncError> { + ) -> Result<()> { use serde_json::Value; for (i, source) in self.definition.params.iter().enumerate() { @@ -639,7 +614,7 @@ impl PreparedPendingStatement { Ok(()) } - pub fn bind_for_delete(&self, id: &str) -> Result<(), PowerSyncError> { + pub fn bind_for_delete(&self, id: &str) -> Result<()> { for (i, source) in self.definition.params.iter().enumerate() { if let PendingStatementValue::Id = source { self.stmt @@ -656,23 +631,14 @@ impl PreparedPendingStatement { /// Executes the prepared statement, contextualizing errors with the id / data that we've tried /// to insert. - pub fn exec( - &self, - db: *mut sqlite::sqlite3, - table: &str, - id: &str, - data: Option<&serde_json::Value>, - ) -> Result<(), PowerSyncError> { - match self.stmt.exec() { - Ok(_) => Ok(()), - Err(rc) => { - let context = match data { - None => format!("deleting from {table}, id = {id}"), - Some(data) => format!("replacing into {table}, id = {id}, data = {data}"), - }; - - Err(PowerSyncError::from_sqlite(db, rc, context)) - } - } + pub fn exec(&self, table: &str, id: &str, data: Option<&serde_json::Value>) -> Result<()> { + self.stmt.exec().map_err(|e| { + let context = match data { + None => format!("deleting from {table}, id = {id}"), + Some(data) => format!("replacing into {table}, id = {id}, data = {data}"), + }; + + e.context(context) + }) } } diff --git a/crates/core/src/sync/sync_status.rs b/crates/core/src/sync/sync_status.rs index 86876c8a..a100aa54 100644 --- a/crates/core/src/sync/sync_status.rs +++ b/crates/core/src/sync/sync_status.rs @@ -12,7 +12,6 @@ use core::{ hash::{BuildHasher, Hash}, ops::AddAssign, }; -use powersync_sqlite_nostd::ResultCode; use rustc_hash::FxBuildHasher; use serde::{ Serialize, @@ -20,6 +19,7 @@ use serde::{ }; use crate::{ + error::PowerSyncError, sync::{ checkpoint::OwnedBucketChecksum, storage_adapter::StorageAdapter, subscriptions::LocallyTrackedSubscription, @@ -358,7 +358,7 @@ impl SyncDownloadProgress { pub fn for_checkpoint<'a>( checkpoint: &OwnedCheckpoint, adapter: &StorageAdapter, - ) -> Result { + ) -> Result { let mut buckets = BTreeMap::::new(); let mut needs_reset = false; for bucket in checkpoint.buckets.values() { diff --git a/crates/core/src/update_hooks.rs b/crates/core/src/update_hooks.rs index 4702cf78..dd0b6226 100644 --- a/crates/core/src/update_hooks.rs +++ b/crates/core/src/update_hooks.rs @@ -60,8 +60,8 @@ extern "C" fn powersync_update_hooks( match op { "install" => { - if let Err(e) = ensure_has_internal_close_vtab(db) { - ctx.result_error_code(e); + if let Err(e) = ensure_has_internal_close_vtab(db.into()) { + e.apply_to_ctx("powersync_update_hooks", ctx); return; }; diff --git a/crates/core/src/utils/database.rs b/crates/core/src/utils/database.rs new file mode 100644 index 00000000..8f7eb53a --- /dev/null +++ b/crates/core/src/utils/database.rs @@ -0,0 +1,197 @@ +use core::ffi::{CStr, c_char}; + +use alloc::ffi::CString; +use num_traits::FromPrimitive; +use powersync_sqlite_nostd::{ + self as sqlite, ColumnType, Destructor, ManagedStmt, ResultCode, convert_rc, +}; + +use crate::error::{PowerSyncError, Result}; + +/// A safe-ish wrapper around SQLite statements, providing better errors including the causing +/// statement. +#[repr(transparent)] +#[derive(Clone, Copy, PartialEq, Eq)] +pub struct Database { + pub sqlite: *mut sqlite::sqlite3, +} + +impl From<*mut sqlite::sqlite3> for Database { + fn from(sqlite: *mut sqlite::sqlite3) -> Self { + Self { sqlite } + } +} + +impl Database { + fn map_error(self, code: ResultCode, sql: Option<&str>) -> PowerSyncError { + PowerSyncError::from_sqlite(self.sqlite, code, sql) + } + + fn map_error_cstr(self, code: ResultCode, sql: Option<&CStr>) -> PowerSyncError { + PowerSyncError::from_sqlite(self.sqlite, code, sql) + } + + pub fn use_inner( + self, + inner: impl FnOnce(*mut sqlite::sqlite3) -> core::result::Result, + ) -> Result { + inner(self.sqlite).map_err(|e| self.map_error(e, None)) + } + + pub fn get_autocommit(self) -> bool { + sqlite::get_autocommit(self.sqlite) != 0 + } + + pub fn prepare_v2(self, sql: &str) -> Result { + self.prepare_v3(sql, 0) + } + + pub fn prepare_v3(self, sql: &str, flags: u32) -> Result { + let mut stmt = core::ptr::null_mut(); + let mut tail = core::ptr::null(); + let rc = ResultCode::from_i32(sqlite::prepare_v3( + self.sqlite, + sql.as_ptr() as *const c_char, + sql.len() as i32, + flags, + &mut stmt as *mut *mut sqlite::stmt, + &mut tail as *mut *const c_char, + )) + .unwrap(); + if rc == ResultCode::OK { + Ok(Statement { + db: self, + stmt: ManagedStmt { stmt }, + }) + } else { + Err(self.map_error(rc, Some(sql))) + } + } + + pub fn exec_safe(self, sql: &CStr) -> Result<()> { + convert_rc(sqlite::exec(self.sqlite, sql.as_ptr())) + .map_err(|e| self.map_error_cstr(e, Some(sql)))?; + Ok(()) + } + + pub fn exec_safe_str(self, sql: &str) -> Result<()> { + self.exec_safe(&CString::new(sql)?) + } + + pub fn exec_text(self, sql: &str, param: &str) -> Result<()> { + let statement = self.prepare_v2(sql)?; + statement.bind_text(1, param, Destructor::STATIC)?; + statement.exec() + } +} + +pub struct Statement { + db: Database, + stmt: ManagedStmt, +} + +impl Statement { + pub fn map_error(&self, code: ResultCode) -> PowerSyncError { + let sql_ptr = sqlite::sql(self.stmt.stmt); + let str = if sql_ptr.is_null() { + None + } else { + Some(unsafe { CStr::from_ptr(sql_ptr) }) + }; + + self.db.map_error_cstr(code, str) + } + + pub fn step(&self) -> Result { + let rc = ResultCode::from_i32(sqlite::step(self.stmt.stmt)).unwrap(); + + match rc { + ResultCode::ROW => Ok(true), + ResultCode::DONE => Ok(false), + _ => Err(self.map_error(rc)), + } + } + + pub fn bind_parameter_count(&self) -> usize { + self.stmt.bind_parameter_count() as usize + } + + pub fn bind_text(&self, i: i32, text: &str, d: Destructor) -> Result<()> { + self.stmt + .bind_text(i, text, d) + .map_err(|e| self.map_error(e))?; + Ok(()) + } + + pub fn bind_int(&self, i: i32, val: i32) -> Result<()> { + self.stmt.bind_int(i, val).map_err(|e| self.map_error(e))?; + Ok(()) + } + + pub fn bind_int64(&self, i: i32, val: i64) -> Result<()> { + self.stmt + .bind_int64(i, val) + .map_err(|e| self.map_error(e))?; + Ok(()) + } + + pub fn bind_double(&self, i: i32, val: f64) -> Result<()> { + self.stmt + .bind_double(i, val) + .map_err(|e| self.map_error(e))?; + Ok(()) + } + + pub fn bind_null(&self, i: i32) -> Result<()> { + self.stmt.bind_null(i).map_err(|e| self.map_error(e))?; + Ok(()) + } + + /// Calls [read] to read a column if it's not null, otherwise returns [None]. + #[inline] + pub fn column_nullable Result>( + &self, + index: i32, + read: R, + ) -> Result> { + if self.stmt.column_type(index) == ColumnType::Null { + Ok(None) + } else { + Ok(Some(read()?)) + } + } + + pub fn column_text(&self, i: i32) -> Result<&str> { + self.stmt.column_text(i).map_err(|e| self.map_error(e)) + } + + pub fn column_int(&self, i: i32) -> i32 { + self.stmt.column_int(i) + } + + pub fn column_int64(&self, i: i32) -> i64 { + self.stmt.column_int64(i) + } + + pub fn reset(&self) -> Result<()> { + self.stmt.reset().map_err(|e| self.map_error(e))?; + Ok(()) + } + + pub fn exec(&self) -> Result<()> { + let result = loop { + break match self.step() { + Ok(row) => { + if row { + continue; + }; + Ok(()) + } + Err(e) => Err(e), + }; + }; + + self.reset()?; + result + } +} diff --git a/crates/core/src/utils/mod.rs b/crates/core/src/utils/mod.rs index 34f233a6..6a118935 100644 --- a/crates/core/src/utils/mod.rs +++ b/crates/core/src/utils/mod.rs @@ -1,14 +1,17 @@ +pub mod database; mod sql_buffer; use core::{cmp::Ordering, fmt::Display, hash::Hash}; use alloc::{boxed::Box, string::String}; -use powersync_sqlite_nostd::{ColumnType, Connection, ManagedStmt, sqlite3}; use serde::Serialize; use serde_json::value::RawValue; pub use sql_buffer::{InsertIntoCrud, SqlBuffer, WriteType}; -use crate::error::{PowerSyncError, RawPowerSyncError}; +use crate::{ + error::{PowerSyncError, RawPowerSyncError}, + utils::database::Database, +}; use uuid::Uuid; #[cold] @@ -17,7 +20,7 @@ fn must_be_in_tx_error() -> PowerSyncError { } #[inline] -pub fn verify_in_transaction(db: *mut sqlite3) -> Result<(), PowerSyncError> { +pub fn verify_in_transaction(db: Database) -> Result<(), PowerSyncError> { if db.get_autocommit() { return Err(must_be_in_tx_error()); } @@ -25,20 +28,6 @@ pub fn verify_in_transaction(db: *mut sqlite3) -> Result<(), PowerSyncError> { Ok(()) } -/// Calls [read] to read a column if it's not null, otherwise returns [None]. -#[inline] -pub fn column_nullable Result>( - stmt: &ManagedStmt, - index: i32, - read: R, -) -> Result, PowerSyncError> { - if stmt.column_type(index)? == ColumnType::Null { - Ok(None) - } else { - Ok(Some(read()?)) - } -} - /// An opaque wrapper around a JSON-serialized value. /// /// This wraps [RawValue] from `serde_json`, adding implementations for comparisons and hashes. diff --git a/crates/core/src/uuid.rs b/crates/core/src/uuid.rs index 3a84b69c..7ee03497 100644 --- a/crates/core/src/uuid.rs +++ b/crates/core/src/uuid.rs @@ -15,7 +15,7 @@ use crate::utils::gen_uuid; fn uuid_v4_impl( _ctx: *mut sqlite::context, _args: &[*mut sqlite::value], -) -> Result { +) -> Result { let id = gen_uuid(); Ok(id.hyphenated().to_string()) } diff --git a/crates/core/src/version.rs b/crates/core/src/version.rs index 603d2150..96214486 100644 --- a/crates/core/src/version.rs +++ b/crates/core/src/version.rs @@ -15,7 +15,7 @@ use crate::error::PowerSyncError; fn powersync_rs_version_impl( _ctx: *mut sqlite::context, _args: &[*mut sqlite::value], -) -> Result { +) -> Result { let version = format!("{}/{}", CORE_PKG_VERSION, short_git_hash()); Ok(version) } diff --git a/crates/core/src/view_admin.rs b/crates/core/src/view_admin.rs index 7c1d517e..cfe8b930 100644 --- a/crates/core/src/view_admin.rs +++ b/crates/core/src/view_admin.rs @@ -11,10 +11,11 @@ use powersync_sqlite_nostd::{Connection, Context}; use sqlite::{ResultCode, Value}; use crate::create_sqlite_text_fn; -use crate::error::{PSResult, PowerSyncError}; +use crate::error::{PowerSyncError, Result}; use crate::migrations::{LATEST_VERSION, powersync_migrate}; use crate::schema::inspection::ExistingView; use crate::state::DatabaseState; +use crate::utils::database::Database; use crate::utils::{SqlBuffer, verify_in_transaction}; // Used in old down migrations, do not remove. @@ -26,16 +27,13 @@ extern "C" fn powersync_drop_view( let args = sqlite::args!(argc, argv); let name = args[0].text(); - if let Err(e) = ExistingView::drop_by_name(ctx.db_handle(), name) { + if let Err(e) = ExistingView::drop_by_name(ctx.db_handle().into(), name) { e.apply_to_ctx("powersync_drop_view", ctx); } } -fn powersync_init_impl( - ctx: *mut sqlite::context, - _args: &[*mut sqlite::value], -) -> Result { - let db = ctx.db_handle(); +fn powersync_init_impl(ctx: *mut sqlite::context, _args: &[*mut sqlite::value]) -> Result { + let db = Database::from(ctx.db_handle()); verify_in_transaction(db)?; powersync_migrate(ctx, LATEST_VERSION)?; @@ -47,8 +45,8 @@ create_sqlite_text_fn!(powersync_init, powersync_init_impl, "powersync_init"); fn powersync_test_migration_impl( ctx: *mut sqlite::context, args: &[*mut sqlite::value], -) -> Result { - let db = ctx.db_handle(); +) -> Result { + let db = Database::from(ctx.db_handle()); verify_in_transaction(db)?; let target_version = args[0].int(); @@ -63,11 +61,8 @@ create_sqlite_text_fn!( "powersync_test_migration" ); -fn powersync_clear_impl( - ctx: *mut sqlite::context, - args: &[*mut sqlite::value], -) -> Result { - let local_db = ctx.db_handle(); +fn powersync_clear_impl(ctx: *mut sqlite::context, args: &[*mut sqlite::value]) -> Result { + let local_db = Database::from(ctx.db_handle()); verify_in_transaction(local_db)?; let state = unsafe { DatabaseState::from_context(&ctx) }; @@ -77,14 +72,14 @@ fn powersync_clear_impl( // With a soft clear, we want to delete public data while keeping internal data around. When // connect() is called with compatible JWTs yielding a large overlap of buckets, this can // speed up the next sync. - local_db.exec_safe("DELETE FROM ps_oplog; DELETE FROM ps_buckets")?; + local_db.exec_safe(c"DELETE FROM ps_oplog; DELETE FROM ps_buckets")?; } else { trigger_resync(local_db, state)?; } // language=SQLite local_db.exec_safe( - "\ + c"\ DELETE FROM ps_crud; DELETE FROM ps_untyped; DELETE FROM ps_updated_rows; @@ -106,7 +101,7 @@ DELETE FROM ps_stream_subscriptions; let mut tables: Vec = alloc::vec![]; - while tables_stmt.step()? == ResultCode::ROW { + while tables_stmt.step()? { let name = tables_stmt.column_text(0)?; tables.push(name.to_string()); } @@ -121,7 +116,7 @@ DELETE FROM {table} WHERE rowid IN (SELECT rowid FROM {table} LIMIT 1); DELETE FROM {table};", table = quoted ); - local_db.exec_safe(&delete_sql)?; + local_db.exec_safe_str(&delete_sql)?; } if let Some(schema) = state.view_schema() { @@ -132,13 +127,9 @@ DELETE FROM {table};", for raw_table in &schema.raw_tables { if let Some(stmt) = &raw_table.clear { - local_db.exec_safe(&stmt).map_err(|e| { - PowerSyncError::from_sqlite( - local_db, - e, - format!("Clearing raw table {}", raw_table.name), - ) - })?; + local_db + .exec_safe_str(&stmt) + .map_err(|e| e.context(format!("Clearing raw table {}", raw_table.name)))?; } } } @@ -146,7 +137,7 @@ DELETE FROM {table};", Ok(String::from("")) } -fn trigger_resync(db: *mut sqlite::sqlite3, state: &DatabaseState) -> Result<(), PowerSyncError> { +fn trigger_resync(db: Database, state: &DatabaseState) -> Result<()> { { let client = state.sync_client.borrow(); if let Some(client) = client.as_ref() @@ -158,22 +149,21 @@ fn trigger_resync(db: *mut sqlite::sqlite3, state: &DatabaseState) -> Result<(), } } - db.exec_safe("UPDATE ps_buckets SET last_applied_op = 0") - .into_db_result(db)?; + db.exec_safe(c"UPDATE ps_buckets SET last_applied_op = 0")?; Ok(Default::default()) } -fn clear_has_synced(db: *mut sqlite::sqlite3) -> Result<(), PowerSyncError> { - db.exec_safe("DELETE FROM ps_sync_state;")?; - db.exec_safe("UPDATE ps_stream_subscriptions SET last_synced_at = NULL")?; +fn clear_has_synced(db: Database) -> Result<()> { + db.exec_safe(c"DELETE FROM ps_sync_state;")?; + db.exec_safe(c"UPDATE ps_stream_subscriptions SET last_synced_at = NULL")?; Ok(()) } fn powersync_trigger_resync_impl( ctx: *mut sqlite::context, args: &[*mut sqlite::value], -) -> Result { - let local_db = ctx.db_handle(); +) -> Result { + let local_db = Database::from(ctx.db_handle()); verify_in_transaction(local_db)?; let state = unsafe { DatabaseState::from_context(&ctx) }; @@ -211,7 +201,10 @@ impl PowerSyncClearFlags { create_sqlite_text_fn!(powersync_clear, powersync_clear_impl, "powersync_clear"); -pub fn register(db: *mut sqlite::sqlite3, state: Rc) -> Result<(), ResultCode> { +pub fn register( + db: *mut sqlite::sqlite3, + state: Rc, +) -> core::result::Result<(), ResultCode> { // This entire module is just making it easier to edit sqlite_master using queries. // Internal function, used exclusively in existing migrations. diff --git a/crates/core/src/views.rs b/crates/core/src/views.rs index d87cdd04..a2c1b51b 100644 --- a/crates/core/src/views.rs +++ b/crates/core/src/views.rs @@ -5,7 +5,7 @@ use alloc::vec; use core::fmt::{Write, from_fn}; use core::mem; -use crate::error::PowerSyncError; +use crate::error::{PowerSyncError, Result}; use crate::schema::{ColumnFilter, SchemaTable, Table}; use crate::utils::{InsertIntoCrud, SqlBuffer, WriteType}; @@ -58,7 +58,7 @@ pub fn powersync_view_sql(table_info: &Table) -> String { return sql.sql; } -pub fn powersync_trigger_delete_sql(table_info: &Table) -> Result { +pub fn powersync_trigger_delete_sql(table_info: &Table) -> Result { if table_info.options.flags.insert_only() { // Insert-only tables have no DELETE triggers return Ok(String::new()); @@ -116,7 +116,7 @@ pub fn powersync_trigger_delete_sql(table_info: &Table) -> Result Result { +pub fn powersync_trigger_insert_sql(table_info: &Table) -> Result { let name = &table_info.name; let view_name = table_info.view_name(); let local_only = table_info.options.flags.local_only(); @@ -167,7 +167,7 @@ pub fn powersync_trigger_insert_sql(table_info: &Table) -> Result Result { +pub fn powersync_trigger_update_sql(table_info: &Table) -> Result { if table_info.options.flags.insert_only() { // Insert-only tables have no UPDATE triggers return Ok(String::new()); @@ -232,7 +232,7 @@ pub fn powersync_trigger_update_sql(table_info: &Table) -> Result( prefix: &str, table: &'a SchemaTable<'a>, -) -> Result { +) -> Result { table_columns_to_json_object_with_filter(prefix, table, None) } @@ -240,7 +240,7 @@ pub fn table_columns_to_json_object_with_filter<'a>( prefix: &str, table: &'a SchemaTable<'a>, filter: Option<&'a ColumnFilter>, -) -> Result { +) -> Result { // floor(SQLITE_MAX_FUNCTION_ARG / 2). // To keep databases portable, we use the default limit of 100 args for this, // and don't try to query the limit dynamically. diff --git a/crates/sqlite_nostd/src/nostd.rs b/crates/sqlite_nostd/src/nostd.rs index d49b9c07..160b2d24 100644 --- a/crates/sqlite_nostd/src/nostd.rs +++ b/crates/sqlite_nostd/src/nostd.rs @@ -227,17 +227,6 @@ pub enum ColumnType { Null = 5, } -pub fn open(filename: *const c_char) -> Result { - let mut db = core::ptr::null_mut(); - let rc = - ResultCode::from_i32(sqlite3_capi::open(filename, &mut db as *mut *mut sqlite3)).unwrap(); - if rc == ResultCode::OK { - Ok(ManagedConnection { db }) - } else { - Err(rc) - } -} - pub fn libversion() -> &'static str { unsafe { CStr::from_ptr(sqlite3_capi::libversion()) } .to_str() @@ -252,10 +241,6 @@ pub fn randomness(blob: &mut [u8]) { sqlite3_capi::randomness(blob.len() as c_int, blob.as_mut_ptr() as *mut c_void) } -pub struct ManagedConnection { - pub db: *mut sqlite3, -} - pub trait Connection { fn commit_hook(&self, callback: Option, user_data: *mut c_void) -> *mut c_void; @@ -285,144 +270,9 @@ pub trait Connection { fn errmsg(&self) -> Result; fn error_offset(&self) -> Option; - fn exec(&self, sql: &CStr) -> Result; - - fn exec_safe(&self, sql: &str) -> Result; - - fn next_stmt(&self, s: Option<*mut stmt>) -> Option<*mut stmt>; - - fn prepare_v2(&self, sql: &str) -> Result; - - fn prepare_v3(&self, sql: &str, flags: u32) -> Result; - - fn set_authorizer( - &self, - x_auth: Option, - user_data: *mut c_void, - ) -> Result; - fn rollback_hook(&self, callback: Option, ctx: *mut c_void) -> *mut c_void; fn update_hook(&self, callback: Option, ctx: *mut c_void) -> *mut c_void; - - fn get_autocommit(&self) -> bool; -} - -impl Connection for ManagedConnection { - fn changes64(&self) -> i64 { - self.db.changes64() - } - - fn commit_hook(&self, callback: Option, user_data: *mut c_void) -> *mut c_void { - self.db.commit_hook(callback, user_data) - } - - /// TODO: create_function is infrequent enough that we can pay the cost of the copy rather than - /// take a c_char - fn create_function_v2( - &self, - name: &str, - n_arg: i32, - flags: u32, - user_data: Option<*mut c_void>, - func: Option, - step: Option, - final_func: Option, - destroy: Option, - ) -> Result { - self.db.create_function_v2( - name, n_arg, flags, user_data, func, step, final_func, destroy, - ) - } - - fn set_authorizer( - &self, - x_auth: Option, - user_data: *mut c_void, - ) -> Result { - self.db.set_authorizer(x_auth, user_data) - } - - fn create_module_v2( - &self, - name: &str, - module: *const module, - user_data: Option<*mut c_void>, - destroy: Option, - ) -> Result { - self.db.create_module_v2(name, module, user_data, destroy) - } - - #[inline] - fn next_stmt(&self, s: Option<*mut stmt>) -> Option<*mut stmt> { - self.db.next_stmt(s) - } - - #[inline] - fn prepare_v2(&self, sql: &str) -> Result { - self.db.prepare_v2(sql) - } - - #[inline] - fn prepare_v3(&self, sql: &str, flags: u32) -> Result { - self.db.prepare_v3(sql, flags) - } - - #[inline] - fn exec(&self, sql: &CStr) -> Result { - self.db.exec(sql) - } - - #[inline] - fn exec_safe(&self, sql: &str) -> Result { - self.db.exec_safe(sql) - } - - #[inline] - fn errmsg(&self) -> Result { - self.db.errmsg() - } - - #[inline] - fn errcode(&self) -> ResultCode { - self.db.errcode() - } - - fn error_offset(&self) -> Option { - self.db.error_offset() - } - - #[inline] - fn get_autocommit(&self) -> bool { - self.db.get_autocommit() - } - - fn rollback_hook(&self, callback: Option, ctx: *mut c_void) -> *mut c_void { - self.db.rollback_hook(callback, ctx) - } - - fn update_hook(&self, callback: Option, ctx: *mut c_void) -> *mut c_void { - self.db.update_hook(callback, ctx) - } -} - -impl Drop for ManagedConnection { - fn drop(&mut self) { - // todo: iterate over all stmts and finalize them? - let rc = sqlite3_capi::close(self.db); - if rc != 0 { - // This seems aggressive... - // The alternative is to make users manually drop connections and manually finalize - // stmts :/ - // Or we could not panic.. but then you will unknowningly have memory - // leaks in your app. The reason being that a failure to close the db - // does not release the memory of that db. - panic!( - "SQLite returned error {:?} when trying to close the db!", - rc - ); - } - } } impl Connection for *mut sqlite3 { @@ -482,79 +332,6 @@ impl Connection for *mut sqlite3 { } } - #[inline] - fn prepare_v2(&self, sql: &str) -> Result { - let mut stmt = core::ptr::null_mut(); - let mut tail = core::ptr::null(); - let rc = ResultCode::from_i32(prepare_v2( - *self, - sql.as_ptr() as *const c_char, - sql.len() as i32, - &mut stmt as *mut *mut stmt, - &mut tail as *mut *const c_char, - )) - .unwrap(); - if rc == ResultCode::OK { - Ok(ManagedStmt { stmt: stmt }) - } else { - Err(rc) - } - } - - #[inline] - fn prepare_v3(&self, sql: &str, flags: u32) -> Result { - let mut stmt = core::ptr::null_mut(); - let mut tail = core::ptr::null(); - let rc = ResultCode::from_i32(prepare_v3( - *self, - sql.as_ptr() as *const c_char, - sql.len() as i32, - flags, - &mut stmt as *mut *mut stmt, - &mut tail as *mut *const c_char, - )) - .unwrap(); - if rc == ResultCode::OK { - Ok(ManagedStmt { stmt: stmt }) - } else { - Err(rc) - } - } - - #[inline] - fn exec(&self, sql: &CStr) -> Result { - convert_rc(exec(*self, sql.as_ptr())) - } - - #[inline] - fn exec_safe(&self, sql: &str) -> Result { - if let Ok(sql) = CString::new(sql) { - convert_rc(exec(*self, sql.as_ptr())) - } else { - return Err(ResultCode::NOMEM); - } - } - - #[inline] - fn next_stmt(&self, s: Option<*mut stmt>) -> Option<*mut stmt> { - let s = if let Some(s) = s { - s - } else { - core::ptr::null_mut() - }; - - let ptr = next_stmt(*self, s); - if ptr.is_null() { None } else { Some(ptr) } - } - - fn set_authorizer( - &self, - x_auth: Option, - user_data: *mut c_void, - ) -> Result { - convert_rc(set_authorizer(*self, x_auth, user_data)) - } - fn errmsg(&self) -> Result { errmsg(*self).into_string() } @@ -570,10 +347,6 @@ impl Connection for *mut sqlite3 { } } - fn get_autocommit(&self) -> bool { - get_autocommit(*self) != 0 - } - fn rollback_hook(&self, callback: Option, ctx: *mut c_void) -> *mut c_void { rollback_hook(*self, callback, ctx) } @@ -638,8 +411,8 @@ impl ManagedStmt { } #[inline] - pub fn column_type(&self, i: i32) -> Result { - ColumnType::from_i32(column_type(self.stmt, i)).ok_or(ResultCode::NULL) + pub fn column_type(&self, i: i32) -> ColumnType { + ColumnType::from_i32(column_type(self.stmt, i)).unwrap_or(ColumnType::Null) } #[inline] diff --git a/dart/test/error_test.dart b/dart/test/error_test.dart index 7ac7c800..b5bc51ff 100644 --- a/dart/test/error_test.dart +++ b/dart/test/error_test.dart @@ -50,7 +50,7 @@ void main() { () => db.executeInTx('SELECT powersync_init()'), throwsA(isSqliteException( 1, - 'powersync_init: internal SQLite call returned ERROR: no such column: id', + 'powersync_init: statement SELECT ifnull(max(id), 0) as version FROM ps_migration: internal SQLite call returned ERROR: no such column: id', )), ); }); diff --git a/dart/test/sync_test.dart b/dart/test/sync_test.dart index 31e68c59..40271975 100644 --- a/dart/test/sync_test.dart +++ b/dart/test/sync_test.dart @@ -1509,8 +1509,8 @@ void _syncTests({ expect( () => syncLine(checkpoint), throwsA( - isSqliteException( - 5, 'powersync_control: internal SQLite call returned BUSY'), + isSqliteException(5, + 'powersync_control: statement DELETE FROM ps_buckets WHERE name = ?1 RETURNING id: internal SQLite call returned BUSY: database is locked'), ), ); secondary.execute('commit'); @@ -1533,7 +1533,7 @@ void _syncTests({ expect( () => pushSyncData('a', '1', '1', 'PUT', {'col': 'hi'}), throwsA(isSqliteException( - 5, 'powersync_control: internal SQLite call returned BUSY')), + 5, contains('internal SQLite call returned BUSY'))), ); // But we should be able to retry @@ -1572,7 +1572,7 @@ void _syncTests({ () => pushCheckpointComplete(), throwsA( isSqliteException( - 5, 'powersync_control: internal SQLite call returned BUSY'), + 5, contains('internal SQLite call returned BUSY')), ), ); secondary.execute('commit'); @@ -1867,9 +1867,9 @@ SELECT throwsA( isSqliteException( 1299, - 'powersync_control: replacing into users, id = my_user, data = {}: ' - 'internal SQLite call returned CONSTRAINT_NOTNULL: ' - 'NOT NULL constraint failed: users.name', + 'powersync_control: statement INSERT OR REPLACE INTO users (id, name) VALUES (?, ?);: ' + 'internal SQLite call returned CONSTRAINT_NOTNULL: NOT NULL constraint failed: users.name ' + '(context: replacing into users, id = my_user, data = {})', ), ), );