add foreign key support to sqlite

ghubertpalo · ghubertpalo · commit f6d97e9cbc9a · 2023-04-03T11:15:08.000+02:00
diff --git a/mithril-aggregator/src/database/migration.rs b/mithril-aggregator/src/database/migration.rs
@@ -128,6 +128,24 @@ insert into certificate (certificate_id,
     from certificate_temp as c;
 create index epoch_index ON certificate(epoch);
 drop table certificate_temp;
+"#,
+        ),
+        // Migration 5
+        // Add the `open_message` table
+        SqlMigration::new(
+            5,
+            r#"
+create table open_message (
+	open_message_id         text    not null,
+    epoch_setting_id        int     not null,
+    beacon                  text    not null,
+    signed_entity_type_id   int     not null,
+    message                 text    not null,
+    created_at              text    not null default current_timestamp,
+    primary key (open_message_id),
+    foreign key (epoch_setting_id)     references epoch_setting (epoch_setting_id),
+    foreign key (signed_entity_type_id) references signed_entity_type (signed_entity_type_id)
+);
 "#,
         ),
     ]
diff --git a/mithril-aggregator/src/database/provider/mod.rs b/mithril-aggregator/src/database/provider/mod.rs
@@ -6,4 +6,5 @@ mod stake_pool;
 
 pub use certificate::*;
 pub use epoch_setting::*;
+pub use open_message::*;
 pub use stake_pool::*;
diff --git a/mithril-aggregator/src/database/provider/open_message.rs b/mithril-aggregator/src/database/provider/open_message.rs
@@ -1,16 +1,18 @@
-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;
 use mithril_common::{
-    entities::SignedEntityType,
+    entities::{Beacon, Epoch, SignedEntityType},
     sqlite::{HydrationError, Projection, SqLiteEntity, WhereCondition},
+    sqlite::{Provider, SourceAlias},
 };
+
+use chrono::NaiveDateTime;
 use sqlite::Row;
 use sqlite::{Connection, Value};
+
+use std::sync::Arc;
+
+use tokio::sync::Mutex;
 use uuid::Uuid;
 
 type StdResult<T> = Result<T, StdError>;
@@ -20,6 +22,7 @@ type StdResult<T> = Result<T, StdError>;
 /// 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.
+#[allow(dead_code)]
 pub struct OpenMessage {
     /// OpenMessage unique identifier
     open_message_id: Uuid,
@@ -53,9 +56,9 @@ impl SqLiteEntity for OpenMessage {
             ))
         })?;
         let message = row.get::<String, _>(4);
