feat: add builder option to select protocol

Author: levydsa

libsql/src/database.rs

@@ -63,6 +63,14 @@ cfg_replication_or_sync! {
             self.frames_synced
         }
     }
+
+    #[derive(Default)]
+    pub enum SyncProtocol {
+        #[default]
+        Auto,
+        V1,
+        V2,
+    }
 }
 
 enum DbType {

libsql/src/database/builder.rs

@@ -67,6 +67,7 @@ impl Builder<()> {
                     http_request_callback: None,
                     namespace: None,
                     skip_safety_assert: false,
+                    sync_protocol: Default::default(),
                 },
             }
         }
@@ -222,6 +223,7 @@ cfg_replication! {
         http_request_callback: Option<crate::util::HttpRequestCallback>,
         namespace: Option<String>,
         skip_safety_assert: bool,
+        sync_protocol: super::SyncProtocol,
     }
 
     /// Local replica configuration type in [`Builder`].
@@ -274,6 +276,14 @@ cfg_replication! {
             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_protocol(mut self, protocol: super::SyncProtocol) -> Builder<RemoteReplica> {
+            self.inner.sync_protocol = protocol;
+            self
+        }
+
         pub fn http_request_callback<F>(mut self, f: F) -> Builder<RemoteReplica>
         where
             F: Fn(&mut http::Request<()>) + Send + Sync + 'static
@@ -326,7 +336,8 @@ cfg_replication! {
                 sync_interval,
                 http_request_callback,
                 namespace,
-                skip_safety_assert
+                skip_safety_assert,
+                sync_protocol,
             } = self.inner;
 
             let connector = if let Some(connector) = connector {
@@ -342,26 +353,44 @@ cfg_replication! {
                 crate::util::ConnectorService::new(svc)
             };
 
-            let client = hyper::client::Client::builder()
-                .build::<_, hyper::Body>(connector.clone());
-
-            let req = http::Request::get(format!("{url}/sync/0/0/0"))
-                .header("Authorization", format!("Bearer {}", auth_token))
-                .body(hyper::Body::empty())
-                .unwrap();
-
-            let res = client
-                .request(req)
-                .await
-                .map_err(|err| crate::Error::Sync(err.into()))?;
-
-            match res.status() {
-                hyper::StatusCode::OK => return Builder::new_synced_database(path, url, auth_token)
-                    .remote_writes(true)
-                    .read_your_writes(read_your_writes)
-                    .build()
-                    .await,
-                _ => {}
+            use super::SyncProtocol;
+
+            match sync_protocol {
+                p @ (SyncProtocol::Auto | SyncProtocol::V2) => {
+                    let client = hyper::client::Client::builder()
+                        .build::<_, hyper::Body>(connector.clone());
+
+                    let req = http::Request::get(format!("{url}/sync/0/0/0"))
+                        .header("Authorization", format!("Bearer {}", auth_token))
+                        .body(hyper::Body::empty())
+                        .unwrap();
+
+                    let res = client
+                        .request(req)
+                        .await
+                        .map_err(|err| crate::Error::Sync(err.into()))?;
+
+                    if matches!(p, SyncProtocol::V2) {
+                        if !res.status().is_success() {
+                            let status = res.status();
+                            let body_bytes = hyper::body::to_bytes(res.into_body())
+                                .await
+                                .map_err(|err| crate::Error::Sync(err.into()))?;
+                            let error_message = String::from_utf8_lossy(&body_bytes);
+                            return Err(crate::Error::Sync(format!("HTTP error {}: {}", status, error_message).into()));
+                        }
+                    }
+
+                    if res.status().is_success() {
+                        return Builder::new_synced_database(path, url, auth_token)
+                            .remote_writes(true)
+                            .read_your_writes(read_your_writes)
+                            .build()
+                            .await;
+                    }
+
+                }
+                SyncProtocol::V1 => {}
             }
 
             let path = path.to_str().ok_or(crate::Error::InvalidUTF8Path)?.to_owned();

libsql/src/lib.rs

@@ -135,6 +135,7 @@ cfg_sync! {
 
 cfg_replication! {
     pub mod replication;
+    pub use database::SyncProtocol;
 }
 
 cfg_core! {