Cleanup

Author: Nao-ris

examples/README.md

@@ -12,7 +12,7 @@ The configuration files that might interest you are:
 - `docker-conf/db-config.json` -> contains the database configuration
 - `docker-conf/sync-function.js` -> contains the sync function used by the Sync Gateway
 
-To start both the Sync Gatewawy and Couchbase Server, move to `docker-conf` through a terminal and use:
+To start both the Sync Gateway and Couchbase Server, move to `docker-conf` through a terminal and use:
 
 ```shell
 $ docker-compose up
@@ -44,16 +44,11 @@ $ curl -XPUT -v "http://localhost:4985/my-db/" -H 'Content-Type: application/jso
 
 ## Running an example
 
-As of now, there is only one example: `sgw_1_cblite`.
+As of now, there is only one example: `ticket_70596`.
 
 It can be run with the following command:
 ```shell
-$ cargo run --features=enterprise --example sgw_1_cblite
+$ cargo run --features=enterprise --example ticket_70596
 ```
 
-What it does:
-- Create a cblite database `test1`
-- Add a user `great_name` to the Sync Gateway
-- Retrieve a session token for the user `great_name` from the Sync Gateway
-- Start a continuous push & pull replicator
-- Create a document, then wait for 5 seconds for the replication to finish
+There are utility functions available to interact with the Sync Gateway or Couchbase Server, feel free to add more if needed.

examples/sgw_1_cblite.rs

@@ -0,0 +1,116 @@
+mod utils;
+
+use std::path::Path;
+use couchbase_lite::*;
+use utils::*;
+
+fn main() {
+    let mut db = Database::open(
+        "test1",
+        Some(DatabaseConfiguration {
+            directory: Path::new("./"),
+            #[cfg(feature = "enterprise")]
+            encryption_key: None,
+        }),
+    )
+    .unwrap();
+
+    add_or_update_user("great_name", vec!["channel1".into()]);
+    let session_token = get_session("great_name");
+    println!("Sync gateway session token: {session_token}");
+
+    let mut repl =
+        setup_replicator(db.clone(), session_token).add_document_listener(Box::new(doc_listener));
+
+    repl.start(false);
+
+    std::thread::sleep(std::time::Duration::from_secs(3));
+
+    // Auto-purge test scenario from support ticket https://support.couchbase.com/hc/en-us/requests/70596?page=1
+    // Testing if documents pushed to inaccessible channels get auto-purged
+    create_doc(&mut db, "doc1", "channel1");
+    create_doc(&mut db, "doc2", "channel2");
+
+    std::thread::sleep(std::time::Duration::from_secs(10));
+    assert!(get_doc(&db, "doc1").is_ok());
+    assert!(get_doc(&db, "doc2").is_ok()); // This looks buggy
+
+    change_channel(&mut db, "doc1", "channel2");
+
+    std::thread::sleep(std::time::Duration::from_secs(10));
+    assert!(get_doc(&db, "doc1").is_err());
+
+    repl.stop(None);
+}
+
+fn create_doc(db: &mut Database, id: &str, channel: &str) {
+    let mut doc = Document::new_with_id(id);
+    doc.set_properties_as_json(
+        &serde_json::json!({
+            "channels": channel,
+        })
+        .to_string(),
+    )
+    .unwrap();
+    db.save_document(&mut doc).unwrap();
+
+    println!(
+        "Created doc {id} with content: {}",
+        doc.properties_as_json()
+    );
+}
+
+fn get_doc(db: &Database, id: &str) -> Result<Document> {
+    db.get_document(id)
+}
+
+fn change_channel(db: &mut Database, id: &str, channel: &str) {
+    let mut doc = get_doc(db, id).unwrap();
+    let mut prop = doc.mutable_properties();
+    prop.at("channels").put_string(channel);
+    let _ = db.save_document(&mut doc);
+    println!(
+        "Changed doc {id} with content: {}",
+        doc.properties_as_json()
+    );
+}
+
+fn setup_replicator(db: Database, session_token: String) -> Replicator {
+    let repl_conf = ReplicatorConfiguration {
+        database: Some(db.clone()),
+        endpoint: Endpoint::new_with_url(SYNC_GW_URL).unwrap(),
+        replicator_type: ReplicatorType::PushAndPull,
+        continuous: true,
+        disable_auto_purge: false,
+        max_attempts: 3,
+        max_attempt_wait_time: 1,
+        heartbeat: 60,
+        authenticator: None,
+        proxy: None,
+        headers: vec![(
+            "Cookie".to_string(),
+            format!("SyncGatewaySession={session_token}"),
+        )]
+        .into_iter()
+        .collect(),
+        pinned_server_certificate: None,
+        trusted_root_certificates: None,
+        channels: MutableArray::default(),
+        document_ids: MutableArray::default(),
+        collections: None,
+        accept_parent_domain_cookies: false,
+        #[cfg(feature = "enterprise")]
+        accept_only_self_signed_server_certificate: false,
+    };
+    let repl_context = ReplicationConfigurationContext::default();
+    Replicator::new(repl_conf, Box::new(repl_context)).unwrap()
+}
+
+fn doc_listener(direction: Direction, documents: Vec<ReplicatedDocument>) {
+    println!("=== Document(s) replicated ===");
+    println!("Direction: {direction:?}");
+    for document in documents {
+        println!("Document: {document:?}");
+    }
+    println!("===");
+}

examples/ticket_70596.rs

@@ -0,0 +1,64 @@
+use crate::utils::constants::*;
+
+pub fn purge_doc_from_cbs(doc_id: &str) {
+    let url = format!("{CBS_URL}/pools/default/buckets/{CBS_BUCKET}/docs/{doc_id}");
+    let response = reqwest::blocking::Client::new()
+        .delete(&url)
+        .basic_auth(CBS_ADMIN_USER, Some(CBS_ADMIN_PWD))
+        .send();
+
+    match response {
+        Ok(resp) => {
+            let status = resp.status();
+            if let Ok(body) = resp.text() {
+                println!("Purge {doc_id} from CBS: status={status}, body={body}");
+            } else {
+                println!("Purge {doc_id} from CBS: status={status}");
+            }
+        }
+        Err(e) => println!("Purge {doc_id} from CBS error: {e}"),
+    }
+}
+
+pub fn compact_cbs_bucket() {
+    let url = format!("{CBS_URL}/pools/default/buckets/{CBS_BUCKET}/controller/compactBucket");
+    let response = reqwest::blocking::Client::new()
+        .post(&url)
+        .basic_auth(CBS_ADMIN_USER, Some(CBS_ADMIN_PWD))
+        .send();
+
+    match response {
+        Ok(resp) => {
+            let status = resp.status();
+            if let Ok(body) = resp.text() {
+                println!("Compact CBS bucket: status={status}, body={body}");
+            } else {
+                println!("Compact CBS bucket: status={status}");
+            }
+        }
+        Err(e) => println!("Compact CBS bucket error: {e}"),
+    }
+}
+
+pub fn check_doc_in_cbs(doc_id: &str) {
+    let url = format!("{CBS_URL}:8093/query/service");
+    let query = format!(
+        "SELECT META().id, META().deleted FROM `{CBS_BUCKET}` WHERE META().id = '{doc_id}'"
+    );
+    let body = serde_json::json!({"statement": query});
+
+    let response = reqwest::blocking::Client::new()
+        .post(&url)
+        .basic_auth(CBS_ADMIN_USER, Some(CBS_ADMIN_PWD))
+        .json(&body)
+        .send();
+
+    match response {
+        Ok(resp) => {
+            if let Ok(text) = resp.text() {
+                println!("CBS check for {doc_id}: {text}");
+            }
+        }
+        Err(e) => println!("CBS check error: {e}"),
+    }
+}

examples/utils/cbs_admin.rs

@@ -0,0 +1,6 @@
+pub const SYNC_GW_URL_ADMIN: &str = "http://localhost:4985/my-db";
+pub const SYNC_GW_URL: &str = "ws://localhost:4984/my-db";
+pub const CBS_URL: &str = "http://localhost:8091";
+pub const CBS_BUCKET: &str = "my-bucket";
+pub const CBS_ADMIN_USER: &str = "cb_admin";
+pub const CBS_ADMIN_PWD: &str = "cb_admin_pwd";

examples/utils/constants.rs

@@ -0,0 +1,8 @@
+pub mod cbs_admin;
+pub mod constants;
+pub mod sgw_admin;
+
+// Re-export commonly used functions
+pub use constants::*;
+pub use sgw_admin::*;
+pub use cbs_admin::*;