graph, chain: make TriggersAdapterTrait methods work with firehose endpoints

Author: incrypto32

chain/ethereum/src/chain.rs

@@ -17,7 +17,7 @@ use graph::prelude::{
     EthereumCallCache, LightEthereumBlock, LightEthereumBlockExt, MetricsRegistry,
 };
 use graph::schema::InputSchema;
-use graph::slog::{debug, error, warn};
+use graph::slog::{debug, error, trace, warn};
 use graph::substreams::Clock;
 use graph::{
     blockchain::{
@@ -827,6 +827,11 @@ impl TriggersAdapterTrait<Chain> for TriggersAdapter {
     ) -> Result<Vec<BlockFinality>> {
         let blocks = match &*self.chain_client {
             ChainClient::Firehose(endpoints, _) => {
+                trace!(
+                                    logger,
+                                    "Loading blocks from firehose";
+                                    "block_numbers" => format!("{:?}", block_numbers)
+                );
                 let endpoint = endpoints.endpoint().await?;
                 let chain_store = self.chain_store.clone();
 
@@ -867,6 +872,11 @@ impl TriggersAdapterTrait<Chain> for TriggersAdapter {
                 blocks
             }
             ChainClient::Rpc(client) => {
+                trace!(
+                    logger,
+                    "Loading blocks from RPC";
+                    "block_numbers" => format!("{:?}", block_numbers)
+                );
                 let adapter = client.cheapest_with(&self.capabilities).await?;
                 let chain_store = self.chain_store.clone();
 
@@ -956,13 +966,25 @@ impl TriggersAdapterTrait<Chain> for TriggersAdapter {
     }
 
     async fn is_on_main_chain(&self, ptr: BlockPtr) -> Result<bool, Error> {
-        self.chain_client
-            .rpc()?
-            .cheapest()
-            .await
-            .ok_or(anyhow!("unable to get adapter for is_on_main_chain"))?
-            .is_on_main_chain(&self.logger, ptr.clone())
-            .await
+        match &*self.chain_client {
+            ChainClient::Firehose(endpoints, _) => {
+                let endpoint = endpoints.endpoint().await?;
+                let block = endpoint
+                    .get_block_by_number::<codec::Block>(ptr.number as u64, &self.logger)
+                    .await
+                    .map_err(|e| anyhow!("Failed to fetch block from firehose: {}", e))?;
+
+                Ok(block.hash() == ptr.hash)
+            }
+            ChainClient::Rpc(adapter) => {
+                let adapter = adapter
+                    .cheapest()
+                    .await
+                    .ok_or_else(|| anyhow!("unable to get adapter for is_on_main_chain"))?;
+
+                adapter.is_on_main_chain(&self.logger, ptr).await
+            }
+        }
     }
 
     async fn ancestor_block(
@@ -992,10 +1014,18 @@ impl TriggersAdapterTrait<Chain> for TriggersAdapter {
         use graph::prelude::LightEthereumBlockExt;
 
         let block = match self.chain_client.as_ref() {
-            ChainClient::Firehose(_, _) => Some(BlockPtr {
-                hash: BlockHash::from(vec![0xff; 32]),
-                number: block.number.saturating_sub(1),
-            }),
+            ChainClient::Firehose(endpoints, _) => {
+                let endpoint = endpoints.endpoint().await?;
+                let block = endpoint
+                    .get_block_by_ptr::<codec::Block>(block, &self.logger)
+                    .await
+                    .context(format!(
+                        "Failed to fetch block by ptr {} from firehose, backtrace: {}",
+                        block,
+                        std::backtrace::Backtrace::force_capture()
+                    ))?;
+                block.parent_ptr()
+            }
             ChainClient::Rpc(adapters) => {
                 let blocks = adapters
                     .cheapest_with(&self.capabilities)

graph/src/firehose/endpoints.rs

@@ -509,6 +509,41 @@ impl FirehoseEndpoint {
         }
     }
 
+    pub async fn get_block_by_ptr<M>(
+        &self,
+        ptr: &BlockPtr,
+        logger: &Logger,
+    ) -> Result<M, anyhow::Error>
+    where
+        M: prost::Message + BlockchainBlock + Default + 'static,
+    {
+        debug!(
+            logger,
+            "Connecting to firehose to retrieve block for ptr {}", ptr;
+            "provider" => self.provider.as_str(),
+        );
+
+        let req = firehose::SingleBlockRequest {
+            transforms: [].to_vec(),
+            reference: Some(
+                firehose::single_block_request::Reference::BlockHashAndNumber(
+                    firehose::single_block_request::BlockHashAndNumber {
+                        hash: ptr.hash.to_string(),
+                        num: ptr.number as u64,
+                    },
+                ),
+            ),
+        };
+
+        let mut client = self.new_client();
+        match client.block(req).await {
+            Ok(v) => Ok(M::decode(
+                v.get_ref().block.as_ref().unwrap().value.as_ref(),
+            )?),
+            Err(e) => return Err(anyhow::format_err!("firehose error {}", e)),
+        }
+    }
+
     pub async fn get_block_by_number<M>(
         &self,
         number: u64,