add OpenMessageRepository

Author: ghubertpalo

Cargo.lock

@@ -32,6 +32,7 @@ tar = "0.4.38"
 thiserror = "1.0.31"
 tokio = { version = "1.17.0", features = ["full"] }
 tokio-util = { version = "0.7.1", features = ["codec"] }
+uuid = { version = "1.3.0", features = ["v4", "fast-rng", "macro-diagnostics"] }
 warp = "0.3"
 
 [dev-dependencies]

mithril-aggregator/Cargo.toml

@@ -1,6 +1,7 @@
 use chrono::NaiveDateTime;
 use mithril_common::entities::Beacon;
 use mithril_common::entities::Epoch;
+use mithril_common::StdError;
 
 use mithril_common::sqlite::Provider;
 use mithril_common::sqlite::SourceAlias;
@@ -10,18 +11,33 @@ use mithril_common::{
 };
 use sqlite::Row;
 use sqlite::{Connection, Value};
+use uuid::Uuid;
+
+type StdResult<T> = Result<T, StdError>;
 
 /// ## OpenMessage
 ///
 /// An open message is a message open for signatures. Every signer may send a
 /// single signature for this message from which a multi signature will be
 /// generated if possible.
-struct OpenMessage {
-    open_message_id: String,
+pub struct OpenMessage {
+    /// OpenMessage unique identifier
+    open_message_id: Uuid,
+
+    /// Epoch
     epoch: Epoch,
+
+    /// Beacon, this is the discriminant of this message type in the current
+    /// Epoch
     beacon: Beacon,
+
+    /// Type of message
     signed_entity_type: SignedEntityType,
+
+    /// Message content
     message: String,
+
+    /// Message creation datetime, it is set by the database.
     created_at: NaiveDateTime,
 }
 
@@ -31,6 +47,11 @@ impl SqLiteEntity for OpenMessage {
         Self: Sized,
     {
         let open_message_id = row.get::<String, _>(0);
+        let open_message_id = Uuid::parse_str(&open_message_id).map_err(|e| {
+            HydrationError::InvalidData(format!(
+                "Invalid UUID in open_message.open_message_id: '{open_message_id}'. Error: {e}"
+            ))
+        })?;
         let message = row.get::<String, _>(4);
         let epoch_settings_id = row.get::<i64, _>(1);
         let epoch_val = u64::try_from(epoch_settings_id)
@@ -96,13 +117,13 @@ struct OpenMessageProvider<'client> {
 }
 
 impl<'client> OpenMessageProvider<'client> {
-    fn new(connection: &'client Connection) -> Self {
+    pub fn new(connection: &'client Connection) -> Self {
         Self { connection }
     }
 
     fn get_epoch_condition(&self, epoch: Epoch) -> WhereCondition {
         WhereCondition::new(
-            "{:open_message:}.epoch_settings_id = ?*",
+            "epoch_settings_id = ?*",
             vec![Value::Integer(epoch.0 as i64)],
         )
     }
@@ -112,14 +133,14 @@ impl<'client> OpenMessageProvider<'client> {
         signed_entity_type: &SignedEntityType,
     ) -> WhereCondition {
         WhereCondition::new(
-            "{:open_message:}.signed_entity_type_id = ?*",
+            "signed_entity_type_id = ?*",
             vec![Value::Integer(*signed_entity_type as i64)],
         )
     }
 
     fn get_open_message_id_condition(&self, open_message_id: &str) -> WhereCondition {
         WhereCondition::new(
-            "{:open_message:}.open_message_id = ?*",
+            "open_message_id = ?*",
             vec![Value::String(open_message_id.to_owned())],
         )
     }
@@ -140,6 +161,128 @@ impl<'client> Provider<'client> for OpenMessageProvider<'client> {
     }
 }
 
+struct InsertOpenMessageProvider<'client> {
+    connection: &'client Connection,
+}
+impl<'client> InsertOpenMessageProvider<'client> {
+    pub fn new(connection: &'client Connection) -> Self {
+        Self { connection }
+    }
+
+    fn get_insert_condition(
+        &self,
+        epoch: Epoch,
+        beacon: &Beacon,
+        signed_entity_type: &SignedEntityType,
+        message: &str,
+    ) -> StdResult<WhereCondition> {
+        let expression = "(open_message_id, epoch_settings_id, beacon, signed_entity_type_id, message) values (?*, ?*, ?*, ?*, ?*)";
+        let parameters = vec![
+            Value::String(Uuid::new_v4().to_string()),
+            Value::Integer(epoch.0 as i64),
+            Value::String(serde_json::to_string(beacon)?),
+            Value::Integer(*signed_entity_type as i64),
+            Value::String(message.to_string()),
+        ];
+
+        Ok(WhereCondition::new(expression, parameters))
+    }
+}
+
+impl<'client> Provider<'client> for InsertOpenMessageProvider<'client> {
+    type Entity = OpenMessage;
+
+    fn get_connection(&'client self) -> &'client Connection {
+        self.connection
+    }
+
+    fn get_definition(&self, condition: &str) -> String {
+        let aliases = SourceAlias::new(&[("{:open_message:}", "open_message")]);
+        let projection = Self::Entity::get_projection().expand(aliases);
+
+        format!("insert into open_message {condition} returning {projection}")
+    }
+}
+
+struct DeleteOpenMessageProvider<'client> {
+    connection: &'client Connection,
+}
+
+impl<'client> DeleteOpenMessageProvider<'client> {
+    pub fn new(connection: &'client Connection) -> Self {
+        Self { connection }
+    }
+
+    fn get_epoch_condition(&self, epoch: Epoch) -> WhereCondition {
+        WhereCondition::new(
+            "epoch_settings_id = ?*",
+            vec![Value::Integer(epoch.0 as i64)],
+        )
+    }
+}
+
+impl<'client> Provider<'client> for DeleteOpenMessageProvider<'client> {
+    type Entity = OpenMessage;
+
+    fn get_connection(&'client self) -> &'client Connection {
+        self.connection
+    }
+
+    fn get_definition(&self, condition: &str) -> String {
+        let aliases = SourceAlias::new(&[("{:open_message:}", "open_message")]);
+        let projection = Self::Entity::get_projection().expand(aliases);
+
+        format!("delete from open_message where {condition} returning {projection}")
+    }
+}
+
+pub struct OpenMessageRepository<'client> {
+    connection: &'client Connection,
+}
+
+impl<'client> OpenMessageRepository<'client> {
+    /// Return the latest [OpenMessage] for the given Epoch and [SignedEntityType].
+    pub fn get_open_message(
+        &self,
+        epoch: Epoch,
+        signed_entity_type: &SignedEntityType,
+    ) -> StdResult<Option<OpenMessage>> {
+        let provider = OpenMessageProvider::new(self.connection);
+        let filters = provider
+            .get_epoch_condition(epoch)
+            .and_where(provider.get_signed_entity_type_condition(signed_entity_type));
+        let mut messages = provider.find(filters)?;
+
+        Ok(messages.next())
+    }
+
+    /// Create a new [OpenMessage] in the database.
+    pub fn create_open_message(
+        &self,
+        epoch: Epoch,
+        beacon: &Beacon,
+        signed_entity_type: &SignedEntityType,
+        message: &str,
+    ) -> StdResult<OpenMessage> {
+        let provider = InsertOpenMessageProvider::new(self.connection);
+        let filters = provider.get_insert_condition(epoch, beacon, signed_entity_type, message)?;
+        let mut cursor = provider.find(filters)?;
+
+        cursor
+            .next()
+            .ok_or_else(|| panic!("Inserting an open_message should not return nothing."))
+    }
+
+    /// Remove all the [OpenMessage] for the given Epoch in the database.
+    pub fn clean_epoch(&self, epoch: Epoch) -> StdResult<()> {
+        let provider = DeleteOpenMessageProvider::new(self.connection);
+        let filters = provider.get_epoch_condition(epoch);
+        let _ = provider.find(filters)?;
+
+        Ok(())
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use mithril_common::sqlite::SourceAlias;
@@ -163,7 +306,7 @@ mod tests {
         let provider = OpenMessageProvider::new(&connection);
         let (expr, params) = provider.get_epoch_condition(Epoch(12)).expand();
 
-        assert_eq!("{:open_message:}.epoch_settings_id = ?1".to_string(), expr);
+        assert_eq!("epoch_settings_id = ?1".to_string(), expr);
         assert_eq!(vec![Value::Integer(12)], params,);
     }
 
@@ -175,10 +318,7 @@ mod tests {
             .get_signed_entity_type_condition(&SignedEntityType::CardanoImmutableFilesFull)
             .expand();
 
-        assert_eq!(
-            "{:open_message:}.signed_entity_type_id = ?1".to_string(),
-            expr
-        );
+        assert_eq!("signed_entity_type_id = ?1".to_string(), expr);
         assert_eq!(vec![Value::Integer(2)], params,);
     }
 
@@ -190,12 +330,46 @@ mod tests {
             .get_open_message_id_condition("cecd7983-8b3a-42b1-b778-6d75e87828ee")
             .expand();
 
-        assert_eq!("{:open_message:}.open_message_id = ?1".to_string(), expr);
+        assert_eq!("open_message_id = ?1".to_string(), expr);
         assert_eq!(
             vec![Value::String(
                 "cecd7983-8b3a-42b1-b778-6d75e87828ee".to_string()
             )],
             params,
         );
     }
+
+    #[test]
+    fn insert_provider_condition() {
+        let connection = Connection::open(":memory:").unwrap();
+        let provider = InsertOpenMessageProvider::new(&connection);
+        let (expr, params) = provider
+            .get_insert_condition(
+                Epoch(12),
+                &Beacon::default(),
+                &SignedEntityType::CardanoStakeDistribution,
+                "This is a message",
+            )
+            .unwrap()
+            .expand();
+
+        assert_eq!("(open_message_id, epoch_settings_id, beacon, signed_entity_type_id, message) values (?1, ?2, ?3, ?4, ?5)".to_string(), expr);
+        assert_eq!(Value::Integer(12), params[1]);
+        assert_eq!(
+            Value::String(r#"{"network":"","epoch":0,"immutable_file_number":0}"#.to_string()),
+            params[2]
+        );
+        assert_eq!(Value::Integer(1), params[3]);
+        assert_eq!(Value::String("This is a message".to_string()), params[4]);
+    }
+
+    #[test]
+    fn delete_provider_epoch_condition() {
+        let connection = Connection::open(":memory:").unwrap();
+        let provider = DeleteOpenMessageProvider::new(&connection);
+        let (expr, params) = provider.get_epoch_condition(Epoch(12)).expand();
+
+        assert_eq!("epoch_settings_id = ?1".to_string(), expr);
+        assert_eq!(vec![Value::Integer(12)], params,);
+    }
 }