lmdb: event_relationships: Create Database, Migration to process all event relationships

src/relationship.rs

@@ -1,7 +1,8 @@
 use nostr_types::MilliSatoshi;
+use speedy::{Readable, Writable};
 
 /// A relationship between events
-#[derive(Clone, Debug, PartialEq, Eq)]
+#[derive(Clone, Debug, PartialEq, Eq, Readable, Writable)]
 pub enum Relationship {
     //Root,
     Reply,

src/storage/migrations/mod.rs

@@ -0,0 +1,58 @@
+use super::Storage;
+use crate::error::{Error, ErrorKind};
+use lmdb::{Cursor, Transaction};
+use nostr_types::Event;
+use speedy::Readable;
+
+impl Storage {
+    const MIGRATION_LEVEL: u32 = 1;
+
+    pub(super) fn migrate(&self, mut level: u32) -> Result<(), Error> {
+        if level > Self::MIGRATION_LEVEL {
+            return Err(ErrorKind::General(format!(
+                "Migration level {} unknown: This client is older than your data.",
+                level
+            )).into());
+        }
+
+        while level < Self::MIGRATION_LEVEL {
+            level += 1;
+            tracing::info!("LMDB Migration to level {}...", level);
+
+            self.migrate_inner(level)?;
+
+            self.write_migration_level(level)?;
+        }
+
+        Ok(())
+    }
+
+    fn migrate_inner(&self, level: u32) -> Result<(), Error> {
+        match level {
+            0 => Ok(()),
+            1 => self.compute_relationships(),
+            n => panic!("Unknown migration level {}", n),
+        }
+    }
+
+    // Load and process every event in order to generate the relationships data
+    fn compute_relationships(&self) -> Result<(), Error> {
+        panic!("Not yet properly implemented");
+
+        let txn = self.env.begin_ro_txn()?;
+        let mut cursor = txn.open_ro_cursor(self.events)?;
+        let iter = cursor.iter_start();
+        for result in iter {
+            match result {
+                Err(e) => return Err(e.into()),
+                Ok((_key, val)) => {
+                    let event = Event::read_from_buffer(val)?;
+                    // FIXME we can't do this async
+                    // crate::process::process_new_event(&event, false, None, None).await?;
+                }
+            }
+        }
+
+        Ok(())
+    }
+}

src/storage/mod.rs

@@ -1,9 +1,11 @@
 mod import;
+mod migrations;
 
 use crate::db::DbRelay;
 use crate::error::{Error, ErrorKind};
 use crate::globals::GLOBALS;
 use crate::profile::Profile;
+use crate::relationship::Relationship;
 use crate::settings::Settings;
 use lmdb::{
     Cursor, Database, DatabaseFlags, Environment, EnvironmentFlags, Stat, Transaction, WriteFlags,
@@ -49,6 +51,9 @@ pub struct Storage {
 
     // Id -> Event
     events: Database,
+
+    // Id:Id -> Relationship
+    relationships: Database,
 }
 
 impl Storage {
@@ -91,6 +96,8 @@ impl Storage {
 
         let events = env.create_db(Some("events"), DatabaseFlags::empty())?;
 
+        let relationships = env.create_db(Some("relationships"), DatabaseFlags::empty())?;
+
         let storage = Storage {
             env,
             general,
@@ -100,6 +107,7 @@ impl Storage {
             relays,
             event_tags,
             events,
+            relationships,
         };
 
         // If migration level is missing, we need to import from legacy sqlite
@@ -108,8 +116,8 @@ impl Storage {
                 // Import from sqlite
                 storage.import()?;
             }
-            Some(_level) => {
-                // migrations happen here
+            Some(level) => {
+                storage.migrate(level)?;
             }
         }
 
@@ -714,4 +722,41 @@ impl Storage {
             Ok(Some(event.id))
         }
     }
+
+    pub fn write_relationship(
+        &self,
+        id: Id,
+        related: Id,
+        relationship: Relationship,
+    ) -> Result<(), Error> {
+        let mut key = id.as_ref().as_slice().to_owned();
+        key.extend(related.as_ref());
+        let value = relationship.write_to_vec()?;
+        let mut txn = self.env.begin_rw_txn()?;
+        txn.put(self.relationships, &key, &value, WriteFlags::empty())?;
+        txn.commit()?;
+        Ok(())
+    }
+
+    pub fn find_relationships(&self, id: Id) -> Result<Vec<(Id, Relationship)>, Error> {
+        let start_key = id.as_ref().to_owned();
+        let txn = self.env.begin_ro_txn()?;
+        let mut cursor = txn.open_ro_cursor(self.relationships)?;
+        let iter = cursor.iter_from(start_key);
+        let mut output: Vec<(Id, Relationship)> = Vec::new();
+        for result in iter {
+            match result {
+                Err(e) => return Err(e.into()),
+                Ok((key, val)) => {
+                    if !key.starts_with(&start_key) {
+                        break;
+                    }
+                    let id2 = Id(key[32..64].try_into().unwrap());
+                    let relationship = Relationship::read_from_buffer(val)?;
+                    output.push((id2, relationship));
+                }
+            }
+        }
+        Ok(output)
+    }
 }