refactor: Add query resolvers into web UI

crates/turborepo-lib/Cargo.toml

@@ -109,6 +109,7 @@ tokio-stream = { version = "0.1.12", features = ["net"] }
 tokio-util = { version = "0.7.7", features = ["compat"] }
 tonic = { version = "0.11.0", features = ["transport"] }
 tower = "0.4.13"
+tower-http = { version = "0.5.2", features = ["cors"] }
 tracing-appender = "0.2.2"
 tracing-chrome = "0.7.1"
 tracing-subscriber = { version = "0.3.17", features = ["env-filter"] }

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

@@ -1,4 +1,4 @@
-use std::fs;
+use std::{fs, sync::Arc};
 
 use async_graphql::{EmptyMutation, EmptySubscription, Schema, ServerError};
 use miette::{Diagnostic, Report, SourceSpan};
@@ -10,7 +10,7 @@ use crate::{
  cli::Command,
  commands::{run::get_signal, CommandBase},
  query,
- query::{Error, Query},
+ query::{Error, RepositoryQuery},
  run::builder::RunBuilder,
  signal::SignalHandler,
 };
@@ -84,7 +84,11 @@ pub async fn run(
  fs::read_to_string(AbsoluteSystemPathBuf::from_unknown(run.repo_root(), query))?
  };
 
