feat(mfe): inject local proxy (#9356)

### Description

This PR adds all support necessary for starting a MFE local proxy:
- Adds new `CommandProvider` to allow for multiple strategies for
constructing a command depending on the type
 - Adds parsing logic for `micro-frontends.jsonc`
 - Parses all `micro-frontends.jsonc` found in the package graph*
- Adding a task definition for the generated `proxy` task if MFE configs
are found
- Explicitly include the `proxy` tasks in engine building step if a MFE
dev task is found in the product of the package `--filter` and the tasks
specified in `turbo run`
- Adds a `CommandProvider` that construct a command that invokes the MFE
local proxy

⚠️ This currently only support MFE configs that are located in shared
packages and doesn't respect a loose MFE configs.

Follow up work:
 - [ ] Handle configurable MFE config location

### Testing Instructions

In a repository with MFE, run a dev task for one of the configured MFE.
This should result in a `#proxy` task being injected for that MFE and
correctly configured to route traffic to the dev tasks or a remote host.

Author: chris-olszewski

Cargo.lock

@@ -57,6 +57,7 @@ turborepo-errors = { path = "crates/turborepo-errors" }
 turborepo-fs = { path = "crates/turborepo-fs" }
 turborepo-lib = { path = "crates/turborepo-lib", default-features = false }
 turborepo-lockfiles = { path = "crates/turborepo-lockfiles" }
+turborepo-micro-frontend = { path = "crates/turborepo-micro-frontend" }
 turborepo-repository = { path = "crates/turborepo-repository" }
 turborepo-ui = { path = "crates/turborepo-ui" }
 turborepo-unescape = { path = "crates/turborepo-unescape" }

Cargo.toml

@@ -133,6 +133,7 @@ turborepo-filewatch = { path = "../turborepo-filewatch" }
 turborepo-fs = { path = "../turborepo-fs" }
 turborepo-graph-utils = { path = "../turborepo-graph-utils" }
 turborepo-lockfiles = { workspace = true }
+turborepo-micro-frontend = { workspace = true }
 turborepo-repository = { path = "../turborepo-repository" }
 turborepo-scm = { workspace = true }
 turborepo-telemetry = { path = "../turborepo-telemetry" }

crates/turborepo-lib/Cargo.toml

@@ -404,7 +404,10 @@ impl Engine<Built> {
             .filter_map(|task| {
                 let pkg_name = PackageName::from(task.package());
                 let json = pkg_graph.package_json(&pkg_name)?;
-                json.command(task.task()).map(|_| task.to_string())
+                // TODO: delegate to command factory to filter down tasks to those that will
+                // have a runnable command.
+                (task.task() == "proxy" || json.command(task.task()).is_some())
+                    .then(|| task.to_string())
             })
             .collect()
     }

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

@@ -23,6 +23,7 @@ mod framework;
 mod gitignore;
 pub(crate) mod globwatcher;
 mod hash;
+mod micro_frontends;
 mod opts;
 mod package_changes_watcher;
 mod panic_handler;

crates/turborepo-lib/src/lib.rs

@@ -0,0 +1,52 @@
+use std::collections::{HashMap, HashSet};
+
+use turbopath::AbsoluteSystemPath;
+use turborepo_micro_frontend::{Config as MFEConfig, Error, DEFAULT_MICRO_FRONTENDS_CONFIG};
+use turborepo_repository::package_graph::PackageGraph;
+
+use crate::run::task_id::TaskId;
+
+#[derive(Debug, Clone)]
+pub struct MicroFrontendsConfigs {
+    configs: HashMap<String, HashSet<TaskId<'static>>>,
+}
+
+impl MicroFrontendsConfigs {
+    pub fn new(
+        repo_root: &AbsoluteSystemPath,
+        package_graph: &PackageGraph,
+    ) -> Result<Option<Self>, Error> {
+        let mut configs = HashMap::new();
+        for (package_name, package_info) in package_graph.packages() {
+            let config_path = repo_root
+                .resolve(package_info.package_path())
+                .join_component(DEFAULT_MICRO_FRONTENDS_CONFIG);
+            let Some(config) = MFEConfig::load(&config_path)? else {
+                continue;
+            };
+            let tasks = config
+                .applications
+                .iter()
+                .map(|(application, options)| {
+                    let dev_task = options.development.task.as_deref().unwrap_or("dev");
+                    TaskId::new(application, dev_task).into_owned()
+                })
+                .collect();
+            configs.insert(package_name.to_string(), tasks);
+        }
+
+        Ok((!configs.is_empty()).then_some(Self { configs }))
+    }
+
+    pub fn contains_package(&self, package_name: &str) -> bool {
+        self.configs.contains_key(package_name)
+    }
+
+    pub fn configs(&self) -> impl Iterator<Item = (&String, &HashSet<TaskId<'static>>)> {
+        self.configs.iter()
+    }
+
+    pub fn get(&self, package_name: &str) -> Option<&HashSet<TaskId<'static>>> {
+        self.configs.get(package_name)
+    }
+}

