//! Turborepo build tool integration.
//!
//! Queries Turborepo for the task graph and source inputs using
//! `turbo run <task> --dry-run=json`. The JSON output contains the full
//! resolved task graph including commands, directories, dependencies,
//! and input file mappings.

use super::{AbortOnDrop, BuildGraphResolver, BuildToolError, ResolvedBuildInfo};
use std::path::Path;

/// Turborepo build graph resolver.
///
/// Shells out to `turbo run --dry-run=json` to get the resolved task graph.
/// Extracts watch paths from the `inputs` and `directory` fields.
/// Falls back to `npx turbo` if `turbo` is not directly on PATH.
pub(crate) struct TurboResolver {
    /// The turbo task name to query (e.g. "dev", "build").
    task: String,
    /// Optional package filter (e.g. "@myorg/api").
    filter: Option<String>,
}

impl TurboResolver {
    /// Create a new resolver for the given turbo task.
    ///
    /// All turbo operations rely on user cancellation (Ctrl+C → `kill_on_drop`)
    /// rather than client-side timeouts.
    pub(crate) fn new(task: &str, filter: Option<&str>) -> Self {
        Self {
            task: task.to_string(),
            filter: filter.map(String::from),
        }
    }

    /// Find the turbo command — either `turbo` on PATH or `npx turbo` as fallback.
    /// Returns the (program, prefix_args) to use for running turbo.
    async fn find_turbo_cmd(&self) -> Result<(String, Vec<String>), BuildToolError> {
        // Try `turbo` directly first.
        let direct = tokio::process::Command::new("turbo")
            .arg("--version")
            .stdout(std::process::Stdio::null())
            .stderr(std::process::Stdio::null())
            .status()
            .await;

        if let Ok(status) = direct
            && status.success()
        {
            return Ok(("turbo".to_string(), Vec::new()));
        }

        // Fall back to `npx turbo`.
        let npx = tokio::process::Command::new("npx")
            .args(["turbo", "--version"])
            .stdout(std::process::Stdio::null())
            .stderr(std::process::Stdio::null())
            .status()
            .await;

        match npx {
            Ok(status) if status.success() => Ok(("npx".to_string(), vec!["turbo".to_string()])),
            _ => Err(BuildToolError::NotInstalled {
                tool: "turbo".to_string(),
            }),
        }
    }

