Subgraph Composition

chain/arweave/src/chain.rs

@@ -3,11 +3,11 @@ use graph::blockchain::client::ChainClient;
 use graph::blockchain::firehose_block_ingestor::FirehoseBlockIngestor;
 use graph::blockchain::{
     BasicBlockchainBuilder, Block, BlockIngestor, BlockchainBuilder, BlockchainKind,
-    EmptyNodeCapabilities, NoopDecoderHook, NoopRuntimeAdapter,
+    EmptyNodeCapabilities, NoopDecoderHook, NoopRuntimeAdapter, TriggerFilterWrapper,
 };
 use graph::cheap_clone::CheapClone;
 use graph::components::network_provider::ChainName;
-use graph::components::store::DeploymentCursorTracker;
+use graph::components::store::{DeploymentCursorTracker, SourceableStore};
 use graph::data::subgraph::UnifiedMappingApiVersion;
 use graph::env::EnvVars;
 use graph::firehose::FirehoseEndpoint;
@@ -27,11 +27,13 @@ use graph::{
     prelude::{async_trait, o, BlockNumber, ChainStore, Error, Logger, LoggerFactory},
 };
 use prost::Message;
+use std::collections::HashSet;
 use std::sync::Arc;
 
 use crate::adapter::TriggerFilter;
 use crate::data_source::{DataSourceTemplate, UnresolvedDataSourceTemplate};
 use crate::trigger::{self, ArweaveTrigger};
