node: Support subgraph datasource in gnd

Author: incrypto32

node/src/bin/dev.rs

@@ -1,4 +1,4 @@
-use std::{collections::HashMap, path::Path, sync::Arc};
+use std::{path::Path, sync::Arc};
 
 use anyhow::{Context, Result};
 use clap::Parser;
@@ -49,6 +49,14 @@ pub struct DevOpt {
     )]
     pub manifests: Vec<String>,
 
+    #[clap(
+        long,
+        value_name = "ALIAS:MANIFEST:[BUILD_DIR]",
+        value_delimiter = ',',
+        help = "The location of the source subgraph manifest files. This is used to resolve aliases in the manifest files for subgraph data sources. The format is ALIAS:MANIFEST:[BUILD_DIR], where ALIAS is the alias name, BUILD_DIR is the build directory relative to the manifest file, and MANIFEST is the manifest file location."
+    )]
+    pub sources: Vec<String>,
+
     #[clap(
         long,
         help = "The location of the database directory.",
@@ -126,8 +134,9 @@ async fn main() -> Result<()> {
     let (tx, rx) = mpsc::channel(1);
     let opt = build_args(&dev_opt, &db.connection_uri())?;
 
-    let manifests_paths = parse_manifest_args(dev_opt.manifests, &logger)?;
-    let file_link_resolver = Arc::new(FileLinkResolver::new(None, HashMap::new()));
+    let (manifests_paths, source_subgraph_aliases) =
+        parse_manifest_args(dev_opt.manifests, dev_opt.sources, &logger)?;
+    let file_link_resolver = Arc::new(FileLinkResolver::new(None, source_subgraph_aliases.clone()));
 
     let ctx = DevModeContext {
         watch: dev_opt.watch,
@@ -142,8 +151,14 @@ async fn main() -> Result<()> {
 
     if dev_opt.watch {
         graph::spawn_blocking(async move {
-            let result =
-                watch_subgraphs(&logger, manifests_paths, vec!["pgtemp-*".to_string()], tx).await;
+            let result = watch_subgraphs(
+                &logger,
+                manifests_paths,
+                source_subgraph_aliases,
+                vec!["pgtemp-*".to_string()],
+                tx,
+            )
+            .await;
             if let Err(e) = result {
                 error!(logger, "Error watching subgraphs"; "error" => e.to_string());
                 std::process::exit(1);

node/src/dev/helpers.rs

@@ -126,6 +126,7 @@ pub async fn watch_subgraph_updates(
     std::process::exit(1);
 }
 
+/// Parse an alias string into a tuple of (alias_name, manifest, Option<build_dir>)
 pub fn parse_alias(alias: &str) -> anyhow::Result<(String, String, Option<String>)> {
     let mut split = alias.split(':');
     let alias_name = split.next();
@@ -139,12 +140,13 @@ pub fn parse_alias(alias: &str) -> anyhow::Result<(String, String, Option<String
     }
 
     let alias_name = alias_name.unwrap().to_owned();
-    let (build_dir, manifest) = parse_manifest_arg(alias_value.unwrap())
+    let (manifest, build_dir) = parse_manifest_arg(alias_value.unwrap())
         .with_context(|| format!("While parsing alias '{}'", alias))?;
 
-    Ok((alias_name, build_dir, manifest))
+    Ok((alias_name, manifest, build_dir))
 }
 
+/// Parse a manifest string into a tuple of (manifest, Option<build_dir>)
 pub fn parse_manifest_arg(value: &str) -> anyhow::Result<(String, Option<String>)> {
     match value.split_once(':') {
         Some((manifest, build_dir)) if !manifest.is_empty() => {

node/src/dev/watcher.rs

@@ -4,18 +4,33 @@ use graph::prelude::{DeploymentHash, SubgraphName};
 use graph::slog::{self, error, info, Logger};
 use graph::tokio::sync::mpsc::Sender;
 use notify::{recommended_watcher, Event, RecursiveMode, Watcher};
+use std::collections::HashMap;
 use std::path::{Path, PathBuf};
 use std::sync::mpsc;
 use std::time::Duration;
 
-use super::helpers::parse_manifest_arg;
+use super::helpers::{parse_alias, parse_manifest_arg};
 
 const WATCH_DELAY: Duration = Duration::from_secs(5);
 const DEFAULT_BUILD_DIR: &str = "build";
 
 // Parses manifest arguments and returns a vector of paths to the manifest files
-pub fn parse_manifest_args(manifests: Vec<String>, logger: &Logger) -> Result<Vec<PathBuf>> {
+pub fn parse_manifest_args(
+    manifests: Vec<String>,
+    subgraph_sources: Vec<String>,
+    logger: &Logger,
+) -> Result<(Vec<PathBuf>, HashMap<String, PathBuf>)> {
     let mut manifests_paths = Vec::new();
+    let mut source_subgraph_aliases = HashMap::new();
+
+    for subgraph_source in subgraph_sources {
+        let (alias_name, manifest_path_str, build_dir_opt) = parse_alias(&subgraph_source)?;
+        let manifest_path =
+            process_manifest(build_dir_opt, &manifest_path_str, Some(&alias_name), logger)?;
+
+        manifests_paths.push(manifest_path.clone());
+        source_subgraph_aliases.insert(alias_name, manifest_path);
+    }
 
     for manifest_str in manifests {
         let (manifest_path_str, build_dir_opt) = parse_manifest_arg(&manifest_str)
@@ -27,7 +42,7 @@ pub fn parse_manifest_args(manifests: Vec<String>, logger: &Logger) -> Result<Ve
         manifests_paths.push(built_manifest_path);
     }
 
-    Ok(manifests_paths)
+    Ok((manifests_paths, source_subgraph_aliases))
 }
 
 /// Helper function to process a manifest
@@ -103,12 +118,20 @@ fn process_manifest(
 pub async fn watch_subgraphs(
     logger: &Logger,
     manifests_paths: Vec<PathBuf>,
+    source_subgraph_aliases: HashMap<String, PathBuf>,
     exclusions: Vec<String>,
     sender: Sender<(DeploymentHash, SubgraphName)>,
 ) -> Result<()> {
     let logger = logger.new(slog::o!("component" => "Watcher"));
 
-    watch_subgraph_dirs(&logger, manifests_paths, exclusions, sender).await?;
+    watch_subgraph_dirs(
+        &logger,
+        manifests_paths,
+        source_subgraph_aliases,
+        exclusions,
+        sender,
+    )
+    .await?;
     Ok(())
 }
 
@@ -117,6 +140,7 @@ pub async fn watch_subgraphs(
 pub async fn watch_subgraph_dirs(
     logger: &Logger,
     manifests_paths: Vec<PathBuf>,
+    source_subgraph_aliases: HashMap<String, PathBuf>,
     exclusions: Vec<String>,
     sender: Sender<(DeploymentHash, SubgraphName)>,
 ) -> Result<()> {
@@ -159,7 +183,15 @@ pub async fn watch_subgraph_dirs(
     }
 
     // Process file change events
-    process_file_events(logger, rx, &exclusion_set, &manifests_paths, sender).await
+    process_file_events(
+        logger,
+        rx,
+        &exclusion_set,
+        &manifests_paths,
+        &source_subgraph_aliases,
+        sender,
+    )
+    .await
 }
 
 /// Processes file change events and triggers redeployments
@@ -168,6 +200,7 @@ async fn process_file_events(
     rx: mpsc::Receiver<Result<Event, notify::Error>>,
     exclusion_set: &GlobSet,
     manifests_paths: &Vec<PathBuf>,
+    source_subgraph_aliases: &HashMap<String, PathBuf>,
     sender: Sender<(DeploymentHash, SubgraphName)>,
 ) -> Result<()> {
     loop {
@@ -203,7 +236,7 @@ async fn process_file_events(
         }
 
         // Redeploy all subgraphs
-        redeploy_all_subgraphs(logger, manifests_paths, &sender).await?;
+        redeploy_all_subgraphs(logger, manifests_paths, source_subgraph_aliases, &sender).await?;
     }
 }
 
@@ -223,15 +256,24 @@ fn is_relevant_event(event: &Event, watched_dirs: Vec<PathBuf>, exclusion_set: &
 async fn redeploy_all_subgraphs(
     logger: &Logger,
     manifests_paths: &Vec<PathBuf>,
+    source_subgraph_aliases: &HashMap<String, PathBuf>,
     sender: &Sender<(DeploymentHash, SubgraphName)>,
 ) -> Result<()> {
     info!(logger, "File change detected, redeploying all subgraphs");
     let mut count = 0;
     for manifest_path in manifests_paths {
+        let alias_name = source_subgraph_aliases
+            .iter()
+            .find(|(_, path)| path == &manifest_path)
+            .map(|(name, _)| name);
+
+        let id = alias_name
+            .map(|s| s.to_owned())
+            .unwrap_or_else(|| manifest_path.display().to_string());
+
         let _ = sender
             .send((
-                DeploymentHash::new(manifest_path.display().to_string())
-                    .map_err(|_| anyhow!("Failed to create deployment hash"))?,
+                DeploymentHash::new(id).map_err(|_| anyhow!("Failed to create deployment hash"))?,
                 SubgraphName::new(format!("subgraph-{}", count))
                     .map_err(|_| anyhow!("Failed to create subgraph name"))?,
             ))