Starting a run with most of the logic stubbed out

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

@@ -19,6 +19,7 @@ pub(crate) mod daemon;
 pub(crate) mod link;
 pub(crate) mod login;
 pub(crate) mod logout;
+pub(crate) mod run;
 pub(crate) mod unlink;
 
 pub struct CommandBase {

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

@@ -0,0 +1,22 @@
+use anyhow::Result;
+use tracing::error;
+use turbopath::AbsoluteSystemPathBuf;
+
+use crate::{
+    cli::Command::Run, commands::CommandBase, manager::Manager, opts::Opts,
+    package_json::PackageJson,
+};
+
+#[tokio::main]
+async fn run(base: &mut CommandBase) -> Result<()> {
+    // equivalent of optsFromArgs
+    let opts: Opts = (&base.args).try_into()?;
+
+    // equivalent of configureRun
+    let run = Run::new(base, opts);
+
+    run.run().await.map_err(|err| {
+        error!("run failed: {}", err);
+        err
+    })
+}

crates/turborepo-lib/src/lib.rs

@@ -7,8 +7,11 @@ mod config;
 mod daemon;
 mod execution_state;
 mod formatter;
+mod manager;
+mod opts;
 mod package_json;
 mod package_manager;
+mod run;
 mod shim;
 mod ui;
 

crates/turborepo-lib/src/manager.rs

@@ -0,0 +1,9 @@
+pub struct Manager {
+    // TODO
+}
+
+impl Manager {
+    pub fn new() -> Self {
+        Self {}
+    }
+}

crates/turborepo-lib/src/opts.rs

@@ -0,0 +1,115 @@
+use anyhow::{anyhow, Result};
+
+use crate::{
+    cli::{Command, DryRunMode, EnvMode, RunArgs},
+    daemon::{DaemonClient, DaemonConnector},
+    Args,
+};
+
+pub struct Opts<'a> {
+    pub run_opts: RunOpts<'a>,
+    pub runcache_opts: RunCacheOpts<'a>,
+}
+
+impl<'a> TryFrom<&'a Args> for Opts<'a> {
+    type Error = anyhow::Error;
+
+    fn try_from(args: &'a Args) -> std::result::Result<Self, Self::Error> {
+        let Command::Run(run_args) = &args.command else {
+          anyhow!("Expected run command")
+        };
+        let run_opts = RunOpts::try_from(run_args)?;
+
+        Ok(Self {
+            run_opts,
+            runcache_opts: RunCacheOpts::default(),
+        })
+    }
+}
+
+#[derive(Debug, Default)]
+pub struct RunCacheOpts {
+    pub(crate) output_watcher: Option<DaemonClient<DaemonConnector>>,
+}
+
+pub struct RunOpts<'a> {
+    tasks: &'a [String],
+    concurrency: u32,
+    parallel: bool,
+    env_mode: EnvMode,
+    profile: &'a str,
+    continue_on_error: bool,
+    passthrough_args: &'a [String],
+    only: bool,
+    dry_run: bool,
+    pub(crate) dry_run_json: bool,
+    pub graph_dot: bool,
+    graph_file: Option<&'a str>,
+    pub(crate) no_daemon: bool,
+    pub(crate) single_package: bool,
+    log_prefix: &'a str,
+    summarize: bool,
+    experimental_space_id: &'a str,
+}
+
+const DEFAULT_CONCURRENCY: u32 = 10;
+
+impl<'a> TryFrom<&'a RunArgs> for RunOpts<'a> {
+    type Error = anyhow::Error;
+
+    fn try_from(args: &'a RunArgs) -> Result<Self> {
+        let concurrency = args
+            .concurrency
+            .map(parse_concurrency)
+            .transpose()?
+            .unwrap_or(DEFAULT_CONCURRENCY);
+        let (graph_dot, graph_file) = match args.graph {
+            Some(file) if file.is_empty() => (true, None),
+            Some(file) => (false, Some(file.as_str())),
+            None => (false, None),
+        };
+
+        Ok(Self {
+            tasks: args.tasks.as_slice(),
+            log_prefix: args.log_prefix.as_str(),
+            summarize: *args.summarize,
+            experimental_space_id: args.experimental_space_id.as_str(),
+            env_mode: args.env_mode,
+            concurrency,
+            parallel: args.parallel,
+            profile: args.profile.as_str(),
+            continue_on_error: *args.continue_on_error,
+            passthrough_args: args.pass_through_args.as_ref(),
+            only: *args.only,
+            no_daemon: *args.no_daemon,
+            single_package: *args.single_package,
+            graph_dot,
+            graph_file,
+            dry_run_json: matches!(args.dry_run, Some(DryRunMode::Json)),
+            dry_run: args.dry_run.is_some(),
+        })
+    }
+}
+
+fn parse_concurrency(concurrency_raw: &str) -> Result<u32> {
+    if let Some(percent) = concurrency_raw.strip_suffix('%') {
+        let percent = percent.parse::<f64>()?;
+        return if percent > 0.0 && percent.is_finite() {
+            Ok((num_cpus::get() as f64 * percent / 100.0).max(1.0) as u32)
+        } else {
+            Err(anyhow!(
+                "invalid percentage value for --concurrency CLI flag. This should be a percentage \
+                 of CPU cores, between 1% and 100% : {}",
+                percent
+            ))
+        };
+    }
+    match concurrency_raw.parse::<u32>() {
+        Ok(concurrency) if concurrency > 1 => Ok(concurrency),
+        Ok(_) | Err(_) => Err(anyhow!(
+            "invalid value for --concurrency CLI flag. This should be a positive integer greater \
+             than or equal to 1: {}",
+            concurrency
+        )),
+    }
+}

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

