wip

libsql/Cargo.toml

@@ -14,7 +14,7 @@ thiserror = "1.0.40"
 futures = { version = "0.3.28", optional = true }
 libsql-sys = { workspace = true, optional = true, default-features = true }
 libsql-hrana = { workspace = true, optional = true }
-tokio = { version = "1.29.1", features = ["sync"], optional = true }
+tokio = { version = "1.29.1", features = ["rt-multi-thread", "sync"], optional = true }
 tokio-util = { version = "0.7", features = ["io-util", "codec"], optional = true }
 parking_lot = { version = "0.12.1", optional = true }
 hyper = { version = "0.14", features = ["client", "http1", "http2", "stream", "runtime"], optional = true }
@@ -46,6 +46,10 @@ async-stream = { version = "0.3.5", optional = true }
 
 crc32fast = { version = "1", optional = true }
 chrono = { version = "0.4", optional = true }
+zstd = "0.13.3"
+rand = "0.8.5"
+lazy_static = "1.5.0"
+arc-swap = "1.7.1"
 
 [dev-dependencies]
 criterion = { version = "0.5", features = ["html_reports", "async", "async_futures", "async_tokio"] }
@@ -57,7 +61,7 @@ tempfile = { version = "3.7.0" }
 rand = "0.8.5"
 
 [features]