+use crate::Block as ArweaveBlock;
 use crate::{
     codec,
     data_source::{DataSource, UnresolvedDataSource},
@@ -119,7 +121,8 @@ impl Blockchain for Chain {
         deployment: DeploymentLocator,
         store: impl DeploymentCursorTracker,
         start_blocks: Vec<BlockNumber>,
-        filter: Arc<Self::TriggerFilter>,
+        _source_subgraph_stores: Vec<Arc<dyn SourceableStore>>,
+        filter: Arc<TriggerFilterWrapper<Self>>,
         unified_api_version: UnifiedMappingApiVersion,
     ) -> Result<Box<dyn BlockStream<Self>>, Error> {
         let adapter = self
@@ -135,7 +138,10 @@ impl Blockchain for Chain {
             .subgraph_logger(&deployment)
             .new(o!("component" => "FirehoseBlockStream"));
 
-        let firehose_mapper = Arc::new(FirehoseMapper { adapter, filter });
+        let firehose_mapper = Arc::new(FirehoseMapper {
+            adapter,
+            filter: filter.chain_filter.clone(),
+        });
 
         Ok(Box::new(FirehoseBlockStream::new(
             deployment.hash,
@@ -199,6 +205,10 @@ impl TriggersAdapterTrait<Chain> for TriggersAdapter {
         panic!("Should never be called since not used by FirehoseBlockStream")
     }
 
+    async fn chain_head_ptr(&self) -> Result<Option<BlockPtr>, Error> {
+        unimplemented!()
+    }
+
     async fn triggers_in_block(
         &self,
         logger: &Logger,
@@ -258,6 +268,14 @@ impl TriggersAdapterTrait<Chain> for TriggersAdapter {
             number: block.number.saturating_sub(1),
         }))
     }
+
+    async fn load_block_ptrs_by_numbers(
+        &self,
+        _logger: Logger,
+        _block_numbers: HashSet<BlockNumber>,
+    ) -> Result<Vec<ArweaveBlock>, Error> {
+        todo!()
+    }
 }
 
 pub struct FirehoseMapper {

chain/cosmos/src/chain.rs

@@ -1,17 +1,18 @@
 use graph::blockchain::firehose_block_ingestor::FirehoseBlockIngestor;
-use graph::blockchain::{BlockIngestor, NoopDecoderHook};
+use graph::blockchain::{BlockIngestor, NoopDecoderHook, TriggerFilterWrapper};
 use graph::components::network_provider::ChainName;
 use graph::env::EnvVars;
 use graph::prelude::MetricsRegistry;
 use graph::substreams::Clock;
+use std::collections::HashSet;
 use std::convert::TryFrom;
 use std::sync::Arc;
 
 use graph::blockchain::block_stream::{BlockStreamError, BlockStreamMapper, FirehoseCursor};
 use graph::blockchain::client::ChainClient;
 use graph::blockchain::{BasicBlockchainBuilder, BlockchainBuilder, NoopRuntimeAdapter};
 use graph::cheap_clone::CheapClone;
-use graph::components::store::DeploymentCursorTracker;
+use graph::components::store::{DeploymentCursorTracker, SourceableStore};
 use graph::data::subgraph::UnifiedMappingApiVersion;
 use graph::{
     blockchain::{
@@ -33,7 +34,7 @@ use crate::data_source::{
     DataSource, DataSourceTemplate, EventOrigin, UnresolvedDataSource, UnresolvedDataSourceTemplate,
 };
 use crate::trigger::CosmosTrigger;
-use crate::{codec, TriggerFilter};
+use crate::{codec, Block, TriggerFilter};
 
 pub struct Chain {
     logger_factory: LoggerFactory,
@@ -113,7 +114,8 @@ impl Blockchain for Chain {
         deployment: DeploymentLocator,
         store: impl DeploymentCursorTracker,
         start_blocks: Vec<BlockNumber>,
-        filter: Arc<Self::TriggerFilter>,
+        _source_subgraph_stores: Vec<Arc<dyn SourceableStore>>,
+        filter: Arc<TriggerFilterWrapper<Self>>,
         unified_api_version: UnifiedMappingApiVersion,
     ) -> Result<Box<dyn BlockStream<Self>>, Error> {
         let adapter = self
@@ -129,7 +131,10 @@ impl Blockchain for Chain {
             .subgraph_logger(&deployment)
             .new(o!("component" => "FirehoseBlockStream"));
 
-        let firehose_mapper = Arc::new(FirehoseMapper { adapter, filter });
+        let firehose_mapper = Arc::new(FirehoseMapper {
+            adapter,
+            filter: filter.chain_filter.clone(),
+        });
 
         Ok(Box::new(FirehoseBlockStream::new(
             deployment.hash,
@@ -192,6 +197,16 @@ impl TriggersAdapterTrait<Chain> for TriggersAdapter {
     ) -> Result<Option<codec::Block>, Error> {
         panic!("Should never be called since not used by FirehoseBlockStream")
     }
+    async fn load_block_ptrs_by_numbers(
+        &self,
+        _logger: Logger,
+        _block_numbers: HashSet<BlockNumber>,
+    ) -> Result<Vec<Block>, Error> {
+        todo!()
+    }
+    async fn chain_head_ptr(&self) -> Result<Option<BlockPtr>, Error> {
+        unimplemented!()
+    }
 
     async fn scan_triggers(
         &self,
@@ -467,9 +482,12 @@ impl FirehoseMapperTrait<Chain> for FirehoseMapper {
 
 #[cfg(test)]
 mod test {
-    use graph::prelude::{
-        slog::{o, Discard, Logger},
-        tokio,
+    use graph::{
+        blockchain::Trigger,
+        prelude::{
+            slog::{o, Discard, Logger},
+            tokio,
+        },
     };
 
     use super::*;
@@ -600,7 +618,10 @@ mod test {
             // they may not be in the same order
             for trigger in expected_triggers {
                 assert!(
-                    triggers.trigger_data.contains(&trigger),
+                    triggers.trigger_data.iter().any(|t| match t {
+                        Trigger::Chain(t) => t == &trigger,
+                        _ => false,
+                    }),
                     "Expected trigger list to contain {:?}, but it only contains: {:?}",
                     trigger,
                     triggers.trigger_data

chain/ethereum/src/adapter.rs

@@ -1,8 +1,9 @@
 use anyhow::Error;
-use ethabi::{Error as ABIError, Function, ParamType, Token};
+use ethabi::{Error as ABIError, ParamType, Token};
 use graph::blockchain::ChainIdentifier;
 use graph::components::subgraph::MappingError;
 use graph::data::store::ethereum::call;
+use graph::data_source::common::ContractCall;
 use graph::firehose::CallToFilter;
 use graph::firehose::CombinedFilter;
 use graph::firehose::LogFilter;
@@ -93,16 +94,6 @@ impl EventSignatureWithTopics {
     }
 }
 
-#[derive(Clone, Debug)]
-pub struct ContractCall {
-    pub contract_name: String,
-    pub address: Address,
-    pub block_ptr: BlockPtr,
-    pub function: Function,
-    pub args: Vec<Token>,
-    pub gas: Option<u32>,
-}
-
 #[derive(Error, Debug)]
 pub enum EthereumRpcError {
     #[error("call error: {0}")]