Implemented web UI

Author: NicholasLYang

Cargo.lock

@@ -39,7 +39,6 @@ pub trait LoginServer {
         token: Arc<OnceCell<String>>,
     ) -> Result<(), Error>;
 }
-
 /// A struct that implements LoginServer.
 ///
 /// Listens on 127.0.0.1 and a port that's passed in.

crates/turborepo-auth/src/login_server.rs

@@ -44,15 +44,15 @@ pub async fn run(base: CommandBase, telemetry: CommandEventBuilder) -> Result<i3
             .build(&handler, telemetry)
             .await?;
 
-        let (sender, handle) = run.start_experimental_ui()?.unzip();
+        let (sender, handle) = run.start_ui()?.unzip();
+
         let result = run.run(sender.clone(), false).await;
 
         if let Some(analytics_handle) = analytics_handle {
             analytics_handle.close_with_timeout().await;
         }
 
-        if let (Some(handle), Some(sender)) = (handle, sender) {
-            sender.stop();
+        if let Some(handle) = handle {
             if let Err(e) = handle.await.expect("render thread panicked") {
                 error!("error encountered rendering tui: {e}");
             }

crates/turborepo-lib/src/commands/run.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/daemon/default_timeout_layer.rs

@@ -12,6 +12,7 @@ use crate::{
     commands::CommandBase,
     config::ConfigurationOptions,
     run::task_id::TaskId,
+    turbo_json::UIMode,
 };
 
 #[derive(Debug, Error)]
@@ -160,6 +161,7 @@ pub struct RunOpts {
     pub summarize: Option<Option<bool>>,
     pub(crate) experimental_space_id: Option<String>,
     pub is_github_actions: bool,
+    pub ui_mode: UIMode,
 }
 
 impl RunOpts {
@@ -260,6 +262,7 @@ impl<'a> TryFrom<OptsInputs<'a>> for RunOpts {
             dry_run: inputs.run_args.dry_run,
             env_mode: inputs.config.env_mode(),
             is_github_actions,
+            ui_mode: inputs.config.ui(),
         })
     }
 }
