node: Add graphman query

node/src/bin/manager.rs

@@ -1,18 +1,19 @@
 use std::{env, sync::Arc};
 
 use git_testament::{git_testament, render_testament};
-use graph::prometheus::Registry;
+use graph::{data::graphql::effort::LoadManager, prometheus::Registry};
 use graph_core::MetricsRegistry;
+use graph_graphql::prelude::GraphQlRunner;
 use lazy_static::lazy_static;
 use structopt::StructOpt;
 
 use graph::{
     log::logger,
     prelude::{info, o, slog, tokio, Logger, NodeId},
 };
-use graph_node::config;
 use graph_node::store_builder::StoreBuilder;
-use graph_store_postgres::{connection_pool::ConnectionPool, SubgraphStore, PRIMARY_SHARD};
+use graph_node::{config, manager::PanicSubscriptionManager};
+use graph_store_postgres::{connection_pool::ConnectionPool, Store, SubgraphStore, PRIMARY_SHARD};
 
 use crate::config::Config as Cfg;
 use graph_node::manager::commands;
@@ -106,6 +107,17 @@ pub enum Command {
     /// Print information about a configuration file without
     /// actually connecting to databases or network clients
     Config(ConfigCommand),
+    /// Run a GraphQL query
+    Query {
+        /// The subgraph to query
+        ///
+        /// Either a deployment id `Qm..` or a subgraph name
+        target: String,
+        /// The GraphQL query
+        query: String,
+        /// The variables in the form `key=value`
+        vars: Vec<String>,
+    },
 }
 
 #[derive(Clone, Debug, StructOpt)]
@@ -189,6 +201,25 @@ fn make_store(logger: &Logger, node_id: &NodeId, config: &Cfg) -> Arc<SubgraphSt
     StoreBuilder::make_sharded_store(logger, node_id, config, make_registry(logger))
 }
 
+fn make_runner(
+    logger: &Logger,
+    node_id: &NodeId,
+    config: &Cfg,
+) -> Arc<GraphQlRunner<Store, PanicSubscriptionManager>> {
+    let registry = make_registry(logger);
+    let network_store =
+        StoreBuilder::new(logger, node_id, config, registry.clone()).network_store(vec![]);
+    let subscription_manager = Arc::new(PanicSubscriptionManager);
+    let load_manager = Arc::new(LoadManager::new(&logger, vec![], registry.clone(), 128));
+
+    Arc::new(GraphQlRunner::new(
+        &logger,
+        network_store.clone(),
+        subscription_manager.clone(),
+        load_manager,
+    ))
+}
+
 #[tokio::main]
 async fn main() {
     let opt = Opt::from_args();
@@ -274,6 +305,14 @@ async fn main() {
             let store = make_store();
             commands::assign::reassign(store, id, node)
         }
+        Query {
+            target,
+            query,
+            vars,
+        } => {
+            let runner = make_runner(&logger, &node, &config);
+            commands::query::run(runner, target, query, vars).await
+        }
     };
     if let Err(e) = result {
         die!("error: {}", e)

node/src/manager/commands/mod.rs

@@ -1,6 +1,7 @@
 pub mod assign;
 pub mod config;
 pub mod info;
+pub mod query;
 pub mod remove;
 pub mod txn_speed;
 pub mod unused_deployments;

node/src/manager/commands/query.rs

@@ -0,0 +1,62 @@
+use std::iter::FromIterator;
+use std::{collections::HashMap, sync::Arc};
+
+use graph::{
+    data::query::QueryTarget,
+    prelude::{
+        anyhow::{self, anyhow},
+        q, serde_json, GraphQlRunner as _, Query, QueryVariables, SubgraphDeploymentId,
+        SubgraphName,
+    },
+};
+use graph_graphql::prelude::GraphQlRunner;
+use graph_store_postgres::Store;
+
+use crate::manager::PanicSubscriptionManager;
+
+pub async fn run(
+    runner: Arc<GraphQlRunner<Store, PanicSubscriptionManager>>,
+    target: String,
+    query: String,
+    vars: Vec<String>,
+) -> Result<(), anyhow::Error> {
+    let target = if target.starts_with("Qm") {
+        let id = SubgraphDeploymentId::new(target)
+            .map_err(|id| anyhow!("illegal deployment id `{}`", id))?;
+        QueryTarget::Deployment(id)
+    } else {
+        let name = SubgraphName::new(target.clone())
+            .map_err(|()| anyhow!("illegal subgraph name `{}`", target))?;
+        QueryTarget::Name(name)
+    };
+
+    let document = graphql_parser::parse_query(&query)?.into_static();
+    let vars: Vec<(String, q::Value)> = vars
+        .into_iter()
+        .map(|v| {
+            let mut pair = v.splitn(2, '=').map(|s| s.to_string());
+            let key = pair.next();
+            let value = pair
+                .next()
+                .map(|s| q::Value::String(s))
+                .unwrap_or(q::Value::Null);
+            match key {
+                Some(key) => Ok((key, value)),
+                None => Err(anyhow!(
+                    "malformed variable `{}`, it must be of the form `key=value`",
+                    v
+                )),
+            }
+        })
+        .collect::<Result<_, _>>()?;
+    let query = Query::new(
+        document,
+        Some(QueryVariables::new(HashMap::from_iter(vars))),
+    );
+
+    let res = runner.run_query(query, target, false).await;
+    let json = serde_json::to_string(&res)?;
+    println!("{}", json);
+
+    Ok(())
+}

node/src/manager/mod.rs

@@ -1,4 +1,18 @@
+use graph::{
+    components::store::SubscriptionManager,
+    prelude::{StoreEventStreamBox, SubscriptionFilter},
+};
+
 pub mod catalog;
 pub mod commands;
 pub mod deployment;
 mod display;
+
+/// A dummy subscription manager that always panics
+pub struct PanicSubscriptionManager;
+
+impl SubscriptionManager for PanicSubscriptionManager {
+    fn subscribe(&self, _: Vec<SubscriptionFilter>) -> StoreEventStreamBox {
+        panic!("we were never meant to call `subscribe`");
+    }
+}