POC for live query

.gitignore

@@ -0,0 +1 @@
+/target

Cargo.toml

@@ -0,0 +1,10 @@
+[package]
+name = "bcup"
+version = "0.1.0"
+edition = "2021"
+
+[dependencies]
+futures = "0.3.30"
+serde = { version = "1.0.207", features = ["derive"] }
+surrealdb = "1.5.4"
+tokio = { version = "1.39.2", features = ["macros", "rt-multi-thread"] }

src/main.rs

@@ -0,0 +1,89 @@
+use futures::StreamExt;
+use serde::{Deserialize, Serialize};
+use surrealdb::{
+    engine::remote::ws::{Client, Ws},
+    opt::PatchOp,
+    sql::Thing,
+    Notification, Result, Surreal,
+};
+
+#[derive(Debug, Serialize, Deserialize)]
+struct State {
+    now: String,
+    required: String,
+}
+
+impl State {
+    fn is_pending(&self) -> bool {
+        self.now == "pending".as_ref()
+    }
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+struct ServerConfiguration {
+    id: Thing,
+    name: String,
+    domain: String,
+    directory: String,
+    volume: String,
+    repo_password: String,
+    rest_password: String,
+    state: State,
+}
+
+#[derive(Debug, Deserialize)]
+struct Record {
+    #[allow(dead_code)]
+    id: Thing,
+}
+
+#[tokio::main]
+async fn main() -> surrealdb::Result<()> {
+    let db = Surreal::new::<Ws>("127.0.0.1:8000").await?;
+
+    db.use_ns("test").use_db("test").await?;
+
+    println!("Prepared to listen");
+    let mut stream = db.select("rest_server").live().await?;
+    while let Some(result) = stream.next().await {
+        handle(result, &db).await;
+    }
+
+    Ok(())
+}
+async fn handle(result: Result<Notification<ServerConfiguration>>, db: &Surreal<Client>) {
+    println!("Something to handle");
+    match result {
+        Ok(notification) => {
+            let conf = &notification.data;
+            if conf.state.is_pending() {
+                process_new_item(conf, db).await;
+            }
+        }
+        Err(error) => eprintln!("{error}"),
+    }
+}
+
+async fn process_new_item(conf: &ServerConfiguration, db: &Surreal<Client>) {
+    println!("The following item has to be processed {:?}", conf);
+    spawn_restic_container(conf.directory.clone());
+    add_proxy_configuration(conf.directory.clone());
+    mark_data_as_processed(conf, db).await;
+}
+
+async fn mark_data_as_processed(conf: &ServerConfiguration, db: &Surreal<Client>) {
+    println!("Updating status on item id:{:?}", conf.id);
+    let updated: Option<Record> = db
+        .update(&conf.id)
+        .patch(PatchOp::replace("/state/now", "processed"))
+        .await
+        .unwrap();
+    dbg!(updated);
+}
+
+fn spawn_restic_container(path: String) {
+    println!("[fake] Spawning container for path {:?}", path);
+}
+fn add_proxy_configuration(path: String) {
+    println!("[fake] Creating caddy configuration {:?}", path);
+}