syncing between 2 databases

src/bin/dag.rs

@@ -1,16 +1,11 @@
-use std::{
-    path::{Path, PathBuf},
-    str::FromStr,
-};
+use std::path::{Path, PathBuf};
 
+use anyhow::Result;
 use clap::Parser;
 use dag_resolve::{
-    actor::{ActorId, ActorSecret},
-    event::{CoreContent, HashType, SignatureType},
-    resolvers::kv::{KVEventContent, KVResolver, KVState},
-    room::Room,
+    actor::{ActorId, ActorSecret}, event::{CoreContent, Event, HashType, SignatureType}, resolvers::kv::{KVEventContent, KVResolver, KVState}, room::Room, Error
 };
-use rusqlite::{Connection, OpenFlags};
+use rusqlite::Connection;
 
 #[derive(Parser)]
 #[clap(version, about)]
@@ -24,9 +19,13 @@ enum Cli {
         key: String,
         value: String,
     },
-}
 
-type Result<T> = std::result::Result<T, anyhow::Error>;
+    /// Synchronize two repositories
+    Sync {
+        repo: PathBuf,
+        other: PathBuf,
+    },
+}
 
 struct State {
     db: Connection,
@@ -36,12 +35,31 @@ struct State {
 
 impl State {
     fn create_event(&mut self, content: CoreContent<KVEventContent>) -> Result<()> {
-        let event = self.room.create_event(content, &self.secret)?;
-        self.db.execute(
-            "INSERT INTO events (id, json) VALUES (?, ?)",
-            (event.id().to_string(), serde_json::to_string(&event)?),
-        )?;
-        Ok(())
+        match self.room.create_event(content, &self.secret) {
+            Ok(event) => {
+                self.db.execute(
+                    "INSERT INTO events (id, json) VALUES (?, ?)",
+                    (event.id().to_string(), serde_json::to_string(&event)?),
+                )?;
+                Ok(())
+            },
+            Err(Error::AlreadyExists) => Ok(()),
+            Err(err) => Err(err.into()),
+        }
+    }
+
+    fn append_event(&mut self, event: Event<KVEventContent>) -> Result<()> {
+        match self.room.append_event(event) {
+            Ok(event) => {
+                self.db.execute(
+                    "INSERT INTO events (id, json) VALUES (?, ?)",
+                    (event.id().to_string(), serde_json::to_string(&event)?),
+                )?;
+                Ok(())
+            },
+            Err(Error::AlreadyExists) => Ok(()),
+            Err(err) => Err(err.into()),
+        }
     }
 }
 
@@ -129,15 +147,28 @@ fn get_repo(path: &Path, create: bool) -> Result<State> {
 
 fn main() -> Result<()> {
     let cli = Cli::parse();
+    let mut repo = match &cli {
+        Cli::Init { repo } => get_repo(repo, true)?,
+        Cli::Set { repo, .. } |
+        Cli::Sync { repo, .. } => get_repo(repo, false)?,
+    };
     match cli {
-        Cli::Init { repo } => {
-            get_repo(&repo, true)?;
-        },
-        Cli::Set { repo, key, value } => {
-            let mut state = get_repo(&repo, false)?;
-            state.create_event(CoreContent::Custom(KVEventContent::Set(key, value)))?;
-            dbg!(&state.room);
-            dbg!(&state.room.get_state());
+        Cli::Init { .. } => {}
+        Cli::Set { key, value, .. } => {
+            repo.create_event(CoreContent::Custom(KVEventContent::Set(key, value)))?;
+            dbg!(&repo.room);
+            dbg!(&repo.room.get_state());
+        }
+        Cli::Sync { other, .. } => {
+            let mut other = get_repo(&other, false)?;
+            for event in &repo.room.events {
+                other.append_event(event.clone())?;
+            }
+            for event in &other.room.events {
+                repo.append_event(event.clone())?;
+            }
+            dbg!(repo.room.get_state(), other.room.get_state());
+            dbg!(repo.room);
         }
     };
     Ok(())

src/error.rs

@@ -43,4 +43,9 @@ pub enum Error {
 
     #[error("Invalid starting sigil (char)")]
     InvalidSigil,
+
+    // TODO: remove, appending events should be idempotent
+    // but trying to get it to work returns lifetime errors?
+    #[error("Event already exists")]
+    AlreadyExists,
 }

src/proto/room.rs

@@ -77,14 +77,15 @@ impl<T: Debug + Serialize + Clone, S> Room<T, S> {
         self.append_event(event)
     }
 
-    pub fn append_event(&mut self, event: Event<T>) -> Result<&Event<T>> {
+    pub fn append_event<'a>(&'a mut self, event: Event<T>) -> Result<&'a Event<T>> {
         event.verify_room(self).expect("event failed verification");
-        if self.events.iter().any(|p| p == &event) {
-            panic!("already exists");
+        if let Some(_) = self.events.iter().find(|p| p == &&event) {
+            return Err(Error::AlreadyExists);
         }
         self.events.push(event);
         let event_ref = self.events.last().unwrap();
-        self.heads = vec![event_ref.id().clone()];
+        self.heads.retain(|id| !event_ref.references().contains(id));
+        self.heads.push(event_ref.id().clone());
         Ok(event_ref)
     }
 }