crates/turborepo-lib/src/micro_frontends.rs

@@ -101,6 +101,11 @@ impl Command {
     pub fn will_open_stdin(&self) -> bool {
         self.open_stdin
     }
+
+    /// Program for the command
+    pub fn program(&self) -> &OsStr {
+        &self.program
+    }
 }
 
 impl From<Command> for tokio::process::Command {

crates/turborepo-lib/src/process/command.rs

@@ -6,6 +6,7 @@ use std::{
 };
 
 use chrono::Local;
+use itertools::Itertools;
 use tracing::debug;
 use turbopath::{AbsoluteSystemPath, AbsoluteSystemPathBuf};
 use turborepo_analytics::{start_analytics, AnalyticsHandle, AnalyticsSender};
@@ -37,10 +38,12 @@ use {
     },
 };
 
+use super::task_id::TaskId;
 use crate::{
     cli::DryRunMode,
     commands::CommandBase,
     engine::{Engine, EngineBuilder},
+    micro_frontends::MicroFrontendsConfigs,
     opts::Opts,
     process::ProcessManager,
     run::{scope, task_access::TaskAccess, task_id::TaskName, Error, Run, RunCache},
@@ -370,6 +373,7 @@ impl RunBuilder {
         repo_telemetry.track_package_manager(pkg_dep_graph.package_manager().to_string());
         repo_telemetry.track_size(pkg_dep_graph.len());
         run_telemetry.track_run_type(self.opts.run_opts.dry_run.is_some());
+        let micro_frontend_configs = MicroFrontendsConfigs::new(&self.repo_root, &pkg_dep_graph)?;
 
         let scm = scm.await.expect("detecting scm panicked");
         let async_cache = AsyncCache::new(
@@ -401,6 +405,13 @@ impl RunBuilder {
                 self.repo_root.clone(),
                 pkg_dep_graph.packages(),
             )
+        } else if let Some(micro_frontends) = &micro_frontend_configs {
+            TurboJsonLoader::workspace_with_microfrontends(
+                self.repo_root.clone(),
+                self.root_turbo_json_path.clone(),
+                pkg_dep_graph.packages(),
+                micro_frontends.clone(),
+            )
         } else {
             TurboJsonLoader::workspace(
                 self.repo_root.clone(),
@@ -427,6 +438,7 @@ impl RunBuilder {
             &root_turbo_json,
             filtered_pkgs.keys(),
             turbo_json_loader.clone(),
+            micro_frontend_configs.as_ref(),
         )?;
 
         if self.opts.run_opts.parallel {
@@ -436,6 +448,7 @@ impl RunBuilder {
                 &root_turbo_json,
                 filtered_pkgs.keys(),
                 turbo_json_loader,
+                micro_frontend_configs.as_ref(),
             )?;
         }
 
@@ -476,6 +489,7 @@ impl RunBuilder {
             signal_handler: signal_handler.clone(),
             daemon,
             should_print_prelude,
+            micro_frontend_configs,
         })
     }
 
@@ -485,7 +499,52 @@ impl RunBuilder {
         root_turbo_json: &TurboJson,
         filtered_pkgs: impl Iterator<Item = &'a PackageName>,
         turbo_json_loader: TurboJsonLoader,
+        micro_frontends_configs: Option<&MicroFrontendsConfigs>,
     ) -> Result<Engine, Error> {
+        let mut tasks = self
+            .opts
+            .run_opts
+            .tasks
+            .iter()
+            .map(|task| {
+                // TODO: Pull span info from command
+                Spanned::new(TaskName::from(task.as_str()).into_owned())
+            })
+            .collect::<Vec<_>>();
+        let mut workspace_packages = filtered_pkgs.cloned().collect::<Vec<_>>();
+        if let Some(micro_frontends_configs) = micro_frontends_configs {
+            // TODO: this logic is very similar to what happens inside engine builder and
+            // could probably be exposed
+            let tasks_from_filter = workspace_packages
+                .iter()
+                .map(|package| package.as_str())
+                .cartesian_product(tasks.iter())
+                .map(|(package, task)| {
+                    task.task_id().map_or_else(
+                        || TaskId::new(package, task.task()).into_owned(),
+                        |task_id| task_id.into_owned(),
+                    )
+                })
+                .collect::<HashSet<_>>();
+            // we need to add the MFE config packages into the scope here to make sure the
+            // proxy gets run
+            // TODO(olszewski): We are relying on the fact that persistent tasks must be
+            // entry points to the task graph so we can get away with only
+            // checking the entrypoint tasks.
+            for (mfe_config_package, dev_tasks) in micro_frontends_configs.configs() {
+                for dev_task in dev_tasks {
+                    if tasks_from_filter.contains(dev_task) {
+                        workspace_packages.push(PackageName::from(mfe_config_package.as_str()));
+                        tasks.push(Spanned::new(
+                            TaskId::new(mfe_config_package, "proxy")
+                                .as_task_name()
+                                .into_owned(),
+                        ));
+                        break;
+                    }
+                }
+            }
+        }
         let mut builder = EngineBuilder::new(
             &self.repo_root,
             pkg_dep_graph,
@@ -494,11 +553,8 @@ impl RunBuilder {
         )
         .with_root_tasks(root_turbo_json.tasks.keys().cloned())
         .with_tasks_only(self.opts.run_opts.only)
-        .with_workspaces(filtered_pkgs.cloned().collect())
-        .with_tasks(self.opts.run_opts.tasks.iter().map(|task| {
-            // TODO: Pull span info from command
-            Spanned::new(TaskName::from(task.as_str()).into_owned())
-        }));
+        .with_workspaces(workspace_packages)
+        .with_tasks(tasks);
 
         if self.add_all_tasks {
             builder = builder.add_all_tasks();

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

@@ -60,4 +60,6 @@ pub enum Error {
     UI(#[from] turborepo_ui::Error),
     #[error(transparent)]
     Tui(#[from] tui::Error),
+    #[error("Error reading micro frontends configuration: {0}")]
+    MicroFrontends(#[from] turborepo_micro_frontend::Error),
 }

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

@@ -41,6 +41,7 @@ pub use crate::run::error::Error;
 use crate::{
     cli::EnvMode,
     engine::Engine,
+    micro_frontends::MicroFrontendsConfigs,
     opts::Opts,
     process::ProcessManager,
     run::{global_hash::get_global_hash_inputs, summary::RunTracker, task_access::TaskAccess},
@@ -73,6 +74,7 @@ pub struct Run {
     task_access: TaskAccess,
     daemon: Option<DaemonClient<DaemonConnector>>,
     should_print_prelude: bool,
+    micro_frontend_configs: Option<MicroFrontendsConfigs>,
 }
 
 type UIResult<T> = Result<Option<(T, JoinHandle<Result<(), turborepo_ui::Error>>)>, Error>;
@@ -460,6 +462,7 @@ impl Run {
             global_env,
             ui_sender,
             is_watch,
+            self.micro_frontend_configs.as_ref(),
         )
         .await;