Listen to real stream events

Author: SupernaviX

Cargo.lock

@@ -9,6 +9,7 @@ aide = { version = "0.13", features = ["axum"] }
 anyhow = "1"
 async-trait = "0.1"
 axum = { version = "0.7", features = ["ws"] }
+blockfrost = "1"
 clap = { version = "4", features = ["derive"] }
 chrono = "0.4"
 dashmap = "6"

firefly-cardanoconnect/Cargo.toml

@@ -1,7 +1,7 @@
 use std::path::PathBuf;
 
 use crate::{
-    config::CardanoConnectConfig,
+    config::{CardanoConnectConfig, Secret},
     streams::{BlockInfo, BlockReference},
 };
 use anyhow::{bail, Result};
@@ -19,8 +19,10 @@ mod n2c;
 #[serde(rename_all = "camelCase")]
 pub struct BlockchainConfig {
     pub socket: PathBuf,
+    pub blockfrost_key: Option<Secret<String>>,
     pub network: Option<Network>,
     pub network_magic: Option<u64>,
+    pub genesis_hash: Option<String>,
     pub era: u16,
 }
 
@@ -43,6 +45,16 @@ impl BlockchainConfig {
             Network::Mainnet => 764824073,
         }
     }
+    fn genesis_hash(&self) -> &str {
+        if let Some(hash) = &self.genesis_hash {
+            return hash;
+        }
+        match self.network.unwrap_or(Network::Mainnet) {
+            Network::PreProd => "f28f1c1280ea0d32f8cd3143e268650d6c1a8e221522ce4a7d20d62fc09783e1",
+            Network::Preview => "83de1d7302569ad56cf9139a41e2e11346d4cb4a31c00142557b6ab3fa550761",
+            Network::Mainnet => "5f20df933584822601f9e3f8c024eb5eb252fe8cefb24d1317dc3d432e940ebb",
+        }
+    }
 }
 
 enum ClientImpl {
@@ -52,6 +64,7 @@ enum ClientImpl {
 
 pub struct BlockchainClient {
     client: ClientImpl,
+    genesis_hash: String,
     era: u16,
 }
 
@@ -60,25 +73,38 @@ impl BlockchainClient {
         let blockchain = &config.connector.blockchain;
 
         let n2c = {
-            let client = NodeToClient::new(&blockchain.socket, blockchain.magic()).await?;
+            let client = NodeToClient::new(
+                &blockchain.socket,
+                blockchain.magic(),
+                blockchain.genesis_hash(),
+                blockchain.blockfrost_key.as_ref(),
+            )
+            .await?;
             RwLock::new(client)
         };
 
         Ok(Self {
             client: ClientImpl::NodeToClient(n2c),
+            genesis_hash: blockchain.genesis_hash().to_string(),
             era: blockchain.era,
         })
     }
 
     #[allow(unused)]
     pub fn mock() -> Self {
         let mock_chain = MockChain::new(3000);
+        let genesis_hash = mock_chain.genesis_hash();
         Self {
             client: ClientImpl::Mock(mock_chain),
+            genesis_hash,
             era: 0,
         }
     }
 
+    pub fn genesis_hash(&self) -> String {
+        self.genesis_hash.clone()
+    }
+
     pub async fn submit(&self, transaction: Tx) -> Result<String> {
         match &self.client {
             ClientImpl::Mock(_) => bail!("mock transaction submission not implemented"),

firefly-cardanoconnect/src/blockchain.rs

@@ -33,15 +33,14 @@ impl ChainSyncClient for MockChainSync {
             }
 
             // what are you waiting for?
-            let requested_block_number = match &self.consumer_tip {
-                BlockReference::Origin => 0,
-                BlockReference::Point(number, _) => *number + 1,
+            let next_index = match &self.consumer_tip {
+                BlockReference::Origin => Some(1),
+                BlockReference::Point(_, hash) => self.chain.indexes.get(hash).map(|i| *i + 1),
             };
 
             // if we have it, give it
-            if let Some(info) = chain.get(requested_block_number as usize) {
-                self.consumer_tip =
-                    BlockReference::Point(requested_block_number, info.block_hash.clone());
+            if let Some(info) = next_index.and_then(|i| chain.get(i)) {
+                self.consumer_tip = BlockReference::Point(info.block_slot, info.block_hash.clone());
                 return Ok(RequestNextResponse::RollForward(info.clone(), tip));
             }
 
@@ -58,9 +57,9 @@ impl ChainSyncClient for MockChainSync {
         let chain = self.chain.read_lock().await;
         let intersect = points.iter().find_map(|point| match point {
             BlockReference::Origin => chain.first(),
-            BlockReference::Point(number, hash) => chain
-                .get(*number as usize)
-                .filter(|b| b.block_hash == *hash),
+            BlockReference::Point(_, hash) => {
+                self.chain.indexes.get(hash).and_then(|i| chain.get(*i))
+            }
         });
         self.consumer_tip = intersect.map(|b| b.as_reference()).unwrap_or_default();
         let tip = chain.last().map(|b| b.as_reference()).unwrap_or_default();
@@ -77,29 +76,42 @@ impl ChainSyncClient for MockChainSync {
 pub struct MockChain {
     // the real implementation of course won't be in memory
     chain: Arc<RwLock<Vec<BlockInfo>>>,
+    indexes: Arc<DashMap<String, usize>>,
     new_block: Arc<Notify>,
     rolled_back: Arc<DashMap<BlockReference, BlockReference>>,
 }
 
 impl MockChain {
     pub fn new(initial_height: usize) -> Self {
         let chain = Arc::new(RwLock::new(vec![]));
+        let indexes = Arc::new(DashMap::new());
         let new_block = Arc::new(Notify::new());
         let rolled_back = Arc::new(DashMap::new());
         tokio::spawn(Self::generate(
             chain.clone(),
+            indexes.clone(),
             new_block.clone(),
             rolled_back.clone(),
             initial_height,
         ));
 
         Self {
             chain,
+            indexes,
             new_block,
             rolled_back,
         }
     }
 
+    pub fn genesis_hash(&self) -> String {
+        self.chain
+            .blocking_read()
+            .first()
+            .unwrap()
+            .block_hash
+            .clone()
+    }
+
     pub fn sync(&self) -> MockChainSync {
         MockChainSync {
             chain: self.clone(),
@@ -130,11 +142,13 @@ impl MockChain {
     pub async fn request_block(&self, block_ref: &BlockReference) -> Result<Option<BlockInfo>> {
         match block_ref {
             BlockReference::Origin => Ok(None),
-            BlockReference::Point(number, hash) => {
+            BlockReference::Point(slot, hash) => {
                 let chain = self.chain.read().await;
                 Ok(chain
-                    .get(*number as usize)
-                    .filter(|b| b.block_hash == *hash)
+                    .iter()
+                    .rev()
+                    .find(|b| b.block_hash == *hash)
+                    .filter(|b| b.block_slot == *slot)
                     .cloned())
             }
         }
@@ -143,6 +157,7 @@ impl MockChain {
     // TODO: roll back sometimes
     async fn generate(
         chain: Arc<RwLock<Vec<BlockInfo>>>,
+        indexes: Arc<DashMap<String, usize>>,
         new_block: Arc<Notify>,
         _rolled_back: Arc<DashMap<BlockReference, BlockReference>>,
         initial_height: usize,
@@ -151,33 +166,39 @@ impl MockChain {
         {
             let mut lock = chain.write().await;
             for _ in 0..initial_height {
-                Self::generate_block(&mut rng, &mut lock);
+                Self::generate_block(&mut rng, &mut lock, &indexes);
             }
         }
         loop {
             time::sleep(Duration::from_secs(1)).await;
             let mut lock = chain.write().await;
-            Self::generate_block(&mut rng, &mut lock);
+            Self::generate_block(&mut rng, &mut lock, &indexes);
             new_block.notify_waiters();
         }
     }
 
-    fn generate_block(rng: &mut ChaChaRng, chain: &mut Vec<BlockInfo>) {
-        let (block_number, parent_hash) = match chain.last() {
-            Some(block) => (block.block_number + 1, block.block_hash.clone()),
-            None => (0, "".into()),
+    fn generate_block(
+        rng: &mut ChaChaRng,
+        chain: &mut Vec<BlockInfo>,
+        indexes: &DashMap<String, usize>,
+    ) {
+        let (block_height, parent_hash) = match chain.last() {
+            Some(block) => (Some(chain.len() as u64), Some(block.block_hash.clone())),
+            None => (None, None),
         };
 
         let mut transaction_hashes = vec![];
         for _ in 0..rng.gen_range(0..10) {
             transaction_hashes.push(Self::generate_hash(rng));
         }
         let block = BlockInfo {
-            block_number,
+            block_height,
+            block_slot: block_height,
             block_hash: Self::generate_hash(rng),
             parent_hash,
             transaction_hashes,
         };
+        indexes.insert(block.block_hash.clone(), chain.len());
         chain.push(block);
     }