@@ -0,0 +1,31 @@
+use anyhow::Result;
+
+use crate::package_json::PackageJson;
+
+struct AcyclicGraph {}
+struct Catalog {}
+
+pub struct Context {
+    workspace_graph: AcyclicGraph,
+    workspace_infos: Catalog,
+}
+
+impl Context {
+    pub fn build_single_package_graph(_root_package_json: PackageJson) -> Result<Context> {
+        // TODO
+        Ok(Context {})
+    }
+
+    pub fn build_multi_package_graph(
+        _repo_root: &std::path::PathBuf,
+        _root_package_json: &PackageJson,
+    ) -> Result<Context> {
+        // TODO
+        Ok(Context {})
+    }
+
+    pub fn validate(&self) -> Result<()> {
+        // TODO
+        Ok(())
+    }
+}

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

@@ -0,0 +1,65 @@
+mod context;
+
+use anyhow::Context;
+use tracing::{debug, info};
+use turbopath::AbsoluteSystemPathBuf;
+
+use crate::{
+    commands::CommandBase, daemon, daemon::DaemonConnector, manager::Manager, opts::Opts,
+    package_json::PackageJson, run::context::Context,
+};
+
+struct Run<'a> {
+    base: &'a mut CommandBase,
+    opts: Opts<'a>,
+    processes: Manager,
+}
+
+impl<'a> Run<'a> {
+    fn new(base: &'a mut CommandBase, opts: Opts<'a>) -> Self {
+        let processes = Manager::new();
+        Self {
+            base,
+            opts,
+            processes,
+        }
+    }
+
+    async fn run(&mut self) {
+        let start_at = std::time::Instant::now();
+        let package_json_path = base.repo_root.join("package.json");
+        let root_package_json = PackageJson::load(&AbsoluteSystemPathBuf::new(package_json_path)?)?;
+
+        let is_structured_output = self.opts.run_opts.graph_dot || self.opts.run_opts.dry_run_json;
+
+        let pkg_dep_graph = if self.opts.run_opts.single_package {
+            Context::build_single_package_graph(root_package_json)?
+        } else {
+            Context::build_multi_package_graph(&self.base.repo_root, &root_package_json)?
+        };
+        // There's some warning handling code in Go that I'm ignoring
+
+        if base.ui.is_ci() && !self.opts.run_opts.no_daemon {
+            info!("skipping turbod since we appear to be in a non-interactive context");
+        } else if !r.opts.run_opts.no_daemon {
+            let connector = DaemonConnector {
+                can_start_server: true,
+                can_kill_server: true,
+                pid_file: base.daemon_file_root().join_relative(
+                    turbopath::RelativeSystemPathBuf::new("turbod.pid").expect("relative system"),
+                ),
+                sock_file: base.daemon_file_root().join_relative(
+                    turbopath::RelativeSystemPathBuf::new("turbod.sock").expect("relative system"),
+                ),
+            };
+
+            let mut client = connector.connect().await?;
+            debug!("running in daemon mode");
+            self.opts.runcache_opts.output_watcher = Some(client);
+        }
+
+        pkg_dep_graph
+            .validate()
+            .context("Invalid package dependency graph")?;
+    }
+}