feat(turborepo): Package Change Watching

crates/turborepo-lib/Cargo.toml

@@ -55,6 +55,7 @@ globwatch = { path = "../turborepo-globwatch" }
 hex = "0.4.3"
 hostname = "0.3.1"
 humantime = "2.1.0"
+ignore = "0.4.22"
 indicatif = { workspace = true }
 itertools = { workspace = true }
 json_comments = "0.2.1"
@@ -109,6 +110,7 @@ capnp = "0.17.2"
 const_format = "0.2.30"
 convert_case = "0.6.0"
 either.workspace = true
+futures-core = "0.3.30"
 globwalk = { version = "0.1.0", path = "../turborepo-globwalk" }
 go-parse-duration = "0.1.1"
 is-terminal = "0.4.7"

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

@@ -263,6 +263,9 @@ pub enum DaemonCommand {
     },
     /// Shows the daemon logs
     Logs,
+    #[clap(hide = true)]
+    /// Watches packages and which are changed
+    Watch,
 }
 
 #[derive(Subcommand, Copy, Clone, Debug, Serialize, PartialEq)]
@@ -1204,7 +1207,6 @@ pub async fn run(
                 let _ = logger.enable_chrome_tracing(file_path, include_args);
             }
             let base = CommandBase::new(cli_args.clone(), repo_root, version, ui);
-
             args.track(&event);
             event.track_run_code_path(CodePath::Rust);
             let exit_code = run::run(base, event).await.inspect(|code| {

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

@@ -18,6 +18,7 @@ use crate::{
         endpoint::SocketOpenError, CloseReason, DaemonConnector, DaemonConnectorError, DaemonError,
         Paths,
     },
+    run::watch::WatchClient,
     tracing::TurboSubscriber,
 };
 
@@ -27,7 +28,7 @@ const DAEMON_NOT_RUNNING_MESSAGE: &str =
 /// Runs the daemon command.
 pub async fn daemon_client(command: &DaemonCommand, base: &CommandBase) -> Result<(), DaemonError> {
     let (can_start_server, can_kill_server) = match command {
-        DaemonCommand::Status { .. } | DaemonCommand::Logs => (false, false),
+        DaemonCommand::Status { .. } | DaemonCommand::Logs | DaemonCommand::Watch => (false, false),
         DaemonCommand::Stop => (false, true),
         DaemonCommand::Restart | DaemonCommand::Start => (true, true),
         DaemonCommand::Clean { .. } => (false, true),
@@ -162,6 +163,9 @@ pub async fn daemon_client(command: &DaemonCommand, base: &CommandBase) -> Resul
             }
             println!("Done");
         }
+        DaemonCommand::Watch => {
+            WatchClient::start(&base.repo_root).await?;
+        }
     };
 
     Ok(())

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

@@ -24,7 +24,7 @@ pub(crate) mod scan;
 pub(crate) mod telemetry;
 pub(crate) mod unlink;
 
-#[derive(Debug)]
+#[derive(Debug, Clone)]
 pub struct CommandBase {
     pub repo_root: AbsoluteSystemPathBuf,
     pub ui: UI,

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

@@ -29,11 +29,10 @@ pub async fn run(base: CommandBase, telemetry: CommandEventBuilder) -> Result<i3
 
     let handler = SignalHandler::new(signal);
 
-    let api_auth = base.api_auth()?;
-    let api_client = base.api_client()?;
-    let run_builder = RunBuilder::new(base, api_auth)?;
+    let run_builder = RunBuilder::new(base)?;
+
     let run_fut = async {
-        let run = run_builder.build(&handler, telemetry, api_client).await?;
+        let run = run_builder.build(&handler, telemetry).await?;
         run.run().await
     };
 

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

@@ -12,7 +12,10 @@ use super::{
     proto::DiscoverPackagesResponse,
     Paths,
 };
-use crate::{daemon::proto, globwatcher::HashGlobSetupError};
+use crate::{
+    daemon::{proto, proto::PackageChangeEvent},
+    globwatcher::HashGlobSetupError,
+};
 
 #[derive(Debug, Clone)]
 pub struct DaemonClient<T> {
@@ -144,6 +147,18 @@ impl<T> DaemonClient<T> {
 
         Ok(response)
     }
+
+    pub async fn package_changes(
+        &mut self,
+    ) -> Result<tonic::codec::Streaming<PackageChangeEvent>, DaemonError> {
+        let response = self
+            .client
+            .package_changes(proto::PackageChangesRequest {})
+            .await?
+            .into_inner();
+
+        Ok(response)
+    }
 }
 
 impl DaemonClient<DaemonConnector> {

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

@@ -416,14 +416,16 @@ mod test {
         select,
         sync::{oneshot::Sender, Mutex},
     };
+    use tokio_stream::wrappers::ReceiverStream;
+    use tonic::{Request, Response, Status};
     use tower::ServiceBuilder;
     use tracing::info;
     use turbopath::AbsoluteSystemPathBuf;
 
     use super::*;
     use crate::daemon::{
         default_timeout_layer::DefaultTimeoutLayer,
-        proto::{self, turbod_client::TurbodClient},
+        proto::{self, turbod_client::TurbodClient, PackageChangesRequest},
     };
 
     #[cfg(not(target_os = "windows"))]
@@ -639,6 +641,14 @@ mod test {
         ) -> Result<tonic::Response<proto::DiscoverPackagesResponse>, tonic::Status> {
             unimplemented!()
         }
+
+        type PackageChangesStream = ReceiverStream<Result<proto::PackageChangeEvent, Status>>;
+        async fn package_changes(
+            &self,
+            _req: Request<PackageChangesRequest>,
+        ) -> Result<Response<Self::PackageChangesStream>, Status> {
+            unimplemented!()
+        }
     }
 
     #[tokio::test]

crates/turborepo-lib/src/daemon/proto/turbod.proto

@@ -25,6 +25,8 @@ service Turbod {
   //
   // Since 1.12.0
   rpc DiscoverPackagesBlocking (DiscoverPackagesRequest) returns (DiscoverPackagesResponse);
+
+  rpc PackageChanges (PackageChangesRequest) returns (stream PackageChangeEvent);
 }
 
 message HelloRequest {
@@ -92,6 +94,26 @@ message DiscoverPackagesRequest {
 
 }
 
+message PackageChangesRequest {}
+
+message PackageChangeEvent {
+  oneof event {
+    PackageChanged package_changed = 1;
+    RediscoverPackages rediscover_packages = 2;
+    PackageChangeError error = 3;
+  }
+}
+
+message PackageChangeError {
+  string message = 1;
+}
+
+message PackageChanged {
+  string package_name = 1;
+}
+
+message RediscoverPackages {}
+
 message DiscoverPackagesResponse {
   repeated PackageFiles package_files = 1;
   PackageManager package_manager = 2;