feat: implement a query interface for turborepo (#8977)

### Description

Adds a GraphQL query interface to turborepo via the `turbo query`
command. If you pass it a query string, then it directly runs the query.
If you pass it a file, it reads the file and runs the query. If you
don't pass a query, then it opens up a server with GraphiQL (we can ITG
this interface).


### Testing Instructions
Added tests to `affected.t` and `command-query.t`

---------

Co-authored-by: Chris Olszewski <[email protected]>
Co-authored-by: Thomas Knickman <[email protected]>

Cargo.toml

@@ -100,8 +100,8 @@ async-compression = { version = "0.3.13", default-features = false, features = [
 ] }
 async-trait = "0.1.64"
 atty = "0.2.14"
-axum = "0.6.2"
-axum-server = "0.4.4"
+axum = "0.7.5"
+axum-server = "0.7.1"
 biome_console = { version = "0.5.7" }
 biome_deserialize = { version = "0.6.0", features = ["serde"] }
 biome_deserialize_macros = { version = "0.6.0" }

crates/turborepo-cache/src/lib.rs

@@ -103,7 +103,7 @@ pub struct CacheHitMetadata {
     pub time_saved: u64,
 }
 
-#[derive(Debug, Default)]
+#[derive(Clone, Debug, Default)]
 pub struct CacheOpts {
     pub cache_dir: Utf8PathBuf,
     pub remote_cache_read_only: bool,

crates/turborepo-lib/Cargo.toml

@@ -32,6 +32,8 @@ turborepo-vercel-api-mock = { workspace = true }
 workspace = true
 
 [dependencies]
+async-graphql = "7.0.7"
+async-graphql-axum = "7.0.7"
 atty = { workspace = true }
 axum = { workspace = true }
 biome_deserialize = { workspace = true }

crates/turborepo-lib/src/cli/error.rs

@@ -10,6 +10,7 @@ use turborepo_ui::{color, BOLD, GREY};
 use crate::{
     commands::{bin, generate, ls, prune, run::get_signal, CommandBase},
     daemon::DaemonError,
+    query,
     rewrite_json::RewriteError,
     run,
     run::{builder::RunBuilder, watch},
@@ -54,6 +55,9 @@ pub enum Error {
     #[diagnostic(transparent)]
     Run(#[from] run::Error),
     #[error(transparent)]
+    #[diagnostic(transparent)]
+    Query(#[from] query::Error),
+    #[error(transparent)]
     SerdeJson(#[from] serde_json::Error),
     #[error(transparent)]
     #[diagnostic(transparent)]

crates/turborepo-lib/src/cli/mod.rs

@@ -22,7 +22,7 @@ use turborepo_ui::{ColorConfig, GREY};
 use crate::{
     cli::error::print_potential_tasks,
     commands::{
-        bin, config, daemon, generate, link, login, logout, ls, prune, run, scan, telemetry,
+        bin, config, daemon, generate, link, login, logout, ls, prune, query, run, scan, telemetry,
         unlink, CommandBase,
     },
     get_version,
@@ -588,6 +588,13 @@ pub enum Command {
         #[clap(flatten)]
         execution_args: Box<ExecutionArgs>,
     },
+    /// Query your monorepo using GraphQL. If no query is provided, spins up a
+    /// GraphQL server with GraphiQL.
+    #[clap(hide = true)]
+    Query {
+        /// The query to run, either a file path or a query string
+        query: Option<String>,
+    },
     Watch(Box<ExecutionArgs>),
     /// Unlink the current directory from your Vercel organization and disable
     /// Remote Caching
@@ -1198,6 +1205,7 @@ pub async fn run(
             let filter = filter.clone();
             let packages = packages.clone();
             let base = CommandBase::new(cli_args, repo_root, version, color_config);
+
             ls::run(base, packages, event, filter, affected, output).await?;
 
             Ok(0)
@@ -1301,6 +1309,17 @@ pub async fn run(
             })?;
             Ok(exit_code)
         }
+        Command::Query { query } => {
+            warn!("query command is experimental and may change in the future");
+            let query = query.clone();
+            let event = CommandEventBuilder::new("query").with_parent(&root_telemetry);
+            event.track_call();
+            let base = CommandBase::new(cli_args, repo_root, version, color_config);
+
+            let query = query::run(base, event, query).await?;
+
+            Ok(query)
+        }
         Command::Watch(_) => {
             let event = CommandEventBuilder::new("watch").with_parent(&root_telemetry);
             event.track_call();

crates/turborepo-lib/src/commands/mod.rs

@@ -22,6 +22,7 @@ pub(crate) mod login;
 pub(crate) mod logout;
 pub(crate) mod ls;
 pub(crate) mod prune;
+pub(crate) mod query;
 pub(crate) mod run;
 pub(crate) mod scan;
 pub(crate) mod telemetry;

crates/turborepo-lib/src/commands/query.rs

@@ -0,0 +1,103 @@
+use std::fs;
+
+use async_graphql::{EmptyMutation, EmptySubscription, Schema, ServerError};
+use miette::{Diagnostic, Report, SourceSpan};
+use thiserror::Error;
+use turbopath::AbsoluteSystemPathBuf;
+use turborepo_telemetry::events::command::CommandEventBuilder;
+
+use crate::{
+    cli::Command,
+    commands::{run::get_signal, CommandBase},
+    query,
+    query::{Error, Query},
+    run::builder::RunBuilder,
+    signal::SignalHandler,
+};
+
+#[derive(Debug, Diagnostic, Error)]
+#[error("{message}")]
+struct QueryError {
+    message: String,
+    #[source_code]
+    query: String,
+    #[label]
+    span: Option<SourceSpan>,
+    #[label]
+    span2: Option<SourceSpan>,
+    #[label]
+    span3: Option<SourceSpan>,
+}
+
+impl QueryError {
+    fn get_index_from_row_column(query: &str, row: usize, column: usize) -> usize {
+        let mut index = 0;
+        for line in query.lines().take(row + 1) {
+            index += line.len() + 1;
+        }
+        index + column
+    }
+    fn new(server_error: ServerError, query: String) -> Self {
+        let span: Option<SourceSpan> = server_error.locations.first().map(|location| {
+            let idx =
+                Self::get_index_from_row_column(query.as_ref(), location.line, location.column);
+            (idx, idx + 1).into()
+        });
+
+        QueryError {
+            message: server_error.message,
+            query,
+            span,
+            span2: None,
+            span3: None,
+        }
+    }
+}
+
+pub async fn run(
+    mut base: CommandBase,
+    telemetry: CommandEventBuilder,
+    query: Option<String>,
+) -> Result<i32, Error> {
+    let signal = get_signal()?;
+    let handler = SignalHandler::new(signal);
+
+    // We fake a run command, so we can construct a `Run` type
+    base.args_mut().command = Some(Command::Run {
+        run_args: Box::default(),
+        execution_args: Box::default(),
+    });
+
+    let run_builder = RunBuilder::new(base)?;
+    let run = run_builder.build(&handler, telemetry).await?;
+
+    if let Some(query) = query {
+        let trimmed_query = query.trim();
+        // If the arg starts with "query" or "mutation", and ends in a bracket, it's
+        // likely a direct query If it doesn't, it's a file path, so we need to
+        // read it
+        let query = if (trimmed_query.starts_with("query") || trimmed_query.starts_with("mutation"))
+            && trimmed_query.ends_with('}')
+        {
+            query
+        } else {
+            fs::read_to_string(AbsoluteSystemPathBuf::from_unknown(run.repo_root(), query))?
+        };
+
+        let schema = Schema::new(Query::new(run), EmptyMutation, EmptySubscription);
+
+        let result = schema.execute(&query).await;
+        if result.errors.is_empty() {
+            println!("{}", serde_json::to_string_pretty(&result)?);
+        } else {
+            for error in result.errors {
+                let error = QueryError::new(error, query.clone());
+                eprintln!("{:?}", Report::new(error));
+            }
+        }
+    } else {
+        query::run_server(run, handler).await?;
+    }
+
+    Ok(0)
+}

crates/turborepo-lib/src/daemon/default_timeout_layer.rs

@@ -85,7 +85,7 @@ mod test {
         sync::{Arc, Mutex},
     };
 
-    use axum::http::HeaderValue;
+    use reqwest::header::HeaderValue;
     use test_case::test_case;
 
     use super::*;

crates/turborepo-lib/src/lib.rs

@@ -29,6 +29,7 @@ mod opts;
 mod package_changes_watcher;
 mod panic_handler;
 mod process;
+mod query;
 mod rewrite_json;
 mod run;
 mod shim;

crates/turborepo-lib/src/opts.rs

@@ -37,7 +37,7 @@ pub enum Error {
     Config(#[from] crate::config::Error),
 }
 
-#[derive(Debug)]
+#[derive(Debug, Clone)]
 pub struct Opts {
     pub cache_opts: CacheOpts,
     pub run_opts: RunOpts,
@@ -127,7 +127,7 @@ struct OptsInputs<'a> {
     api_auth: &'a Option<APIAuth>,
 }
 
-#[derive(Debug, Default)]
+#[derive(Clone, Debug, Default)]
 pub struct RunCacheOpts {
     pub(crate) skip_reads: bool,
     pub(crate) skip_writes: bool,
@@ -144,7 +144,7 @@ impl<'a> From<OptsInputs<'a>> for RunCacheOpts {
     }
 }
 
-#[derive(Debug)]
+#[derive(Clone, Debug)]
 pub struct RunOpts {
     pub(crate) tasks: Vec<String>,
     pub(crate) concurrency: u32,
@@ -183,7 +183,7 @@ impl RunOpts {
     }
 }
 
-#[derive(Debug)]
+#[derive(Clone, Debug)]
 pub enum GraphOpts {
     Stdout,
     File(String),
@@ -302,7 +302,7 @@ impl From<LogPrefix> for ResolvedLogPrefix {
     }
 }
 
-#[derive(Debug)]
+#[derive(Clone, Debug)]
 pub struct ScopeOpts {
     pub pkg_inference_root: Option<AnchoredSystemPathBuf>,
     pub global_deps: Vec<String>,