feat: initial pass on remote writes for offline databases

Author: levydsa

libsql/src/database.rs

@@ -8,9 +8,11 @@ pub use builder::Builder;
 pub use libsql_sys::{Cipher, EncryptionConfig};
 
 use crate::{Connection, Result};
-use std::fmt;
-use std::sync::atomic::AtomicU64;
-use std::sync::Arc;
+use std::{
+    fmt,
+    sync::{atomic::AtomicU64, Arc},
+};
+use tokio::sync::Mutex;
 
 cfg_core! {
     bitflags::bitflags! {
@@ -82,7 +84,14 @@ enum DbType {
         encryption_config: Option<EncryptionConfig>,
     },
     #[cfg(feature = "sync")]
-    Offline { db: crate::local::Database },
+    Offline {
+        db: crate::local::Database,
+        remote_writes: bool,
+        read_your_writes: bool,
+        url: String,
+        auth_token: String,
+        connector: crate::util::ConnectorService,
+    },
     #[cfg(feature = "remote")]
     Remote {
         url: String,
@@ -375,7 +384,7 @@ cfg_replication! {
                 #[cfg(feature = "replication")]
                 DbType::Sync { db, encryption_config: _ } => db.sync().await,
                 #[cfg(feature = "sync")]
-                DbType::Offline { db } => db.sync_offline().await,
+                DbType::Offline { db, .. } => db.sync_offline().await,
                 _ => Err(Error::SyncNotSupported(format!("{:?}", self.db_type))),
             }
         }
@@ -542,7 +551,7 @@ impl Database {
 
                 let conn = db.connect()?;
 
-                let conn = std::sync::Arc::new(LibsqlConnection { conn });
+                let conn = Arc::new(LibsqlConnection { conn });
 
                 Ok(Connection { conn })
             }
@@ -590,7 +599,7 @@ impl Database {
                     }
                 }
 
-                let conn = std::sync::Arc::new(LibsqlConnection { conn });
+                let conn = Arc::new(LibsqlConnection { conn });
 
                 Ok(Connection { conn })
             }
@@ -636,19 +645,47 @@ impl Database {
                     writer,
                     self.max_write_replication_index.clone(),
                 );
-                let conn = std::sync::Arc::new(remote);
+                let conn = Arc::new(remote);
 
                 Ok(Connection { conn })
             }
 
             #[cfg(feature = "sync")]
-            DbType::Offline { db } => {
-                use crate::local::impls::LibsqlConnection;
-
-                let conn = db.connect()?;
+            DbType::Offline {
+                db,
+                remote_writes,
+                read_your_writes,
+                url,
+                auth_token,
+                connector,
+            } => {
+                use crate::{
+                    hrana::{connection::HttpConnection, hyper::HttpSender},
+                    local::impls::LibsqlConnection,
+                    replication::connection::State,
+                    sync::connection::SyncedConnection,
+                };
 
-                let conn = std::sync::Arc::new(LibsqlConnection { conn });
+                let local = db.connect()?;
+
+                if *remote_writes {
+                    let synced = SyncedConnection {
+                        local,
+                        remote: HttpConnection::new(
+                            url.clone(),
+                            auth_token.clone(),
+                            HttpSender::new(connector.clone(), None),
+                        ),
+                        read_your_writes: *read_your_writes,
+                        context: db.sync_ctx.clone().unwrap(),
+                        state: Arc::new(Mutex::new(State::Init)),
+                    };
+
+                    let conn = Arc::new(synced);
+                    return Ok(Connection { conn });
+                }
 
+                let conn = Arc::new(LibsqlConnection { conn: local });
                 Ok(Connection { conn })
             }
 
@@ -659,7 +696,7 @@ impl Database {
                 connector,
                 version,
             } => {
-                let conn = std::sync::Arc::new(
+                let conn = Arc::new(
                     crate::hrana::connection::HttpConnection::new_with_connector(
                         url,
                         auth_token,

libsql/src/database/builder.rs

@@ -102,7 +102,9 @@ impl Builder<()> {
                         connector: None,
                         version: None,
                     },
-                    connector:None,
+                    connector: None,
+                    read_your_writes: true,
+                    remote_writes: false,
                 },
             }
         }
@@ -463,6 +465,8 @@ cfg_sync! {
         flags: crate::OpenFlags,
         remote: Remote,
         connector: Option<crate::util::ConnectorService>,
+        remote_writes: bool,
+        read_your_writes: bool,
     }
 
     impl Builder<SyncedDatabase> {
@@ -472,6 +476,16 @@ cfg_sync! {
             self
         }
 
+        pub fn read_your_writes(mut self, v: bool) -> Builder<SyncedDatabase> {
+            self.inner.read_your_writes = v;
+            self
+        }
+
+        pub fn remote_writes(mut self, v: bool) -> Builder<SyncedDatabase> {
+            self.inner.remote_writes = v;
+            self
+        }
+
         /// Provide a custom http connector that will be used to create http connections.
         pub fn connector<C>(mut self, connector: C) -> Builder<SyncedDatabase>
         where
@@ -497,6 +511,8 @@ cfg_sync! {
                         version: _,
                     },
                 connector,
+                remote_writes,
+                read_your_writes,
             } = self.inner;
 
             let path = path.to_str().ok_or(crate::Error::InvalidUTF8Path)?.to_owned();
@@ -515,16 +531,23 @@ cfg_sync! {
             let connector = crate::util::ConnectorService::new(svc);
 
             let db = crate::local::Database::open_local_with_offline_writes(
-                connector,
+                connector.clone(),
                 path,
                 flags,
-                url,
-                auth_token,
+                url.clone(),
+                auth_token.clone(),
             )
             .await?;
 
             Ok(Database {
-                db_type: DbType::Offline { db },
+                db_type: DbType::Offline {
+                    db,
+                    remote_writes,
+                    read_your_writes,
+                    url,
+                    auth_token,
+                    connector,
+                },
                 max_write_replication_index: Default::default(),
             })
         }

libsql/src/hrana/hyper.rs

@@ -305,14 +305,17 @@ impl Conn for HranaStream<HttpSender> {
         let parse = crate::parser::Statement::parse(sql);
         for s in parse {
             let s = s?;
-            if s.kind == crate::parser::StmtKind::TxnBegin
-                || s.kind == crate::parser::StmtKind::TxnBeginReadOnly
-                || s.kind == crate::parser::StmtKind::TxnEnd
-            {
+
+            use crate::parser::StmtKind;
+            if matches!(
+                s.kind,
+                StmtKind::TxnBegin | StmtKind::TxnBeginReadOnly | StmtKind::TxnEnd
+            ) {
                 return Err(Error::TransactionalBatchError(
                     "Transactions forbidden inside transactional batch".to_string(),
                 ));
             }
+
             stmts.push(Stmt::new(s.stmt, false));
         }
         let res = self

libsql/src/hrana/mod.rs

@@ -3,7 +3,7 @@
 pub mod connection;
 
 cfg_remote! {
-    mod hyper;
+    pub mod hyper;
 }
 
 mod cursor;