Implement Cardano stake distribution HTTP routes

CHANGELOG.md

@@ -13,7 +13,8 @@ As a minor extension, we have adopted a slightly different versioning convention
 
 - **UNSTABLE** Cardano stake distribution certification:
 
-  - Implement the signable and artifact builders for the signed entity type `CardanoStakeDistribution`
+  - Implement the signable and artifact builders for the signed entity type `CardanoStakeDistribution`.
+  - Implement the HTTP routes related to the signed entity type `CardanoStakeDistribution` on the aggregator REST API.
 
 - Crates versions:
 

mithril-aggregator/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "mithril-aggregator"
-version = "0.5.53"
+version = "0.5.54"
 description = "A Mithril Aggregator server"
 authors = { workspace = true }
 edition = { workspace = true }

mithril-aggregator/src/database/migration.rs

@@ -752,5 +752,13 @@ pragma foreign_keys=true;
                 SignedEntityTypeDiscriminants::CardanoTransactions.index()
             ),
         ),
+        // Migration 26
+        // Alter `signed_entity` table, add a unique index on `signed_entity_type_id` and `beacon`
+        SqlMigration::new(
+            26,
+            r#"
+create unique index signed_entity_unique_index on signed_entity(signed_entity_type_id, beacon);
+"#,
+        ),
     ]
 }

mithril-aggregator/src/database/query/signed_entity/get_signed_entity.rs

@@ -1,6 +1,6 @@
 use sqlite::Value;
 
-use mithril_common::entities::SignedEntityTypeDiscriminants;
+use mithril_common::entities::{Epoch, SignedEntityTypeDiscriminants};
 use mithril_common::StdResult;
 use mithril_persistence::sqlite::{Query, SourceAlias, SqLiteEntity, WhereCondition};
 
@@ -60,6 +60,19 @@ impl GetSignedEntityRecordQuery {
             ),
         })
     }