-default = ["core", "replication", "remote", "sync", "tls"]
+default = ["core", "replication", "remote", "sync", "tls", "lazy"]
 core = [
   "libsql-sys",
   "dep:bitflags",
@@ -116,6 +120,27 @@ sync = [
   "dep:uuid",
   "tokio/fs"
 ]
+lazy = [
+  "core",
+  "parser",
+  "serde",
+  "stream",
+  "remote",
+  "replication",
+  "dep:tower",
+  "dep:hyper",
+  "dep:http",
+  "dep:tokio",
+  "dep:zerocopy",
+  "dep:bytes",
+  "dep:tokio",
+  "dep:futures",
+  "dep:serde_json",
+  "dep:crc32fast",
+  "dep:chrono",
+  "dep:uuid",
+  "tokio/fs"
+]
 hrana = [
   "parser",
   "serde",

libsql/src/database.rs

@@ -76,7 +76,7 @@ cfg_sync! {
     }
 }
 
-enum DbType {
+pub(crate) enum DbType {
     #[cfg(feature = "core")]
     Memory { db: crate::local::Database },
     #[cfg(feature = "core")]
@@ -107,6 +107,13 @@ enum DbType {
         connector: crate::util::ConnectorService,
         version: Option<String>,
     },
+    #[cfg(feature = "lazy")]
+    Lazy {
+        db: crate::local::Database,
+        url: String,
+        auth_token: String,
+        connector: crate::util::ConnectorService,
+    },
 }
 
 impl fmt::Debug for DbType {
@@ -131,10 +138,10 @@ impl fmt::Debug for DbType {
 /// A struct that knows how to build [`Connection`]'s, this type does
 /// not do much work until the [`Database::connect`] fn is called.
 pub struct Database {
-    db_type: DbType,
+    pub(crate) db_type: DbType,
     /// The maximum replication index returned from a write performed using any connection created using this Database object.
     #[allow(dead_code)]
-    max_write_replication_index: std::sync::Arc<AtomicU64>,
+    pub(crate) max_write_replication_index: std::sync::Arc<AtomicU64>,
 }
 
 cfg_core! {
@@ -727,7 +734,7 @@ impl Database {
     all(feature = "tls", feature = "remote"),
     all(feature = "tls", feature = "sync")
 ))]
-fn connector() -> Result<hyper_rustls::HttpsConnector<hyper::client::HttpConnector>> {
+pub(crate) fn connector() -> Result<hyper_rustls::HttpsConnector<hyper::client::HttpConnector>> {
     let mut http = hyper::client::HttpConnector::new();
     http.enforce_http(false);
     http.set_nodelay(true);

libsql/src/database/builder.rs

@@ -2,6 +2,9 @@ cfg_core! {
     use crate::EncryptionConfig;
 }
 
+use anyhow::anyhow;
+use hyper::server::conn;
+
 use crate::{Database, Result};
 
 use super::DbType;
@@ -126,6 +129,18 @@ impl Builder<()> {
             }
         }
     }
+
+    cfg_lazy! {
+        pub fn new_lazy(
+            path: impl AsRef<std::path::Path>,
+            url: String,
+            auth_token: String,
+        ) -> Builder<LazyReplica> {
+            Builder {
+                inner: LazyReplica { path: todo!(), remote: todo!(), read_your_writes: todo!(), sync_interval: todo!() }
+            }
+        }
+    }
 }
 
 cfg_replication_or_remote_or_sync! {
@@ -531,6 +546,84 @@ cfg_replication! {
     }
 }
 
+pub struct LazyReplica {
+    path: std::path::PathBuf,
+    remote: Remote,
+    read_your_writes: bool,
+    sync_interval: Option<std::time::Duration>,
+}
+
+impl Builder<LazyReplica> {
+    /// Set weather you want writes to be visible locally before the write query returns. This
+    /// means that you will be able to read your own writes if this is set to `true`.
+    ///
+    /// # Default
+    ///
+    /// This defaults to `true`.
+    pub fn read_your_writes(mut self, read_your_writes: bool) -> Builder<LazyReplica> {
+        self.inner.read_your_writes = read_your_writes;
+        self
+    }
+
+    /// Set the duration at which the replicator will automatically call `sync` in the
+    /// background. The sync will continue for the duration that the resulted `Database`
+    /// type is alive for, once it is dropped the background task will get dropped and stop.
+    pub fn sync_interval(mut self, duration: std::time::Duration) -> Builder<LazyReplica> {
+        self.inner.sync_interval = Some(duration);
+        self
+    }
+
+    /// Build the remote embedded replica database.
+    pub async fn build(self) -> Result<Database> {
+        let LazyReplica {
+            path,
+            remote:
+                Remote {
+                    url,
+                    auth_token,
+                    connector,
+                    version,
+                },
+            read_your_writes,
+            sync_interval,
+        } = self.inner;
+
+        let connector = if let Some(connector) = connector {
+            connector
+        } else {
+            let https = super::connector()?;
+            use tower::ServiceExt;
+
+            let svc = https
+                .map_err(|e| e.into())
+                .map_response(|s| Box::new(s) as Box<dyn crate::util::Socket>);
+
+            crate::util::ConnectorService::new(svc)
+        };
+
+        let path = path
+            .to_str()
+            .ok_or(anyhow!("unable to convert path to string"))?
+            .to_owned();
+        Ok(Database {
+            db_type: DbType::Lazy {
+                db: crate::local::Database::open_local_lazy(
+                    connector.clone(),
+                    path,
+                    crate::OpenFlags::default(),
+                    url.clone(),
+                    auth_token.clone(),
+                )
+                .await?,
+                url,
+                auth_token,
+                connector,
+            },
+            max_write_replication_index: Default::default(),
+        })
+    }
+}
+
 cfg_sync! {
     /// Remote replica configuration type in [`Builder`].
     pub struct SyncedDatabase {
@@ -694,7 +787,7 @@ cfg_remote! {
 }
 
 cfg_replication_or_remote_or_sync! {
-    fn wrap_connector<C>(connector: C) -> crate::util::ConnectorService
+pub(crate) fn wrap_connector<C>(connector: C) -> crate::util::ConnectorService
     where
         C: tower::Service<http::Uri> + Send + Clone + Sync + 'static,
         C::Response: crate::util::Socket,

libsql/src/errors.rs

@@ -57,6 +57,8 @@ pub enum Error {
     InvalidBlobSize(usize),
     #[error("sync error: {0}")]
     Sync(crate::BoxError),
+    #[error("lazy error: {0}")]
+    Lazy(crate::BoxError),
     #[error("WAL frame insert conflict")]
     WalConflict,
 }
@@ -75,6 +77,13 @@ impl From<crate::sync::SyncError> for Error {
     }
 }
 
+#[cfg(feature = "lazy")]
+impl From<anyhow::Error> for Error {
+    fn from(e: anyhow::Error) -> Self {
+        Error::Lazy(e.into())
+    }
+}
+
 impl From<std::convert::Infallible> for Error {
     fn from(_: std::convert::Infallible) -> Self {
         unreachable!()