    /// Run `turbo run <build_task>` for the given package filters.
    ///
    /// Turbo handles parallelism internally across workspace packages.
    /// Build output is streamed line-by-line through the provided callback.
    pub(crate) async fn build_packages<F>(
        &self,
        build_task: &str,
        filters: &[String],
        working_dir: &Path,
        mut on_line: F,
        emitter: Option<&crate::output::LifecycleEmitter>,
    ) -> Result<super::BatchBuildResult, BuildToolError>
    where
        F: FnMut(&str) + Send + 'static,
    {
        if filters.is_empty() {
            return Ok(super::BatchBuildResult {
                succeeded: Vec::new(),
                failed: Vec::new(),
            });
        }

        let (program, prefix_args) = self.find_turbo_cmd().await?;

        let mut cmd = tokio::process::Command::new(&program);
        cmd.args(&prefix_args);
        cmd.args(["run", build_task]);

        for filter in filters {
            cmd.args(["--filter", filter]);
        }

        cmd.current_dir(working_dir)
            .stdout(std::process::Stdio::piped())
            .stderr(std::process::Stdio::piped())
            // Drop = SIGKILL on shutdown mid-startup.
            .kill_on_drop(true);

        if let Some(em) = emitter {
            let mut args: Vec<String> = prefix_args.to_vec();
            args.push("run".to_string());
            args.push(build_task.to_string());
            for filter in filters {
                args.push("--filter".to_string());
                args.push(filter.clone());
            }
            em.debug_spawn("turbo", &program, &args);
        }

        let mut child = cmd.spawn().map_err(|e| BuildToolError::Io {
            tool: "turbo".to_string(),
            source: e,
        })?;

        // Stream stdout. Wrap in AbortOnDrop so cancellation tears the
        // reader down — see comment in `bazel.rs::build_targets`.
        let stdout = child.stdout.take();
        let stream_handle = AbortOnDrop::new(tokio::spawn(async move {
            if let Some(stdout) = stdout {
                let mut reader = tokio::io::BufReader::new(stdout);
                let mut line_buf = Vec::new();
                loop {
                    line_buf.clear();
                    match tokio::io::AsyncBufReadExt::read_until(&mut reader, b'\n', &mut line_buf)
                        .await
                    {
                        Ok(0) => break,
                        Ok(_) => {
                            if line_buf.last() == Some(&b'\n') {
                                line_buf.pop();
                            }
                            if line_buf.last() == Some(&b'\r') {
                                line_buf.pop();
                            }
                            let text = String::from_utf8_lossy(&line_buf);
                            on_line(&text);
                        }
                        Err(_) => break,
                    }
                }
            }
        }));

        // Drain stderr. Same drop-on-cancel rationale.
        let stderr = child.stderr.take();
        let stderr_handle = AbortOnDrop::new(tokio::spawn(async move {
            let mut collected = String::new();
            if let Some(stderr) = stderr {
                let mut reader = tokio::io::BufReader::new(stderr);
                let _ = tokio::io::AsyncReadExt::read_to_string(&mut reader, &mut collected).await;
            }
            collected
        }));

        let status = child.wait().await.map_err(|e| BuildToolError::Io {
            tool: "turbo".to_string(),
            source: e,
        })?;

        if let Some(h) = stream_handle.into_inner() {
            let _ = h.await;
        }
        let stderr_output = match stderr_handle.into_inner() {
            Some(h) => h.await.unwrap_or_default(),
            None => String::new(),
        };

        if status.success() {
            Ok(super::BatchBuildResult {
                succeeded: filters.to_vec(),
                failed: Vec::new(),
            })
        } else {
            let code = status.code().unwrap_or(-1);
            let error_msg = if stderr_output.trim().is_empty() {
                format!("turbo run {build_task} failed (exit code {code})")
            } else {
                let truncated = if stderr_output.len() > 300 {
                    format!("{}...", &stderr_output[..300])
                } else {
                    stderr_output.trim().to_string()
                };
                format!("turbo run {build_task} failed: {truncated}")
            };
            // Turbo doesn't easily report per-package failures in a parseable way,
            // so conservatively mark all filtered packages as failed.
            Ok(super::BatchBuildResult {
                succeeded: Vec::new(),
                failed: filters
                    .iter()
                    .map(|f| (f.clone(), error_msg.clone()))
                    .collect(),
            })
        }
    }
}

/// A task from Turborepo's `--dry-run=json` output.
#[derive(serde::Deserialize)]
struct TurboTask {
    /// Workspace-relative directory of the package.
    directory: String,
    /// Input files mapped to their hashes. Keys are relative file paths.
    #[serde(default)]
    inputs: std::collections::HashMap<String, String>,
}

/// Top-level structure of `turbo run --dry-run=json` output.
#[derive(serde::Deserialize)]
struct TurboDryRun {
    tasks: Vec<TurboTask>,
}