-        let epoch_settings_id = row.get::<i64, _>(1);
-        let epoch_val = u64::try_from(epoch_settings_id)
-            .map_err(|e| panic!("Integer field open_message.epoch_settings_id (value={epoch_settings_id}) is incompatible with u64 Epoch representation. Error = {e}"))?;
+        let epoch_setting_id = row.get::<i64, _>(1);
+        let epoch_val = u64::try_from(epoch_setting_id)
+            .map_err(|e| panic!("Integer field open_message.epoch_setting_id (value={epoch_setting_id}) is incompatible with u64 Epoch representation. Error = {e}"))?;
 
         let signed_entity_type_id = usize::try_from(row.get::<i64, _>(3)).map_err(|e| {
             panic!(
@@ -94,10 +97,14 @@ impl SqLiteEntity for OpenMessage {
 
     fn get_projection() -> Projection {
         Projection::from(&[
-            ("open_message_id", "{:open_message:}.open_message_id", "int"),
             (
-                "epoch_settings_id",
-                "{:open_message:}.epoch_settings_id",
+                "open_message_id",
+                "{:open_message:}.open_message_id",
+                "text",
+            ),
+            (
+                "epoch_setting_id",
+                "{:open_message:}.epoch_setting_id",
                 "int",
             ),
             ("beacon", "{:open_message:}.beacon", "text"),
@@ -123,7 +130,7 @@ impl<'client> OpenMessageProvider<'client> {
 
     fn get_epoch_condition(&self, epoch: Epoch) -> WhereCondition {
         WhereCondition::new(
-            "epoch_settings_id = ?*",
+            "epoch_setting_id = ?*",
             vec![Value::Integer(epoch.0 as i64)],
         )
     }
@@ -138,6 +145,8 @@ impl<'client> OpenMessageProvider<'client> {
         )
     }
 
+    // Useful in test and probably in the future.
+    #[allow(dead_code)]
     fn get_open_message_id_condition(&self, open_message_id: &str) -> WhereCondition {
         WhereCondition::new(
             "open_message_id = ?*",
@@ -176,7 +185,7 @@ impl<'client> InsertOpenMessageProvider<'client> {
         signed_entity_type: &SignedEntityType,
         message: &str,
     ) -> StdResult<WhereCondition> {
-        let expression = "(open_message_id, epoch_settings_id, beacon, signed_entity_type_id, message) values (?*, ?*, ?*, ?*, ?*)";
+        let expression = "(open_message_id, epoch_setting_id, beacon, signed_entity_type_id, message) values (?*, ?*, ?*, ?*, ?*)";
         let parameters = vec![
             Value::String(Uuid::new_v4().to_string()),
             Value::Integer(epoch.0 as i64),
@@ -215,7 +224,7 @@ impl<'client> DeleteOpenMessageProvider<'client> {
 
     fn get_epoch_condition(&self, epoch: Epoch) -> WhereCondition {
         WhereCondition::new(
-            "epoch_settings_id = ?*",
+            "epoch_setting_id = ?*",
             vec![Value::Integer(epoch.0 as i64)],
         )
     }
@@ -236,18 +245,28 @@ impl<'client> Provider<'client> for DeleteOpenMessageProvider<'client> {
     }
 }
 
-pub struct OpenMessageRepository<'client> {
-    connection: &'client Connection,
+/// ## Open message repository
+///
+/// This is a business oriented layer to perform actions on the database through
+/// providers.
+pub struct OpenMessageRepository {
+    connection: Arc<Mutex<Connection>>,
 }
 
-impl<'client> OpenMessageRepository<'client> {
+impl OpenMessageRepository {
+    /// Instanciate service
+    pub fn new(connection: Arc<Mutex<Connection>>) -> Self {
+        Self { connection }
+    }
+
     /// Return the latest [OpenMessage] for the given Epoch and [SignedEntityType].
-    pub fn get_open_message(
+    pub async fn get_open_message(
         &self,
         epoch: Epoch,
         signed_entity_type: &SignedEntityType,
     ) -> StdResult<Option<OpenMessage>> {
-        let provider = OpenMessageProvider::new(self.connection);
+        let lock = self.connection.lock().await;
+        let provider = OpenMessageProvider::new(&lock);
         let filters = provider
             .get_epoch_condition(epoch)
             .and_where(provider.get_signed_entity_type_condition(signed_entity_type));
@@ -257,14 +276,15 @@ impl<'client> OpenMessageRepository<'client> {
     }
 
     /// Create a new [OpenMessage] in the database.
-    pub fn create_open_message(
+    pub async fn create_open_message(
         &self,
         epoch: Epoch,
         beacon: &Beacon,
         signed_entity_type: &SignedEntityType,
         message: &str,
     ) -> StdResult<OpenMessage> {
-        let provider = InsertOpenMessageProvider::new(self.connection);
+        let lock = self.connection.lock().await;
+        let provider = InsertOpenMessageProvider::new(&lock);
         let filters = provider.get_insert_condition(epoch, beacon, signed_entity_type, message)?;
         let mut cursor = provider.find(filters)?;
 
@@ -274,19 +294,23 @@ impl<'client> OpenMessageRepository<'client> {
     }
 
     /// 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);
+    /// It returns the number of messages removed.
+    pub async fn clean_epoch(&self, epoch: Epoch) -> StdResult<usize> {
+        let lock = self.connection.lock().await;
+        let provider = DeleteOpenMessageProvider::new(&lock);
         let filters = provider.get_epoch_condition(epoch);
-        let _ = provider.find(filters)?;
+        let cursor = provider.find(filters)?;
 
-        Ok(())
+        Ok(cursor.count())
     }
 }
 
 #[cfg(test)]
 mod tests {
     use mithril_common::sqlite::SourceAlias;
 
+    use crate::{dependency_injection::DependenciesBuilder, Configuration};
+
     use super::*;
 
     #[test]
@@ -295,7 +319,7 @@ mod tests {
         let aliases = SourceAlias::new(&[("{:open_message:}", "open_message")]);
 
         assert_eq!(
-            "open_message.open_message_id as open_message_id, open_message.epoch_settings_id as epoch_settings_id, open_message.beacon as beacon, open_message.signed_entity_type_id as signed_entity_type_id, open_message.message as message, open_message.created_at as created_at".to_string(),
+            "open_message.open_message_id as open_message_id, open_message.epoch_setting_id as epoch_setting_id, open_message.beacon as beacon, open_message.signed_entity_type_id as signed_entity_type_id, open_message.message as message, open_message.created_at as created_at".to_string(),
             projection.expand(aliases)
         )
     }
@@ -306,7 +330,7 @@ mod tests {
         let provider = OpenMessageProvider::new(&connection);
         let (expr, params) = provider.get_epoch_condition(Epoch(12)).expand();
 
-        assert_eq!("epoch_settings_id = ?1".to_string(), expr);
+        assert_eq!("epoch_setting_id = ?1".to_string(), expr);
         assert_eq!(vec![Value::Integer(12)], params,);
     }
 
@@ -353,7 +377,7 @@ mod tests {
             .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!("(open_message_id, epoch_setting_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()),
@@ -369,7 +393,87 @@ mod tests {
         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!("epoch_setting_id = ?1".to_string(), expr);
         assert_eq!(vec![Value::Integer(12)], params,);
     }
+
+    async fn get_connection() -> Arc<Mutex<Connection>> {
+        let config = Configuration::new_sample();
+        let mut builder = DependenciesBuilder::new(config);
+        let connection = builder.get_sqlite_connection().await.unwrap();
+        {
+            let lock = connection.lock().await;
+            lock.execute(r#"insert into epoch_setting(epoch_setting_id, protocol_parameters) values (1, '{"k": 100, "m": 5, "phi": 0.65 }');"#).unwrap();
+        }
+
+        connection
+    }
+
+    #[tokio::test]
+    async fn repository_create_open_message() {
+        let connection = get_connection().await;
+        let repository = OpenMessageRepository::new(connection.clone());
+        let open_message = repository
+            .create_open_message(
+                Epoch(1),
+                &Beacon::default(),
+                &SignedEntityType::CardanoImmutableFilesFull,
+                "this is a message",
+            )
+            .await
+            .unwrap();
+
+        assert_eq!(Epoch(1), open_message.epoch);
+        assert_eq!("this is a message".to_string(), open_message.message);
+        assert_eq!(
+            SignedEntityType::CardanoImmutableFilesFull,
+            open_message.signed_entity_type
+        );
+
+        let message = {
+            let lock = connection.lock().await;
+            let provider = OpenMessageProvider::new(&lock);
+            let mut cursor = provider
+                .find(WhereCondition::new(
+                    "open_message_id = ?*",
+                    vec![Value::String(open_message.open_message_id.to_string())],
+                ))
+                .unwrap();
+
+            cursor.next().expect(&format!(
+                "OpenMessage ID='{}' should exist in the database.",
+                open_message.open_message_id.to_string()
+            ))
+        };
+
+        assert_eq!(open_message.message, message.message);
+        assert_eq!(open_message.epoch, message.epoch);
+    }
+
+    #[tokio::test]
+    async fn repository_clean_open_message() {
+        let connection = get_connection().await;
+        let repository = OpenMessageRepository::new(connection.clone());
+        let _ = repository
+            .create_open_message(
+                Epoch(1),
+                &Beacon::default(),
+                &SignedEntityType::CardanoImmutableFilesFull,
+                "this is a message",
+            )
+            .await
+            .unwrap();
+        let _ = repository
+            .create_open_message(
+                Epoch(1),
+                &Beacon::default(),
+                &SignedEntityType::MithrilStakeDistribution,
+                "this is a stake distribution",
+            )
+            .await
+            .unwrap();
+        let count = repository.clean_epoch(Epoch(1)).await.unwrap();
+
+        assert_eq!(2, count);
+    }
 }
diff --git a/mithril-aggregator/src/dependency_injection/builder.rs b/mithril-aggregator/src/dependency_injection/builder.rs
@@ -184,16 +184,19 @@ impl DependenciesBuilder {
     }
 
     async fn build_sqlite_connection(&self) -> Result<Arc<Mutex<Connection>>> {
-        let connection = match self.configuration.environment {
+        let path = match self.configuration.environment {
             ExecutionEnvironment::Production => {
-                Connection::open(self.configuration.get_sqlite_dir().join(SQLITE_FILE))
+                self.configuration.get_sqlite_dir().join(SQLITE_FILE)
             }
-            _ => Connection::open(":memory:"),
+            _ => ":memory:".into(),
         };
-        let connection = connection
-            .map(|conn| Arc::new(Mutex::new(conn)))
+        let connection = Connection::open(&path)
+            .map(|c| Arc::new(Mutex::new(c)))
             .map_err(|e| DependenciesBuilderError::Initialization {
-                message: "Could not initialize SQLite driver.".to_string(),
+                message: format!(
+                    "SQLite initialization: could not open connection with string '{}'.",
+                    path.display()
+                ),
                 error: Some(Box::new(e)),
             })?;
         // Check database migrations
@@ -207,6 +210,16 @@ impl DependenciesBuilder {
             db_checker.add_migration(migration);
         }
 
+        // configure session
+        connection
+            .lock()
+            .await
+            .execute("pragma foreign_keys=true")
+            .map_err(|e| DependenciesBuilderError::Initialization {
+                message: "SQLite initialization: could not enable FOREIGN KEY support.".to_string(),
+                error: Some(e.into()),
+            })?;
+
         db_checker.apply().await?;
 
         Ok(connection)
