core: Load dynamic data sources directly

This avoids both running a GraphQL query as well as resolving things
against IPFS

Author: lutter

core/src/subgraph/loader.rs

@@ -4,207 +4,71 @@ use std::time::Instant;
 
 use async_trait::async_trait;
 
-use graph::data::subgraph::schema::*;
-use graph::data::subgraph::UnresolvedDataSource;
-use graph::prelude::{DataSourceLoader as DataSourceLoaderTrait, GraphQlRunner, *};
-use graph_graphql::graphql_parser::{parse_query, query as q};
+use graph::components::store::StoredDynamicDataSource;
+use graph::prelude::{DataSourceLoader as DataSourceLoaderTrait, *};
 
-pub struct DataSourceLoader<L, Q, S> {
+pub struct DataSourceLoader<S> {
     store: Arc<S>,
-    link_resolver: Arc<L>,
-    graphql_runner: Arc<Q>,
 }
 
-impl<L, Q, S> DataSourceLoader<L, Q, S>
+impl<S> DataSourceLoader<S>
 where
-    L: LinkResolver,
     S: Store + SubgraphDeploymentStore,
-    Q: GraphQlRunner,
 {
-    pub fn new(store: Arc<S>, link_resolver: Arc<L>, graphql_runner: Arc<Q>) -> Self {
-        Self {
-            store,
-            link_resolver,
-            graphql_runner,
-        }
-    }
-
-    fn dynamic_data_sources_query(
-        &self,
-        deployment: &SubgraphDeploymentId,
-        skip: i32,
-    ) -> Result<Query, Error> {
-        // Obtain the "subgraphs" schema
-        let schema = self.store.api_schema(&SUBGRAPHS_ID)?;
-
-        // Construct a query for the subgraph deployment and all its
-        // dynamic data sources
-        //
-        // See also: ed42d219c6704a4aab57ce1ea66698e7.
-        // Note: This query needs to be in sync with the metadata schema.
-        let document = parse_query(
-            r#"
-            query deployment($id: ID!, $skip: Int!) {
-              subgraphDeployment(id: $id) {
-                dynamicDataSources(orderBy: id, skip: $skip) {
-                  kind
-                  network
-                  name
-                  context
-                  source { address abi }
-                  mapping {
-                    kind
-                    apiVersion
-                    language
-                    file
-                    entities
-                    abis { name file }
-                    blockHandlers { handler filter }
-                    callHandlers {  function handler }
-                    eventHandlers { event handler topic0 }
-                  }
-                  templates {
-                    kind
-                    network
-                    name
-                    source { abi }
-                    mapping {
-                      kind
-                      apiVersion
-                      language
-                      file
-                      entities
-                      abis { name file }
-                      blockHandlers { handler filter }
-                      callHandlers { function handler }
-                      eventHandlers { event handler topic0 }
-                    }
-                  }
-                }
-              }
-            }
-            "#,
-        )
-        .expect("invalid query for dynamic data sources");
-        let variables = Some(QueryVariables::new(HashMap::from_iter(
-            vec![
-                (String::from("id"), q::Value::String(deployment.to_string())),
-                (String::from("skip"), q::Value::Int(skip.into())),
-            ]
-            .into_iter(),
-        )));
-
-        Ok(Query::new(schema, document, variables, None))
-    }
-
-    fn parse_data_sources(
-        &self,
-        deployment_id: &SubgraphDeploymentId,
-        query_result: q::Value,
-    ) -> Result<Vec<UnresolvedDataSource>, Error> {
-        let data = match query_result {
-            q::Value::Object(obj) => Ok(obj),
-            _ => Err(format_err!(
-                "Query result for deployment `{}` is not an on object",
-                deployment_id,
-            )),
-        }?;
-
-        // Extract the deployment from the query result
-        let deployment = match data.get("subgraphDeployment") {
-            Some(q::Value::Object(obj)) => Ok(obj),
-            _ => Err(format_err!(
-                "Deployment `{}` is not an object",
-                deployment_id,
-            )),
-        }?;
-
-        // Extract the dynamic data sources from the query result
-        let values = match deployment.get("dynamicDataSources") {
-            Some(q::Value::List(objs)) => {
-                if objs.iter().all(|obj| match obj {
-                    q::Value::Object(_) => true,
-                    _ => false,
-                }) {
-                    Ok(objs)
-                } else {
-                    Err(format_err!(
-                        "Not all dynamic data sources of deployment `{}` are objects",
-                        deployment_id
-                    ))
-                }
-            }
-            _ => Err(format_err!(
-                "Dynamic data sources of deployment `{}` are not a list",
-                deployment_id
-            )),
-        }?;
-
-        // Parse the raw data sources into typed entities
-        let entities = values.iter().try_fold(vec![], |mut entities, value| {
-            entities.push(UnresolvedDataSource::try_from_value(value)?);
-            Ok(entities)
-        });
-
-        entities.map_err(|e: Error| {
-            format_err!(
-                "Failed to parse dynamic data source entities of deployment `{}`: {}",
-                deployment_id,
-                e
-            )
-        })
-    }
-
-    async fn resolve_data_sources(
-        &self,
-        unresolved_data_sources: Vec<UnresolvedDataSource>,
-        logger: &Logger,
-    ) -> Result<Vec<DataSource>, Error> {
-        // Resolve the data sources and return them
-        let mut result = Vec::new();
-        for item in unresolved_data_sources.into_iter() {
-            let resolved = item.resolve(&*self.link_resolver, logger).await?;
-            result.push(resolved);
-        }
-        Ok(result)
+    pub fn new(store: Arc<S>) -> Self {
+        Self { store }
     }
 }
 
 #[async_trait]
-impl<L, Q, S> DataSourceLoaderTrait for DataSourceLoader<L, Q, S>
+impl<S> DataSourceLoaderTrait for DataSourceLoader<S>
 where
-    L: LinkResolver,
-    Q: GraphQlRunner,
     S: Store + SubgraphDeploymentStore,
 {
     async fn load_dynamic_data_sources(
         &self,
         deployment_id: SubgraphDeploymentId,
         logger: Logger,
+        manifest: SubgraphManifest,
     ) -> Result<Vec<DataSource>, Error> {
         let start_time = Instant::now();
 
+        let template_map: HashMap<&str, &DataSourceTemplate> = HashMap::from_iter(
+            manifest
+                .templates
+                .iter()
+                .map(|template| (template.name.as_str(), template)),
+        );
         let mut data_sources = vec![];
 
-        loop {
-            let skip = data_sources.len() as i32;
-            let query = self.dynamic_data_sources_query(&deployment_id, skip)?;
-            let query_result = self
-                .graphql_runner
-                .cheap_clone()
-                .query_metadata(query)
-                .await?;
-            let unresolved_data_sources =
-                self.parse_data_sources(&deployment_id, query_result.as_ref().clone())?;
-            let next_data_sources = self
-                .resolve_data_sources(unresolved_data_sources, &logger)
-                .await?;
-
-            if next_data_sources.is_empty() {
-                break;
-            }
-
-            data_sources.extend(next_data_sources);
+        for stored in self.store.load_dynamic_data_sources(&deployment_id)? {
+            let StoredDynamicDataSource {
+                name,
+                source,
+                context,
+            } = stored;
+
+            let template = template_map.get(name.as_str()).ok_or_else(|| {
+                format_err!(
+                    "deployment `{}` does not have a template called `{}`",
+                    deployment_id.as_str(),
+                    name
+                )
+            })?;
+            let context = context
+                .map(|ctx| serde_json::from_str::<Entity>(&ctx))
+                .transpose()?;
+
+            let ds = DataSource {
+                kind: template.kind.clone(),
+                network: template.network.clone(),
+                name,
+                source,
+                mapping: template.mapping.clone(),
+                context,
+                templates: Vec::new(),
+            };
+            data_sources.push(ds);
         }
 
         trace!(

core/src/subgraph/provider.rs

@@ -88,11 +88,7 @@ where
         let store = self.store.clone();
         let subgraph_id = id.clone();
 
-        let loader = Arc::new(DataSourceLoader::new(
-            store.clone(),
-            self.resolver.clone(),
-            self.graphql_runner.clone(),
-        ));
+        let loader = Arc::new(DataSourceLoader::new(store.clone()));
 
         let link = format!("/ipfs/{}", id);
 
@@ -112,7 +108,7 @@ where
             .await?;
 
             let data_sources = loader
-                .load_dynamic_data_sources(id.clone(), logger.clone())
+                .load_dynamic_data_sources(id.clone(), logger.clone(), subgraph.clone())
                 .map_err(SubgraphAssignmentProviderError::DynamicDataSourcesError)
                 .await?;
 

graph/src/components/subgraph/loader.rs

@@ -8,5 +8,6 @@ pub trait DataSourceLoader {
         &self,
         id: SubgraphDeploymentId,
         logger: Logger,
+        manifest: SubgraphManifest,
     ) -> Result<Vec<DataSource>, Error>;
 }