/// Parse the dry-run JSON and extract watch paths.
///
/// For each task in the output, the `directory` field is the workspace package
/// root and `inputs` contains the individual files. We use directory-level
/// patterns for watch efficiency.
pub(crate) fn parse_dry_run(json: &str) -> Result<Vec<ParsedTurboTask>, BuildToolError> {
    let dry_run: TurboDryRun =
        serde_json::from_str(json).map_err(|e| BuildToolError::ParseError {
            tool: "turbo".to_string(),
            message: format!("failed to parse dry-run JSON: {e}"),
        })?;

    Ok(dry_run
        .tasks
        .into_iter()
        .map(|task| {
            // Collect unique subdirectories from input file paths.
            // This gives us directory-level watch patterns rather than
            // watching every individual file.
            let mut input_dirs: std::collections::HashSet<String> =
                std::collections::HashSet::new();
            for file_path in task.inputs.keys() {
                // Get the parent directory of each input file
                if let Some(parent) = std::path::Path::new(file_path).parent() {
                    let dir_str = parent.to_string_lossy();
                    if !dir_str.is_empty() && dir_str != "." {
                        input_dirs.insert(format!("{dir_str}/**"));
                    }
                }
            }

            // If no specific input dirs were found, watch the whole package directory
            let watch_paths = if input_dirs.is_empty() {
                vec![format!("{}/**", task.directory)]
            } else {
                // Prefix input dirs with the task directory since inputs are
                // relative to the workspace root, not the package
                let mut paths: Vec<String> = input_dirs.into_iter().collect();
                paths.sort();
                paths
            };

            ParsedTurboTask {
                directory: task.directory,
                watch_paths,
            }
        })
        .collect())
}

/// A parsed turbo task with resolved watch information.
#[derive(Debug, Clone)]
pub(crate) struct ParsedTurboTask {
    /// Workspace-relative directory.
    pub directory: String,
    /// Glob patterns to watch for source changes.
    pub watch_paths: Vec<String>,
}

impl TurboResolver {
    /// Run `turbo run <task> --dry-run=json` with the given filters (may be
    /// empty for "all packages") and collect the dry-run task graph.
    async fn run_dry_run(
        &self,
        filters: &[&str],
        working_dir: &Path,
    ) -> Result<Vec<ParsedTurboTask>, BuildToolError> {
        let (program, prefix_args) = self.find_turbo_cmd().await?;

        let mut cmd = tokio::process::Command::new(&program);
        cmd.args(&prefix_args);
        cmd.args(["run", &self.task, "--dry-run=json"]);
        for filter in filters {
            cmd.args(["--filter", filter]);
        }

        cmd.current_dir(working_dir)
            .stdout(std::process::Stdio::piped())
            .stderr(std::process::Stdio::piped())
            .kill_on_drop(true);

        let child = cmd.spawn().map_err(|e| BuildToolError::Io {
            tool: "turbo".to_string(),
            source: e,
        })?;

        let output = child
            .wait_with_output()
            .await
            .map_err(|e| BuildToolError::Io {
                tool: "turbo".to_string(),
                source: e,
            })?;

        if !output.status.success() {
            let stderr = String::from_utf8_lossy(&output.stderr);
            let truncated = if stderr.len() > 500 {
                format!("{}...", &stderr[..500])
            } else {
                stderr.to_string()
            };
            return Err(BuildToolError::QueryFailed {
                tool: "turbo".to_string(),
                message: truncated.trim().to_string(),
            });
        }

        let stdout = String::from_utf8(output.stdout).map_err(|e| BuildToolError::ParseError {
            tool: "turbo".to_string(),
            message: format!("non-UTF-8 output: {e}"),
        })?;

        parse_dry_run(&stdout)
    }

    /// Aggregate tier-2 source globs + tier-1 per-package `package.json`
    /// globs from a parsed turbo dry-run task graph.
    fn aggregate(tasks: &[ParsedTurboTask]) -> ResolvedBuildInfo {
        let mut all_watch_paths = Vec::new();
        let mut seen_paths: std::collections::HashSet<String> = std::collections::HashSet::new();
        let mut seen_dirs: std::collections::HashSet<String> = std::collections::HashSet::new();
        let mut graph_definition_globs: Vec<String> = Vec::new();

        for task in tasks {
            for path in &task.watch_paths {
                if seen_paths.insert(path.clone()) {
                    all_watch_paths.push(path.clone());
                }
            }
            if seen_dirs.insert(task.directory.clone()) {
                graph_definition_globs.push(format!("{}/package.json", task.directory));
            }
        }

        ResolvedBuildInfo {
            watch_paths: all_watch_paths,
            graph_definition_globs,
        }
    }