@@ -379,6 +382,7 @@ mod test {
     use crate::{
         cli::DryRunMode,
         opts::{Opts, RunCacheOpts, ScopeOpts},
+        turbo_json::UIMode,
     };
 
     #[derive(Default)]
@@ -465,6 +469,7 @@ mod test {
             only: opts_input.only,
             dry_run: opts_input.dry_run,
             graph: None,
+            ui_mode: UIMode::Tui,
             single_package: false,
             log_prefix: crate::opts::ResolvedLogPrefix::Task,
             log_order: crate::opts::ResolvedLogOrder::Stream,

crates/turborepo-lib/src/opts.rs

@@ -56,7 +56,6 @@ pub struct RunBuilder {
     repo_root: AbsoluteSystemPathBuf,
     ui: UI,
     version: &'static str,
-    ui_mode: UIMode,
     api_client: APIClient,
     analytics_sender: Option<AnalyticsSender>,
     // In watch mode, we can have a changed package that we want to serve as an entrypoint.
@@ -77,13 +76,12 @@ impl RunBuilder {
         let allow_missing_package_manager = config.allow_no_package_manager();
 
         let version = base.version();
-        let ui_mode = config.ui();
         let processes = ProcessManager::new(
             // We currently only use a pty if the following are met:
             // - we're attached to a tty
             atty::is(atty::Stream::Stdout) &&
             // - if we're on windows, we're using the UI
-            (!cfg!(windows) || matches!(ui_mode, UIMode::Tui)),
+            (!cfg!(windows) || matches!(opts.run_opts.ui_mode, UIMode::Tui)),
         );
         let CommandBase { repo_root, ui, .. } = base;
 
@@ -94,7 +92,6 @@ impl RunBuilder {
             repo_root,
             ui,
             version,
-            ui_mode,
             api_auth,
             analytics_sender: None,
             entrypoint_packages: None,
@@ -384,7 +381,6 @@ impl RunBuilder {
         Ok(Run {
             version: self.version,
             ui: self.ui,
-            ui_mode: self.ui_mode,
             start_at,
             processes: self.processes,
             run_telemetry,
@@ -439,7 +435,11 @@ impl RunBuilder {
 
         if !self.opts.run_opts.parallel {
             engine
-                .validate(pkg_dep_graph, self.opts.run_opts.concurrency, self.ui_mode)
+                .validate(
+                    pkg_dep_graph,
+                    self.opts.run_opts.concurrency,
+                    self.opts.run_opts.ui_mode,
+                )
                 .map_err(Error::EngineValidation)?;
         }
 

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

@@ -57,5 +57,7 @@ pub enum Error {
     #[error(transparent)]
     Daemon(#[from] daemon::DaemonError),
     #[error(transparent)]
+    UI(#[from] turborepo_ui::Error),
+    #[error(transparent)]
     Tui(#[from] tui::Error),
 }

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

@@ -31,7 +31,9 @@ use turborepo_env::EnvironmentVariableMap;
 use turborepo_repository::package_graph::{PackageGraph, PackageName, PackageNode};
 use turborepo_scm::SCM;
 use turborepo_telemetry::events::generic::GenericEventBuilder;
-use turborepo_ui::{cprint, cprintln, tui, tui::AppSender, BOLD_GREY, GREY, UI};
+use turborepo_ui::{
+    cprint, cprintln, sender::UISender, tui, tui::TuiSender, wui::WebUISender, BOLD_GREY, GREY, UI,
+};
 
 pub use crate::run::error::Error;
 use crate::{
@@ -69,9 +71,12 @@ pub struct Run {
     task_access: TaskAccess,
     daemon: Option<DaemonClient<DaemonConnector>>,
     should_print_prelude: bool,
-    ui_mode: UIMode,
 }
 
+type UIResult = Result<Option<(UISender, JoinHandle<Result<(), turborepo_ui::Error>>)>, Error>;
+type WuiResult = Result<Option<(WebUISender, JoinHandle<Result<(), turborepo_ui::Error>>)>, Error>;
+type TuiResult = Result<Option<(TuiSender, JoinHandle<Result<(), turborepo_ui::Error>>)>, Error>;
+
 impl Run {
     fn has_persistent_tasks(&self) -> bool {
         self.engine.has_persistent_tasks
@@ -173,19 +178,36 @@ impl Run {
     }
 
     pub fn has_tui(&self) -> bool {
-        self.ui_mode.use_tui()
+        self.opts.run_opts.ui_mode.use_tui()
     }
 
     pub fn should_start_ui(&self) -> Result<bool, Error> {
-        Ok(self.ui_mode.use_tui()
+        Ok(self.opts.run_opts.ui_mode.use_tui()
             && self.opts.run_opts.dry_run.is_none()
             && tui::terminal_big_enough()?)
     }
 
+    pub fn start_ui(&self) -> UIResult {
+        match self.opts.run_opts.ui_mode {
+            UIMode::Tui => self
+                .start_terminal_ui()
+                .map(|res| res.map(|(sender, handle)| (UISender::Tui(sender), handle))),
+            UIMode::Stream => Ok(None),
+            UIMode::Web => self
+                .start_web_ui()
+                .map(|res| res.map(|(sender, handle)| (UISender::Wui(sender), handle))),
+        }
+    }
+    pub fn start_web_ui(&self) -> WuiResult {
+        let (tx, rx) = tokio::sync::broadcast::channel(100);
+
+        let handle = tokio::spawn(turborepo_ui::wui::start_ws_server(rx));
+
+        Ok(Some((WebUISender { tx }, handle)))
+    }
+
     #[allow(clippy::type_complexity)]
-    pub fn start_experimental_ui(
-        &self,
-    ) -> Result<Option<(AppSender, JoinHandle<Result<(), tui::Error>>)>, Error> {
+    pub fn start_terminal_ui(&self) -> TuiResult {
         // Print prelude here as this needs to happen before the UI is started
         if self.should_print_prelude {
             self.print_run_prelude();
@@ -201,8 +223,8 @@ impl Run {
             return Ok(None);
         }
 
-        let (sender, receiver) = AppSender::new();
-        let handle = tokio::task::spawn_blocking(move || tui::run_app(task_names, receiver));
+        let (sender, receiver) = TuiSender::new();
+        let handle = tokio::task::spawn_blocking(move || Ok(tui::run_app(task_names, receiver)?));
 
         Ok(Some((sender, handle)))
     }
@@ -214,11 +236,7 @@ impl Run {
         }
     }
 
-    pub async fn run(
-        &mut self,
-        experimental_ui_sender: Option<AppSender>,
-        is_watch: bool,
-    ) -> Result<i32, Error> {
+    pub async fn run(&mut 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();
@@ -405,7 +423,7 @@ impl Run {
             self.processes.clone(),
             &self.repo_root,
             global_env,
-            experimental_ui_sender,
+            ui_sender,
             is_watch,
         );