- let schema = Schema::new(Query::new(run), EmptyMutation, EmptySubscription);
+ let schema = Schema::new(
+ RepositoryQuery::new(Arc::new(run)),
+ EmptyMutation,
+ EmptySubscription,
+ );
 
  let result = schema.execute(&query).await;
  if result.errors.is_empty() {

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

@@ -1,4 +1,4 @@
-use std::future::Future;
+use std::{future::Future, sync::Arc};
 
 use tracing::error;
 use turborepo_telemetry::events::command::CommandEventBuilder;
@@ -40,10 +40,12 @@ pub async fn run(base: CommandBase, telemetry: CommandEventBuilder) -> Result<i3
 
  let run_fut = async {
  let (analytics_sender, analytics_handle) = run_builder.start_analytics();
- let mut run = run_builder
- .with_analytics_sender(analytics_sender)
- .build(&handler, telemetry)
- .await?;
+ let run = Arc::new(
+ run_builder
+ .with_analytics_sender(analytics_sender)
+ .build(&handler, telemetry)
+ .await?,
+ );
 
  let (sender, handle) = run.start_ui()?.unzip();
 

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

@@ -41,13 +41,13 @@ pub enum Error {
  Path(#[from] turbopath::PathError),
 }
 
-pub struct Query {
+pub struct RepositoryQuery {
  run: Arc<Run>,
 }
 
-impl Query {
- pub fn new(run: Run) -> Self {
- Self { run: Arc::new(run) }
+impl RepositoryQuery {
+ pub fn new(run: Arc<Run>) -> Self {
+ Self { run }
  }
 }
 
@@ -267,7 +267,7 @@ impl PackagePredicate {
 }
 
 #[Object]
-impl Query {
+impl RepositoryQuery {
  async fn affected_packages(
  &self,
  base: Option<String>,
@@ -339,12 +339,16 @@ impl Query {
  }
 }
 
-async fn graphiql() -> impl IntoResponse {
+pub async fn graphiql() -> impl IntoResponse {
  response::Html(GraphiQLSource::build().endpoint("/").finish())
 }
 
 pub async fn run_server(run: Run, signal: SignalHandler) -> Result<(), Error> {
- let schema = Schema::new(Query::new(run), EmptyMutation, EmptySubscription);
+ let schema = Schema::new(
+ RepositoryQuery::new(Arc::new(run)),
+ EmptyMutation,
+ EmptySubscription,
+ );
  let app = Router::new().route("/", get(graphiql).post_service(GraphQL::new(schema)));
 
  let subscriber = signal.subscribe().ok_or(Error::NoSignalHandler)?;

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

@@ -0,0 +1 @@
+

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

@@ -10,6 +10,7 @@ mod scope;
 pub(crate) mod summary;
 pub mod task_access;
 pub mod task_id;
+mod ui;
 pub mod watch;
 
 use std::{
@@ -211,7 +212,7 @@ impl Run {
  && tui::terminal_big_enough()?)
  }
 
- pub fn start_ui(&self) -> UIResult<UISender> {
+ pub fn start_ui(self: &Arc<Self>) -> UIResult<UISender> {
  // Print prelude here as this needs to happen before the UI is started
  if self.should_print_prelude {
  self.print_run_prelude();
@@ -227,10 +228,10 @@ impl Run {
  .map(|res| res.map(|(sender, handle)| (UISender::Wui(sender), handle))),
  }
  }
- fn start_web_ui(&self) -> WuiResult {
+ fn start_web_ui(self: &Arc<Self>) -> WuiResult {
  let (tx, rx) = tokio::sync::mpsc::unbounded_channel();
 
- let handle = tokio::spawn(turborepo_ui::wui::server::start_server(rx));
+ let handle = tokio::spawn(ui::start_web_ui_server(rx, self.clone()));
 
  Ok(Some((WebUISender { tx }, handle)))
  }
@@ -260,7 +261,7 @@ impl Run {
  }
  }
 
- pub async fn run(&mut self, ui_sender: Option<UISender>, is_watch: bool) -> Result<i32, Error> {
+ pub async fn run(&self, ui_sender: Option<UISender>, is_watch: bool) -> Result<i32, Error> {
  let skip_cache_writes = self.opts.runcache_opts.skip_writes;
  if let Some(subscriber) = self.signal_handler.subscribe() {
  let run_cache = self.run_cache.clone();
@@ -356,7 +357,7 @@ impl Run {
  self.engine.task_definitions(),
  &self.repo_root,
  &self.run_telemetry,
- &mut self.daemon,
+ &self.daemon,
  )?;
 
  let root_workspace = self

crates/turborepo-lib/src/run/ui.rs

@@ -0,0 +1,55 @@
+use std::sync::Arc;
+
+use async_graphql::{EmptyMutation, EmptySubscription, MergedObject, Schema};
+use async_graphql_axum::GraphQL;
+use axum::{http::Method, routing::get, Router};
+use tokio::net::TcpListener;
+use tower_http::cors::{Any, CorsLayer};
+use turborepo_ui::wui::{event::WebUIEvent, server::SharedState};
+
+use crate::{query, query::graphiql, run::Run};
+
+pub async fn start_web_ui_server(
+ rx: tokio::sync::mpsc::UnboundedReceiver<WebUIEvent>,
+ run: Arc<Run>,
+) -> Result<(), turborepo_ui::Error> {
+ let state = SharedState::default();
+ let subscriber = turborepo_ui::wui::subscriber::Subscriber::new(rx);
+ tokio::spawn(subscriber.watch(state.clone()));
+
+ run_server(state.clone(), run).await?;
+
+ Ok(())
+}
+
+#[derive(MergedObject)]
+struct Query(turborepo_ui::wui::RunQuery, query::RepositoryQuery);
+
+async fn run_server(state: SharedState, run: Arc<Run>) -> Result<(), turborepo_ui::Error> {
+ let cors = CorsLayer::new()
+ // allow `GET` and `POST` when accessing the resource
+ .allow_methods([Method::GET, Method::POST])
+ .allow_headers(Any)
+ // allow requests from any origin
+ .allow_origin(Any);
+
+ let web_ui_query = turborepo_ui::wui::RunQuery::new(state.clone());
+ let turbo_query = query::RepositoryQuery::new(run);
+ let combined_query = Query(web_ui_query, turbo_query);
+
+ let schema = Schema::new(combined_query, EmptyMutation, EmptySubscription);
+ let app = Router::new()
+ .route("/", get(graphiql).post_service(GraphQL::new(schema)))
+ .layer(cors);
+
+ axum::serve(
+ TcpListener::bind("127.0.0.1:8000")
+ .await
+ .map_err(turborepo_ui::wui::Error::Server)?,
+ app,
+ )
+ .await
+ .map_err(turborepo_ui::wui::Error::Server)?;
+
+ Ok(())
+}

crates/turborepo-lib/src/run/watch.rs

@@ -46,7 +46,7 @@ impl ChangedPackages {
 }
 
 pub struct WatchClient {
- run: Run,
+ run: Arc<Run>,
  watched_packages: HashSet<PackageName>,
  persistent_tasks_handle: Option<PersistentRunHandle>,
  connector: DaemonConnector,
@@ -130,9 +130,11 @@ impl WatchClient {
  execution_args: execution_args.clone(),
  });
 
- let run = RunBuilder::new(new_base)?
- .build(&handler, telemetry.clone())
- .await?;
+ let run = Arc::new(
+ RunBuilder::new(new_base)?
+ .build(&handler, telemetry.clone())
+ .await?,
+ );
 
  let watched_packages = run.get_relevant_packages();
 
@@ -288,7 +290,7 @@ impl WatchClient {
  let signal_handler = self.handler.clone();
  let telemetry = self.telemetry.clone();
 
- let mut run = RunBuilder::new(new_base)?
+ let run = RunBuilder::new(new_base)?
  .with_entrypoint_packages(packages)
  .hide_prelude()
  .build(&signal_handler, telemetry)
@@ -331,7 +333,8 @@ impl WatchClient {
  self.run = RunBuilder::new(base.clone())?
  .hide_prelude()
  .build(&self.handler, self.telemetry.clone())
- .await?;
+ .await?
+ .into();
 
  self.watched_packages = self.run.get_relevant_packages();
 
@@ -357,7 +360,7 @@ impl WatchClient {
  self.persistent_tasks_handle.is_none(),
  "persistent handle should be empty before creating a new one"
  );
- let mut persistent_run = self.run.create_run_for_persistent_tasks();
+ let persistent_run = self.run.create_run_for_persistent_tasks();
  let ui_sender = self.ui_sender.clone();
  // If we have persistent tasks, we run them on a separate thread
  // since persistent tasks don't finish
@@ -369,7 +372,7 @@ impl WatchClient {
  });
 
  // But we still run the regular tasks blocking
- let mut non_persistent_run = self.run.create_run_without_persistent_tasks();
+ let non_persistent_run = self.run.create_run_without_persistent_tasks();
  Ok(non_persistent_run.run(self.ui_sender.clone(), true).await?)
  } else {
  Ok(self.run.run(self.ui_sender.clone(), true).await?)

crates/turborepo-lib/src/task_hash.rs

@@ -76,7 +76,7 @@ impl PackageInputsHashes {
  task_definitions: &HashMap<TaskId<'static>, TaskDefinition>,
  repo_root: &AbsoluteSystemPath,
  telemetry: &GenericEventBuilder,
- daemon: &mut Option<DaemonClient<DaemonConnector>>,
+ daemon: &Option<DaemonClient<DaemonConnector>>,
  ) -> Result<PackageInputsHashes, Error> {
  tracing::trace!(scm_manual=%scm.is_manual(), "scm running in {} mode", if scm.is_manual() { "manual" } else { "git" });
 

crates/turborepo-ui/Cargo.toml

@@ -35,7 +35,6 @@ serde_json = { workspace = true }
 thiserror = { workspace = true }
 tokio = { workspace = true }
 
-tower-http = { version = "0.5.2", features = ["cors"] }
 tracing = { workspace = true }
 tui-term = { workspace = true }
 turbopath = { workspace = true }

crates/turborepo-ui/src/wui/mod.rs

@@ -1,12 +1,13 @@
 //! Web UI for Turborepo. Creates a WebSocket server that can be subscribed to
 //! by a web client to display the status of tasks.
 
-mod event;
+pub mod event;
 pub mod sender;
 pub mod server;
-mod subscriber;
+pub mod subscriber;
 
 use event::WebUIEvent;
+pub use server::RunQuery;
 use thiserror::Error;
 
 #[derive(Debug, Error)]