+
+    pub fn cardano_stake_distribution_by_epoch(epoch: Epoch) -> Self {
+        let signed_entity_type_id =
+            SignedEntityTypeDiscriminants::CardanoStakeDistribution.index() as i64;
+        let epoch = *epoch as i64;
+
+        Self {
+            condition: WhereCondition::new(
+                "signed_entity_type_id = ?* and beacon = ?*",
+                vec![Value::Integer(signed_entity_type_id), Value::Integer(epoch)],
+            ),
+        }
+    }
 }
 
 impl Query for GetSignedEntityRecordQuery {
@@ -80,13 +93,96 @@ impl Query for GetSignedEntityRecordQuery {
 
 #[cfg(test)]
 mod tests {
-    use mithril_common::entities::{CardanoDbBeacon, SignedEntityType};
+    use chrono::DateTime;
+    use mithril_common::{
+        entities::{CardanoDbBeacon, SignedEntityType},
+        test_utils::fake_data,
+    };
     use mithril_persistence::sqlite::ConnectionExtensions;
+    use sqlite::ConnectionThreadSafe;
 
     use crate::database::test_helper::{insert_signed_entities, main_db_connection};
 
     use super::*;
 
+    fn create_database_with_cardano_stake_distributions<T: Into<SignedEntityRecord>>(
+        cardano_stake_distributions: Vec<T>,
+    ) -> (ConnectionThreadSafe, Vec<SignedEntityRecord>) {
+        let records = cardano_stake_distributions
+            .into_iter()
+            .map(|cardano_stake_distribution| cardano_stake_distribution.into())
+            .collect::<Vec<_>>();
+
+        let connection = create_database(&records);
+
+        (connection, records)
+    }
+
+    fn create_database(records: &[SignedEntityRecord]) -> ConnectionThreadSafe {
+        let connection = main_db_connection().unwrap();
+        insert_signed_entities(&connection, records.to_vec()).unwrap();
+        connection
+    }
+
+    #[test]
+    fn cardano_stake_distribution_by_epoch_returns_records_filtered_by_epoch() {
+        let mut cardano_stake_distributions = fake_data::cardano_stake_distributions(3);
+        cardano_stake_distributions[0].epoch = Epoch(3);
+        cardano_stake_distributions[1].epoch = Epoch(4);
+        cardano_stake_distributions[2].epoch = Epoch(5);
+
+        let (connection, records) =
+            create_database_with_cardano_stake_distributions(cardano_stake_distributions);
+
+        let records_retrieved: Vec<SignedEntityRecord> = connection
+            .fetch_collect(
+                GetSignedEntityRecordQuery::cardano_stake_distribution_by_epoch(Epoch(4)),
+            )
+            .unwrap();
+
+        assert_eq!(vec![records[1].clone()], records_retrieved);
+    }
+
+    #[test]
+    fn cardano_stake_distribution_by_epoch_returns_records_returns_only_cardano_stake_distribution_records(
+    ) {
+        let cardano_stake_distributions_record: SignedEntityRecord = {
+            let mut cardano_stake_distribution = fake_data::cardano_stake_distribution(Epoch(4));
+            cardano_stake_distribution.hash = "hash-123".to_string();
+            cardano_stake_distribution.into()
+        };
+
+        let snapshots_record = {
+            let mut snapshot = fake_data::snapshots(1)[0].clone();
+            snapshot.beacon.epoch = Epoch(4);
+            SignedEntityRecord::from_snapshot(snapshot, "whatever".to_string(), DateTime::default())
+        };
+
+        let mithril_stake_distribution_record: SignedEntityRecord = {
+            let mithril_stake_distributions = fake_data::mithril_stake_distributions(1);
+            let mut mithril_stake_distribution = mithril_stake_distributions[0].clone();
+            mithril_stake_distribution.epoch = Epoch(4);
+            mithril_stake_distribution.into()
+        };
+
+        let connection = create_database(&[
+            cardano_stake_distributions_record.clone(),
+            snapshots_record,
+            mithril_stake_distribution_record,
+        ]);
+
+        let records_retrieved: Vec<SignedEntityRecord> = connection
+            .fetch_collect(
+                GetSignedEntityRecordQuery::cardano_stake_distribution_by_epoch(Epoch(4)),
+            )
+            .unwrap();
+
+        assert_eq!(
+            vec![cardano_stake_distributions_record.clone()],
+            records_retrieved,
+        );
+    }
+
     #[test]
     fn test_get_signed_entity_records() {
         let signed_entity_records = SignedEntityRecord::fake_records(5);

mithril-aggregator/src/database/record/signed_entity.rs

@@ -2,8 +2,13 @@ use chrono::{DateTime, Utc};
 use serde::{Deserialize, Serialize};
 
 use mithril_common::crypto_helper::ProtocolParameters;
-use mithril_common::entities::{BlockNumber, Epoch, SignedEntity, SignedEntityType, Snapshot};
+use mithril_common::entities::{
+    BlockNumber, Epoch, SignedEntity, SignedEntityType, Snapshot, StakeDistribution,
+};
+#[cfg(test)]
+use mithril_common::entities::{CardanoStakeDistribution, MithrilStakeDistribution};
 use mithril_common::messages::{
+    CardanoStakeDistributionListItemMessage, CardanoStakeDistributionMessage,
     CardanoTransactionSnapshotListItemMessage, CardanoTransactionSnapshotMessage,
     MithrilStakeDistributionListItemMessage, MithrilStakeDistributionMessage,
     SignerWithStakeMessagePart, SnapshotListItemMessage, SnapshotMessage,
@@ -32,6 +37,32 @@ pub struct SignedEntityRecord {
     pub created_at: DateTime<Utc>,
 }
 
+#[cfg(test)]
+impl From<CardanoStakeDistribution> for SignedEntityRecord {
+    fn from(cardano_stake_distribution: CardanoStakeDistribution) -> Self {
+        SignedEntityRecord::from_cardano_stake_distribution(cardano_stake_distribution)
+    }
+}
+
+#[cfg(test)]
+impl From<MithrilStakeDistribution> for SignedEntityRecord {
+    fn from(mithril_stake_distribution: MithrilStakeDistribution) -> Self {
+        let entity = serde_json::to_string(&mithril_stake_distribution).unwrap();
+
+        SignedEntityRecord {
+            signed_entity_id: mithril_stake_distribution.hash.clone(),
+            signed_entity_type: SignedEntityType::MithrilStakeDistribution(
+                mithril_stake_distribution.epoch,
+            ),
+            certificate_id: format!("certificate-{}", mithril_stake_distribution.hash),
+            artifact: entity,
+            created_at: DateTime::parse_from_rfc3339("2023-01-19T13:43:05.618857482Z")
+                .unwrap()
+                .with_timezone(&Utc),
+        }
+    }
+}
+
 #[cfg(test)]
 impl SignedEntityRecord {
     pub(crate) fn from_snapshot(
@@ -50,6 +81,24 @@ impl SignedEntityRecord {
         }
     }
 
+    pub(crate) fn from_cardano_stake_distribution(
+        cardano_stake_distribution: CardanoStakeDistribution,
+    ) -> Self {
+        let entity = serde_json::to_string(&cardano_stake_distribution).unwrap();
+
+        SignedEntityRecord {
+            signed_entity_id: cardano_stake_distribution.hash.clone(),
+            signed_entity_type: SignedEntityType::CardanoStakeDistribution(
+                cardano_stake_distribution.epoch,
+            ),
+            certificate_id: format!("certificate-{}", cardano_stake_distribution.hash),
+            artifact: entity,
+            created_at: DateTime::parse_from_rfc3339("2023-01-19T13:43:05.618857482Z")
+                .unwrap()
+                .with_timezone(&Utc),
+        }
+    }
+
     pub(crate) fn fake_records(number_if_records: usize) -> Vec<SignedEntityRecord> {
         use mithril_common::test_utils::fake_data;
 
@@ -233,6 +282,52 @@ impl TryFrom<SignedEntityRecord> for SnapshotListItemMessage {
     }
 }
 
+impl TryFrom<SignedEntityRecord> for CardanoStakeDistributionMessage {
+    type Error = StdError;
+
+    fn try_from(value: SignedEntityRecord) -> Result<Self, Self::Error> {
+        #[derive(Deserialize)]
+        struct TmpCardanoStakeDistribution {
+            hash: String,
+            stake_distribution: StakeDistribution,
+        }
+        let artifact = serde_json::from_str::<TmpCardanoStakeDistribution>(&value.artifact)?;
+        let cardano_stake_distribution_message = CardanoStakeDistributionMessage {
+            // The epoch stored in the signed entity type beacon corresponds to epoch
+            // at the end of which the Cardano stake distribution is computed by the Cardano node.
+            epoch: value.signed_entity_type.get_epoch(),
+            stake_distribution: artifact.stake_distribution,
+            hash: artifact.hash,
+            certificate_hash: value.certificate_id,
+            created_at: value.created_at,
+        };
+
+        Ok(cardano_stake_distribution_message)
+    }
+}
+
+impl TryFrom<SignedEntityRecord> for CardanoStakeDistributionListItemMessage {
+    type Error = StdError;
+
+    fn try_from(value: SignedEntityRecord) -> Result<Self, Self::Error> {
+        #[derive(Deserialize)]
+        struct TmpCardanoStakeDistribution {
+            hash: String,
+        }
+        let artifact = serde_json::from_str::<TmpCardanoStakeDistribution>(&value.artifact)?;
+        let message = CardanoStakeDistributionListItemMessage {
+            // The epoch stored in the signed entity type beacon corresponds to epoch
+            // at the end of which the Cardano stake distribution is computed by the Cardano node.
+            epoch: value.signed_entity_type.get_epoch(),
+            hash: artifact.hash,
+            certificate_hash: value.certificate_id,
+            created_at: value.created_at,
+        };
+
+        Ok(message)
+    }
+}
+
 impl SqLiteEntity for SignedEntityRecord {
     fn hydrate(row: sqlite::Row) -> Result<Self, HydrationError>
     where

mithril-aggregator/src/database/repository/signed_entity_store.rs

@@ -3,7 +3,7 @@ use std::sync::Arc;
 use anyhow::Context;
 use async_trait::async_trait;
 
-use mithril_common::entities::SignedEntityTypeDiscriminants;
+use mithril_common::entities::{Epoch, SignedEntityTypeDiscriminants};
 use mithril_common::StdResult;
 use mithril_persistence::sqlite::{ConnectionExtensions, SqliteConnection};
 
@@ -44,6 +44,12 @@ pub trait SignedEntityStorer: Sync + Send {
         total: usize,
     ) -> StdResult<Vec<SignedEntityRecord>>;
 
+    /// Get Cardano stake distribution signed entity by epoch
+    async fn get_cardano_stake_distribution_signed_entity_by_epoch(
+        &self,
+        epoch: Epoch,
+    ) -> StdResult<Option<SignedEntityRecord>>;
+
     /// Perform an update for all the given signed entities.
     async fn update_signed_entities(
         &self,
@@ -127,6 +133,14 @@ impl SignedEntityStorer for SignedEntityStore {
         Ok(signed_entities)
     }
 
+    async fn get_cardano_stake_distribution_signed_entity_by_epoch(
+        &self,
+        epoch: Epoch,
+    ) -> StdResult<Option<SignedEntityRecord>> {
+        self.connection
+            .fetch_first(GetSignedEntityRecordQuery::cardano_stake_distribution_by_epoch(epoch))
+    }
+
     async fn update_signed_entities(
         &self,
         signed_entities: Vec<SignedEntityRecord>,
@@ -150,7 +164,10 @@ impl SignedEntityStorer for SignedEntityStore {
 
 #[cfg(test)]
 mod tests {
-    use mithril_common::entities::{MithrilStakeDistribution, SignedEntity, Snapshot};
+    use mithril_common::{
+        entities::{Epoch, MithrilStakeDistribution, SignedEntity, Snapshot},
+        test_utils::fake_data,
+    };
 
     use crate::database::test_helper::{insert_signed_entities, main_db_connection};
 
@@ -310,4 +327,37 @@ mod tests {
         assert_eq!(records_to_update, updated_records);
         assert_eq!(expected_records, stored_records);
     }
+
+    #[tokio::test]
+    async fn get_cardano_stake_distribution_signed_entity_by_epoch_when_nothing_found() {
+        let epoch_to_retrieve = Epoch(4);
+        let connection = main_db_connection().unwrap();
+        let store = SignedEntityStore::new(Arc::new(connection));
+
+        let record = store
+            .get_cardano_stake_distribution_signed_entity_by_epoch(epoch_to_retrieve)
+            .await
+            .unwrap();
+
+        assert_eq!(None, record);
+    }
+
+    #[tokio::test]
+    async fn get_cardano_stake_distribution_signed_entity_by_epoch_when_signed_entity_found_for_epoch(
+    ) {
+        let cardano_stake_distribution = fake_data::cardano_stake_distribution(Epoch(4));
+
+        let expected_record: SignedEntityRecord = cardano_stake_distribution.into();
+
+        let connection = main_db_connection().unwrap();
+        insert_signed_entities(&connection, vec![expected_record.clone()]).unwrap();
+        let store = SignedEntityStore::new(Arc::new(connection));
+
+        let record = store
+            .get_cardano_stake_distribution_signed_entity_by_epoch(Epoch(4))
+            .await
+            .unwrap();
+
+        assert_eq!(Some(expected_record), record);
+    }
 }