    /// Resolve the union of watch paths / build-graph globs for multiple
    /// filters in ONE `turbo run --dry-run` invocation. Filters are passed
    /// as repeated `--filter` flags, which turbo treats as a union.
    pub(crate) async fn resolve_union(
        &self,
        filters: &[String],
        working_dir: &Path,
    ) -> Result<ResolvedBuildInfo, BuildToolError> {
        if filters.is_empty() {
            return Ok(ResolvedBuildInfo {
                watch_paths: Vec::new(),
                graph_definition_globs: Vec::new(),
            });
        }
        let filter_refs: Vec<&str> = filters.iter().map(String::as_str).collect();
        let tasks = self.run_dry_run(&filter_refs, working_dir).await?;
        Ok(Self::aggregate(&tasks))
    }
}

impl BuildGraphResolver for TurboResolver {
    async fn resolve(
        &self,
        _target: &str,
        working_dir: &Path,
    ) -> Result<ResolvedBuildInfo, BuildToolError> {
        let mut filters: Vec<&str> = Vec::new();
        if let Some(ref filter) = self.filter {
            filters.push(filter.as_str());
        }
        let tasks = self.run_dry_run(&filters, working_dir).await?;
        Ok(Self::aggregate(&tasks))
    }
}

#[cfg(test)]
#[allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]
mod tests {
    use super::*;

    const SAMPLE_DRY_RUN: &str = r#"{
        "id": "abc123",
        "version": "1",
        "turboVersion": "2.0.0",
        "monorepo": true,
        "tasks": [
            {
                "taskId": "@myorg/utils#build",
                "task": "build",
                "package": "@myorg/utils",
                "directory": "packages/utils",
                "dependencies": [],
                "inputs": {
                    "packages/utils/src/index.ts": "abc123",
                    "packages/utils/src/helpers.ts": "def456",
                    "packages/utils/package.json": "ghi789"
                },
                "resolvedTaskDefinition": {
                    "persistent": false
                }
            },
            {
                "taskId": "@myorg/web#dev",
                "task": "dev",
                "package": "@myorg/web",
                "directory": "apps/web",
                "dependencies": ["@myorg/utils#build"],
                "inputs": {
                    "apps/web/src/App.tsx": "aaa111",
                    "apps/web/src/index.tsx": "bbb222",
                    "apps/web/pages/index.tsx": "ccc333",
                    "apps/web/package.json": "ddd444"
                },
                "resolvedTaskDefinition": {
                    "persistent": true
                }
            }
        ]
    }"#;

    #[test]
    fn test_parse_dry_run_basic() {
        let tasks = parse_dry_run(SAMPLE_DRY_RUN).unwrap();
        assert_eq!(tasks.len(), 2);

        let utils = &tasks[0];
        assert_eq!(utils.directory, "packages/utils");
        // Should have directory-level patterns from input files
        assert!(
            utils
                .watch_paths
                .iter()
                .any(|p| p.contains("packages/utils/src"))
        );

        let web = &tasks[1];
        assert_eq!(web.directory, "apps/web");
        assert!(web.watch_paths.iter().any(|p| p.contains("apps/web/src")));
    }

    #[test]
    fn test_parse_dry_run_empty_inputs_fallback() {
        let json = r#"{"tasks": [{"taskId": "a#dev", "directory": "apps/web", "dependencies": [], "inputs": {}}]}"#;
        let tasks = parse_dry_run(json).unwrap();
        assert_eq!(tasks[0].watch_paths, vec!["apps/web/**"]);
    }

    #[test]
    fn test_parse_dry_run_invalid_json() {
        let result = parse_dry_run("not json");
        assert!(result.is_err());
        let err = result.unwrap_err();
        assert!(err.to_string().contains("turbo"));
    }

    #[test]
    fn test_parse_dry_run_empty_tasks() {
        let json = r#"{"tasks": []}"#;
        let tasks = parse_dry_run(json).unwrap();
        assert!(tasks.is_empty());
    }
}
