wsb 0.164.718

use crate::prelude::*;
use axum::{
    extract::{Query, State},
    http::StatusCode,
    response::sse::{Event, KeepAlive, Sse},
    Json,
};
use futures::stream::Stream;
use std::convert::Infallible;

use super::sessions::ErrorResponse;
use crate::fleet::config::GitModules;
use crate::fleet::git::GitOps;
use crate::git_backend::{self, FileStatusKind};
use crate::web::github;
use crate::web::ops::OpKind;
use crate::web::state::WebState;

// ---------------------------------------------------------------------------
// Shared types
// ---------------------------------------------------------------------------

#[derive(Serialize, Clone, Debug)]
pub struct GitFileStatus {
    pub path: String,
    pub status: FileStatusKind,
}

#[derive(Serialize, Clone, Debug)]
pub struct RepoDetailedStatus {
    pub name: String,
    pub path: String,
    pub branch: Option<String>,
    pub is_clean: bool,
    pub is_detached: bool,
    pub ahead: usize,
    pub behind: usize,
    pub files: Vec<GitFileStatus>,
    pub exists: bool,
    pub is_superproject: bool,
}

/// Legacy compat-free: the old RepoStatus is replaced by RepoDetailedStatus.
/// The fleet_status endpoint now returns full file-level status.
pub(crate) type RepoStatus = RepoDetailedStatus;

#[derive(Debug, Clone)]
pub(crate) struct GitHubRepoRef {
    pub(crate) owner: String,
    pub(crate) repo: String,
}

#[derive(Debug, Clone)]
pub(crate) struct IssueRepo {
    pub(crate) github: GitHubRepoRef,
}

// ---------------------------------------------------------------------------
// Request / Response types for mutation endpoints
// ---------------------------------------------------------------------------

#[derive(Deserialize)]
pub struct RepoPathRequest {
    pub repo_path: Option<String>,
}

#[derive(Deserialize)]
pub struct CommitRequest {
    pub repo_path: Option<String>,
    pub message: String,
}

#[derive(Deserialize)]
pub struct CheckoutRequest {
    pub repo_path: Option<String>,
    pub branch: String,
    pub create: Option<bool>,
}

#[derive(Deserialize)]
pub struct CreateBranchRequest {
    pub repo_path: Option<String>,
    pub branch: String,
    pub from_ref: Option<String>,
}

#[derive(Deserialize)]
pub struct FleetCommitRequest {
    pub message: String,
}

#[derive(Deserialize)]
pub struct FleetCheckoutRequest {
    pub branch: String,
    pub create: Option<bool>,
}

#[derive(Deserialize)]
pub struct FleetMergeBackRequest {
    pub message: Option<String>,
}

#[derive(Deserialize)]
pub struct BranchesQuery {
    pub repo_path: Option<String>,
}

#[derive(Deserialize)]
pub struct DiffQuery {
    /// Workspace-relative path to the file or directory (e.g. "src/main.rs"
    /// or "machfab/src/main.rs" for a submodule file).
    pub path: String,
    /// Optional repo root (workspace-relative). Defaults to superproject.
    pub repo_path: Option<String>,
}

#[derive(Deserialize)]
pub struct FocusHintRequest {
    pub repo_path: Option<String>,
}

#[derive(Serialize)]
pub struct GitOpResult {
    pub success: bool,
    pub message: String,
}

#[derive(Serialize)]
pub struct BranchListResponse {
    pub branches: Vec<String>,
    pub current: Option<String>,
}

// ---------------------------------------------------------------------------
// Helpers
// ---------------------------------------------------------------------------

pub(crate) fn workspace_root() -> Result<PathBuf, (StatusCode, Json<ErrorResponse>)> {
    let cwd = std::env::current_dir().map_err(|e| {
        (
            StatusCode::INTERNAL_SERVER_ERROR,
            Json(ErrorResponse {
                error: format!("cannot determine current directory: {}", e),
            }),
        )
    })?;

    GitOps::discover_repo_root(&cwd).map_err(|e| {
        (
            StatusCode::INTERNAL_SERVER_ERROR,
            Json(ErrorResponse {
                error: format!(
                    "cannot resolve workspace root from '{}': {}",
                    cwd.display(),
                    e
                ),
            }),
        )
    })
}

/// Resolve an optional relative repo_path to an absolute path under the workspace root.
/// If repo_path is None or ".", returns the workspace root itself.
/// Sync version of `resolve_repo_path` for use inside `spawn_blocking` closures.
fn resolve_repo_path_sync(
    root: &Path,
    repo_path: Option<&str>,
) -> Result<PathBuf, (StatusCode, Json<ErrorResponse>)> {
    resolve_repo_path(root, repo_path)
}

fn resolve_repo_path(root: &Path, repo_path: Option<&str>) -> Result<PathBuf, (StatusCode, Json<ErrorResponse>)> {
    let rel = repo_path.unwrap_or(".");
    if rel == "." || rel.is_empty() {
        return Ok(root.to_path_buf());
    }
    let resolved = root.join(rel);
    // Security: ensure resolved path stays under workspace root
    let canonical_root = root.canonicalize().map_err(|e| {
        (
            StatusCode::INTERNAL_SERVER_ERROR,
            Json(ErrorResponse {
                error: format!("cannot canonicalize workspace root: {}", e),
            }),
        )
    })?;
    let canonical_resolved = resolved.canonicalize().map_err(|e| {
        (
            StatusCode::BAD_REQUEST,
            Json(ErrorResponse {
                error: format!("cannot resolve repo path '{}': {}", rel, e),
            }),
        )
    })?;
    if !canonical_resolved.starts_with(&canonical_root) {
        return Err((
            StatusCode::BAD_REQUEST,
            Json(ErrorResponse {
                error: format!("repo path '{}' escapes workspace root", rel),
            }),
        ));
    }
    Ok(canonical_resolved)
}

fn internal_error(error: crate::fleet::error::SuperGitError) -> (StatusCode, Json<ErrorResponse>) {
    tracing::error!(error = %error, error_debug = ?error, "fleet handler error");
    (
        StatusCode::INTERNAL_SERVER_ERROR,
        Json(ErrorResponse {
            error: format!("{:#}", error),
        }),
    )
}

fn anyhow_error(error: anyhow::Error) -> (StatusCode, Json<ErrorResponse>) {
    tracing::error!(error = %error, "fleet handler error");
    (
        StatusCode::INTERNAL_SERVER_ERROR,
        Json(ErrorResponse {
            error: format!("{:#}", error),
        }),
    )
}

fn ok_result(msg: impl Into<String>) -> Json<GitOpResult> {
    Json(GitOpResult {
        success: true,
        message: msg.into(),
    })
}

/// Extract a plain error string from the handler error tuple, for use in op failure messages.
fn handler_err_msg(e: &(StatusCode, Json<ErrorResponse>)) -> String {
    e.1.error.clone()
}

// ---------------------------------------------------------------------------
// Status collection
// ---------------------------------------------------------------------------

fn collect_detailed_repo_status(
    repo_root: &Path,
    repo_path: &Path,
    name: &str,
    relative_path: &str,
    is_superproject: bool,
) -> Result<RepoDetailedStatus, crate::fleet::error::SuperGitError> {
    if !repo_path.exists() {
        return Ok(RepoDetailedStatus {
            name: name.to_string(),
            path: relative_path.to_string(),
            branch: None,
            is_clean: true,
            is_detached: false,
            ahead: 0,
            behind: 0,
            files: Vec::new(),
            exists: false,
            is_superproject,
        });
    }

    let repo = GitOps::open_repo(repo_path)?;
    let is_detached = GitOps::is_detached_head(&repo)?;
    let branch = if is_detached {
        None
    } else {
        Some(GitOps::current_branch(&repo)?)
    };

    let entries = git_backend::status_entries(&repo)?;
    let is_clean = !git_backend::has_uncommitted_changes(&entries);
    let files: Vec<GitFileStatus> = entries
        .into_iter()
        .map(|e| GitFileStatus {
            path: e.path,
            status: e.status,
        })
        .collect();

    let (ahead, behind) = if is_detached {
        (0, 0)
    } else {
        GitOps::get_ahead_behind(&repo).unwrap_or((0, 0))
    };

    Ok(RepoDetailedStatus {
        name: name.to_string(),
        path: relative_path.to_string(),
        branch,
        is_clean,
        is_detached,
        ahead,
        behind,
        files,
        exists: true,
        is_superproject,
    })
}

pub(crate) fn collect_repo_statuses(
    repo_root: &Path,
) -> crate::fleet::error::Result<Vec<RepoDetailedStatus>> {
    let super_name = repo_root
        .file_name()
        .map(|name| name.to_string_lossy().into_owned())
        .unwrap_or_else(|| repo_root.display().to_string());

    let mut repos = vec![collect_detailed_repo_status(
        repo_root,
        repo_root,
        &super_name,
        ".",
        true,
    )?];

    let submodules = match GitModules::parse(repo_root) {
        Ok(gitmodules) => gitmodules.submodules,
        Err(crate::fleet::error::SuperGitError::NoSubmodules) => Vec::new(),
        Err(error) => return Err(error),
    };

    for submodule in &submodules {
        let sub_path = repo_root.join(&submodule.path);
        let rel = submodule.path.display().to_string();
        repos.push(collect_detailed_repo_status(
            repo_root,
            &sub_path,
            &submodule.name,
            &rel,
            false,
        )?);
    }

    Ok(repos)
}

pub(crate) fn collect_issue_repos(
    repo_root: &Path,
) -> crate::fleet::error::Result<Vec<IssueRepo>> {
    let repo = GitOps::open_repo(repo_root)?;
    let origin_url = GitOps::remote_url(&repo, "origin")?;
    let mut repos = Vec::new();
    let mut seen = std::collections::BTreeSet::new();

    if let Some((owner, repo_name)) = github::parse_github_url(&origin_url) {
        let key = format!("{owner}/{repo_name}");
        if seen.insert(key) {
            repos.push(IssueRepo {
                github: GitHubRepoRef {
                    owner,
                    repo: repo_name,
                },
            });
        }
    }

    let submodules = match GitModules::parse(repo_root) {
        Ok(gitmodules) => gitmodules.submodules,
        Err(crate::fleet::error::SuperGitError::NoSubmodules) => Vec::new(),
        Err(error) => return Err(error),
    };

    for submodule in submodules {
        let Some((owner, repo_name)) = github::parse_github_url(&submodule.url) else {
            continue;
        };
        let key = format!("{owner}/{repo_name}");
        if seen.insert(key) {
            repos.push(IssueRepo {
                github: GitHubRepoRef {
                    owner,
                    repo: repo_name,
                },
            });
        }
    }

    Ok(repos)
}

// ---------------------------------------------------------------------------
// GET endpoints
// ---------------------------------------------------------------------------

/// GET /api/v1/fleet/status — full detailed status for all repos.
///
/// Served from the in-memory snapshot maintained by the git watcher — instant,
/// never blocks, always up to date (watcher pushes on every change).
pub async fn fleet_status(
    State(state): State<WebState>,
) -> Result<Json<serde_json::Value>, (StatusCode, Json<ErrorResponse>)> {
    let snapshot = state.git_status_snapshot.read().await;
    let repos: Vec<&RepoDetailedStatus> = snapshot.values().collect();
    Ok(Json(serde_json::json!({ "repos": repos })))
}

/// GET /api/v1/fleet/repos — list GitHub repos for issues
pub async fn fleet_repos(
    State(_state): State<WebState>,
) -> Result<Json<serde_json::Value>, (StatusCode, Json<ErrorResponse>)> {
    let repo_root = workspace_root()?;
    let repos = tokio::task::spawn_blocking(move || {
        collect_issue_repos(&repo_root)
            .map(|rs| {
                rs.into_iter()
                    .map(|repo| format!("{}/{}", repo.github.owner, repo.github.repo))
                    .collect::<Vec<_>>()
            })
    })
    .await
    .map_err(|e| anyhow_error(anyhow::anyhow!("spawn_blocking panicked: {:?}", e)))?
    .map_err(internal_error)?;
    Ok(Json(serde_json::json!({ "repos": repos })))
}

/// GET /api/v1/fleet/detailed-status — enriched status with per-file FileStatusKind
pub async fn fleet_detailed_status(
    State(_state): State<WebState>,
) -> Result<Json<serde_json::Value>, (StatusCode, Json<ErrorResponse>)> {
    let repo_root = workspace_root()?;
    let repos = tokio::task::spawn_blocking(move || collect_repo_statuses(&repo_root))
        .await
        .map_err(|e| anyhow_error(anyhow::anyhow!("spawn_blocking panicked: {:?}", e)))?
        .map_err(internal_error)?;
    Ok(Json(serde_json::json!({ "repos": repos })))
}

/// GET /api/v1/fleet/branches — list branches for a repo
pub async fn fleet_branches(
    State(_state): State<WebState>,
    Query(query): Query<BranchesQuery>,
) -> Result<Json<BranchListResponse>, (StatusCode, Json<ErrorResponse>)> {
    let root = workspace_root()?;
    let repo_path = resolve_repo_path(&root, query.repo_path.as_deref())?;
    let result = tokio::task::spawn_blocking(move || {
        let repo = git_backend::open_repo(&repo_path).map_err(anyhow_error)?;
        let branches = git_backend::local_branches(&repo).map_err(anyhow_error)?;
        let is_detached = git_backend::is_detached_head(&repo).map_err(anyhow_error)?;
        let current = if is_detached {
            None
        } else {
            Some(git_backend::current_branch(&repo).map_err(anyhow_error)?)
        };
        Ok::<BranchListResponse, (StatusCode, Json<ErrorResponse>)>(BranchListResponse {
            branches,
            current,
        })
    })
    .await
    .map_err(|e| anyhow_error(anyhow::anyhow!("spawn_blocking panicked: {:?}", e)))??;
    Ok(Json(result))
}

/// GET /api/v1/fleet/diff — unified diff for a path vs HEAD
///
/// `path` is the repo-relative path of the file or directory to diff.
/// `repo_path` is the workspace-relative repo root (defaults to superproject).
/// Returns plain text unified diff, or empty string if path has no changes.
pub async fn fleet_diff(
    State(_state): State<WebState>,
    Query(query): Query<DiffQuery>,
) -> Result<axum::response::Response, (StatusCode, Json<ErrorResponse>)> {
    let root = workspace_root()?;
    let repo_path = resolve_repo_path(&root, query.repo_path.as_deref())?;
    let path = query.path.clone();
    let diff = tokio::task::spawn_blocking(move || {
        let repo = git_backend::open_repo(&repo_path).map_err(anyhow_error)?;
        git_backend::diff_file(&repo, &path).map_err(anyhow_error)
    })
    .await
    .map_err(|e| anyhow_error(anyhow::anyhow!("spawn_blocking panicked: {:?}", e)))??;
    Ok(axum::response::Response::builder()
        .status(StatusCode::OK)
        .header("content-type", "text/plain; charset=utf-8")
        .body(axum::body::boxed(axum::body::Full::from(diff)))
        .unwrap())
}

// ---------------------------------------------------------------------------
// SSE — real-time git status stream with focus tracking
// ---------------------------------------------------------------------------

/// GET /api/v1/fleet/watch — SSE stream that pushes git status updates.
///
/// The background watcher thread continuously polls git state and broadcasts
/// changes.  Clients receive the full fleet status on each change.
pub async fn fleet_watch(
    State(state): State<WebState>,
) -> Sse<impl Stream<Item = Result<Event, Infallible>>> {
    let mut rx = state.git_status_tx.subscribe();

    let stream = async_stream::stream! {
        // Send the current snapshot as individual per-repo events so new
        // subscribers see the full picture immediately without waiting for
        // the next watcher cycle.
        {
            let snapshot = state.git_status_snapshot.read().await;
            for repo in snapshot.values() {
                if let Ok(json) = serde_json::to_string(repo) {
                    yield Ok(Event::default().event("repo").data(json));
                }
            }
        }

        loop {
            match rx.recv().await {
                Ok(repo) => {
                    if let Ok(json) = serde_json::to_string(&repo) {
                        yield Ok(Event::default().event("repo").data(json));
                    }
                }
                Err(tokio::sync::broadcast::error::RecvError::Lagged(n)) => {
                    // We fell behind; re-send the full snapshot so the client
                    // is consistent, then continue from the live stream.
                    tracing::warn!("git SSE client lagged by {} events, re-sending snapshot", n);
                    let snapshot = state.git_status_snapshot.read().await;
                    for repo in snapshot.values() {
                        if let Ok(json) = serde_json::to_string(repo) {
                            yield Ok(Event::default().event("repo").data(json));
                        }
                    }
                }
                Err(tokio::sync::broadcast::error::RecvError::Closed) => break,
            }
        }
    };

    Sse::new(stream).keep_alive(KeepAlive::default())
}

/// POST /api/v1/fleet/focus — hint the watcher about user's current focus
pub async fn fleet_focus(
    State(state): State<WebState>,
    Json(body): Json<FocusHintRequest>,
) -> Result<Json<GitOpResult>, (StatusCode, Json<ErrorResponse>)> {
    let hint = body.repo_path.unwrap_or_else(|| ".".to_string());
    // Send focus hint — non-blocking, ignore if channel is full
    let _ = state.git_focus_tx.send(hint.clone());
    Ok(ok_result(format!("focus set to '{}'", hint)))
}

// ---------------------------------------------------------------------------
// Single-repo mutation endpoints
// ---------------------------------------------------------------------------

/// POST /api/v1/fleet/stage — stage all changes in a repo
pub async fn fleet_stage(
    State(state): State<WebState>,
    Json(body): Json<RepoPathRequest>,
) -> Result<Json<GitOpResult>, (StatusCode, Json<ErrorResponse>)> {
    let root = workspace_root()?;
    let repo_path = resolve_repo_path(&root, body.repo_path.as_deref())?;
    let rel = body.repo_path.clone().unwrap_or_else(|| ".".to_string());
    let label = format!("Stage: {}", rel);
    let op = state.ops.start_op(OpKind::GitMutation, label).await;
    let result = tokio::task::spawn_blocking(move || {
        let repo = git_backend::open_repo(&repo_path).map_err(anyhow_error)?;
        git_backend::stage_all(&repo).map_err(anyhow_error)?;
        Ok::<(), (StatusCode, Json<ErrorResponse>)>(())
    })
    .await
    .map_err(|e| anyhow_error(anyhow::anyhow!("spawn_blocking panicked: {:?}", e)))?;
    match result {
        Ok(()) => {
            op.complete().await;
            notify_git_change_repo(&state, &rel).await;
            Ok(ok_result("staged all changes"))
        }
        Err(e) => {
            op.fail(handler_err_msg(&e)).await;
            Err(e)
        }
    }
}

/// POST /api/v1/fleet/commit — commit staged changes in a repo
pub async fn fleet_commit(
    State(state): State<WebState>,
    Json(body): Json<CommitRequest>,
) -> Result<Json<GitOpResult>, (StatusCode, Json<ErrorResponse>)> {
    let root = workspace_root()?;
    let repo_path = resolve_repo_path(&root, body.repo_path.as_deref())?;
    let rel = body.repo_path.clone().unwrap_or_else(|| ".".to_string());
    let message = body.message.clone();
    let label = format!("Commit: {}", rel);
    let op = state.ops.start_op(OpKind::GitMutation, label).await;
    let sha = tokio::task::spawn_blocking(move || {
        let repo = git_backend::open_repo(&repo_path).map_err(anyhow_error)?;
        git_backend::commit_all(&repo, &message).map_err(anyhow_error)
    })
    .await
    .map_err(|e| anyhow_error(anyhow::anyhow!("spawn_blocking panicked: {:?}", e)));
    match sha {
        Ok(Ok(sha)) => {
            op.complete().await;
            notify_git_change_repo(&state, &rel).await;
            let msg = match sha {
                Some(sha) => format!("committed: {}", sha),
                None => "nothing to commit".to_string(),
            };
            Ok(ok_result(msg))
        }
        Ok(Err(e)) => {
            op.fail(handler_err_msg(&e)).await;
            Err(e)
        }
        Err(e) => {
            let err = anyhow_error(anyhow::anyhow!("spawn_blocking panicked: {:?}", e));
            op.fail(handler_err_msg(&err)).await;
            Err(err)
        }
    }
}

/// POST /api/v1/fleet/push — push current branch
pub async fn fleet_push(
    State(state): State<WebState>,
    Json(body): Json<RepoPathRequest>,
) -> Result<Json<GitOpResult>, (StatusCode, Json<ErrorResponse>)> {
    let root = workspace_root()?;
    let repo_path = resolve_repo_path(&root, body.repo_path.as_deref())?;
    let rel = body.repo_path.clone().unwrap_or_else(|| ".".to_string());
    let label = format!("Push: {}", rel);
    let op = state.ops.start_op(OpKind::GitMutation, label).await;
    let result = tokio::task::spawn_blocking(move || {
        let repo = git_backend::open_repo(&repo_path).map_err(anyhow_error)?;
        git_backend::push_current_branch(&repo, Some("origin"), true, false)
            .map_err(anyhow_error)
    })
    .await
    .map_err(|e| anyhow_error(anyhow::anyhow!("spawn_blocking panicked: {:?}", e)));
    match result {
        Ok(Ok(outcome)) => {
            op.complete().await;
            notify_git_change_repo(&state, &rel).await;
            let msg = match outcome {
                git_backend::PushOutcome::Updated => "pushed successfully",
                git_backend::PushOutcome::UpToDate => "already up to date",
            };
            Ok(ok_result(msg))
        }
        Ok(Err(e)) => { op.fail(handler_err_msg(&e)).await; Err(e) }
        Err(e) => { let err = anyhow_error(anyhow::anyhow!("spawn_blocking panicked: {:?}", e)); op.fail(handler_err_msg(&err)).await; Err(err) }
    }
}

/// POST /api/v1/fleet/pull — pull current branch
pub async fn fleet_pull(
    State(state): State<WebState>,
    Json(body): Json<RepoPathRequest>,
) -> Result<Json<GitOpResult>, (StatusCode, Json<ErrorResponse>)> {
    let root = workspace_root()?;
    let repo_path = resolve_repo_path(&root, body.repo_path.as_deref())?;
    let rel = body.repo_path.clone().unwrap_or_else(|| ".".to_string());
    let label = format!("Pull: {}", rel);
    let op = state.ops.start_op(OpKind::GitMutation, label).await;
    let result = tokio::task::spawn_blocking(move || {
        let repo = git_backend::open_repo(&repo_path).map_err(anyhow_error)?;
        git_backend::pull(&repo, false).map_err(anyhow_error)?;
        Ok::<(), (StatusCode, Json<ErrorResponse>)>(())
    })
    .await
    .map_err(|e| anyhow_error(anyhow::anyhow!("spawn_blocking panicked: {:?}", e)));
    match result {
        Ok(Ok(())) => { op.complete().await; notify_git_change_repo(&state, &rel).await; Ok(ok_result("pulled successfully")) }
        Ok(Err(e)) => { op.fail(handler_err_msg(&e)).await; Err(e) }
        Err(e) => { let err = anyhow_error(anyhow::anyhow!("spawn_blocking panicked: {:?}", e)); op.fail(handler_err_msg(&err)).await; Err(err) }
    }
}

/// POST /api/v1/fleet/checkout — checkout a branch
pub async fn fleet_checkout(
    State(state): State<WebState>,
    Json(body): Json<CheckoutRequest>,
) -> Result<Json<GitOpResult>, (StatusCode, Json<ErrorResponse>)> {
    let root = workspace_root()?;
    let repo_path = resolve_repo_path(&root, body.repo_path.as_deref())?;
    let rel = body.repo_path.clone().unwrap_or_else(|| ".".to_string());
    let branch = body.branch.clone();
    let create = body.create.unwrap_or(false);
    let label = format!("Checkout {}: {}", body.branch, rel);
    let op = state.ops.start_op(OpKind::GitMutation, label).await;
    let branch_name = body.branch.clone();
    let result = tokio::task::spawn_blocking(move || {
        let repo = git_backend::open_repo(&repo_path).map_err(anyhow_error)?;
        if create {
            git_backend::create_branch(&repo, &branch, None).map_err(anyhow_error)?;
        } else {
            git_backend::checkout_branch(&repo, &branch).map_err(anyhow_error)?;
        }
        Ok::<(), (StatusCode, Json<ErrorResponse>)>(())
    })
    .await
    .map_err(|e| anyhow_error(anyhow::anyhow!("spawn_blocking panicked: {:?}", e)));
    match result {
        Ok(Ok(())) => { op.complete().await; notify_git_change_repo(&state, &rel).await; Ok(ok_result(format!("checked out '{}'", branch_name))) }
        Ok(Err(e)) => { op.fail(handler_err_msg(&e)).await; Err(e) }
        Err(e) => { let err = anyhow_error(anyhow::anyhow!("spawn_blocking panicked: {:?}", e)); op.fail(handler_err_msg(&err)).await; Err(err) }
    }
}

/// POST /api/v1/fleet/branch/create — create a new branch
pub async fn fleet_create_branch(
    State(state): State<WebState>,
    Json(body): Json<CreateBranchRequest>,
) -> Result<Json<GitOpResult>, (StatusCode, Json<ErrorResponse>)> {
    let root = workspace_root()?;
    let repo_path = resolve_repo_path(&root, body.repo_path.as_deref())?;
    let rel = body.repo_path.clone().unwrap_or_else(|| ".".to_string());
    let branch = body.branch.clone();
    let from_ref = body.from_ref.clone();
    let label = format!("Create branch {}: {}", body.branch, rel);
    let op = state.ops.start_op(OpKind::GitMutation, label).await;
    let branch_name = body.branch.clone();
    let result = tokio::task::spawn_blocking(move || {
        let repo = git_backend::open_repo(&repo_path).map_err(anyhow_error)?;
        git_backend::create_branch(&repo, &branch, from_ref.as_deref())
            .map_err(anyhow_error)?;
        Ok::<(), (StatusCode, Json<ErrorResponse>)>(())
    })
    .await
    .map_err(|e| anyhow_error(anyhow::anyhow!("spawn_blocking panicked: {:?}", e)));
    match result {
        Ok(Ok(())) => { op.complete().await; notify_git_change_repo(&state, &rel).await; Ok(ok_result(format!("created branch '{}'", branch_name))) }
        Ok(Err(e)) => { op.fail(handler_err_msg(&e)).await; Err(e) }
        Err(e) => { let err = anyhow_error(anyhow::anyhow!("spawn_blocking panicked: {:?}", e)); op.fail(handler_err_msg(&err)).await; Err(err) }
    }
}

// ---------------------------------------------------------------------------
// Fleet-wide mutation endpoints
// ---------------------------------------------------------------------------

/// POST /api/v1/fleet/stage-all — stage all changes in every dirty repo
pub async fn fleet_stage_all(
    State(state): State<WebState>,
) -> Result<Json<GitOpResult>, (StatusCode, Json<ErrorResponse>)> {
    let root = workspace_root()?;
    let root_clone = root.clone();
    let op = state.ops.start_op(OpKind::GitMutation, "Stage all repos").await;
    let repos = tokio::task::spawn_blocking(move || collect_repo_statuses(&root_clone))
        .await
        .map_err(|e| anyhow_error(anyhow::anyhow!("spawn_blocking panicked: {:?}", e)))?
        .map_err(internal_error)?;

    // Collect dirty repo paths for the blocking work.
    let dirty_paths: Vec<String> = repos
        .iter()
        .filter(|r| !r.is_clean && r.exists)
        .map(|r| r.path.clone())
        .collect();

    let result = tokio::task::spawn_blocking(move || {
        let mut count = 0usize;
        for rel_path in &dirty_paths {
            let repo_path = resolve_repo_path_sync(&root, Some(rel_path))?;
            let repo = git_backend::open_repo(&repo_path).map_err(anyhow_error)?;
            git_backend::stage_all(&repo).map_err(anyhow_error)?;
            count += 1;
        }
        Ok::<usize, (StatusCode, Json<ErrorResponse>)>(count)
    })
    .await
    .map_err(|e| anyhow_error(anyhow::anyhow!("spawn_blocking panicked: {:?}", e)));
    match result {
        Ok(Ok(staged)) => { op.complete().await; notify_git_change_all(&state).await; Ok(ok_result(format!("staged {} repos", staged))) }
        Ok(Err(e)) => { op.fail(handler_err_msg(&e)).await; Err(e) }
        Err(e) => { let err = anyhow_error(anyhow::anyhow!("spawn_blocking panicked: {:?}", e)); op.fail(handler_err_msg(&err)).await; Err(err) }
    }
}

/// POST /api/v1/fleet/commit-all — commit in every dirty repo
pub async fn fleet_commit_all(
    State(state): State<WebState>,
    Json(body): Json<FleetCommitRequest>,
) -> Result<Json<GitOpResult>, (StatusCode, Json<ErrorResponse>)> {
    let root = workspace_root()?;
    let root_clone = root.clone();
    let op = state.ops.start_op(OpKind::GitMutation, "Commit all repos").await;
    let repos = tokio::task::spawn_blocking(move || collect_repo_statuses(&root_clone))
        .await
        .map_err(|e| anyhow_error(anyhow::anyhow!("spawn_blocking panicked: {:?}", e)))?
        .map_err(internal_error)?;

    let dirty_paths: Vec<String> = repos
        .iter()
        .filter(|r| !r.is_clean && r.exists)
        .map(|r| r.path.clone())
        .collect();
    let message = body.message.clone();

    let result = tokio::task::spawn_blocking(move || {
        let mut count = 0usize;
        for rel_path in &dirty_paths {
            let repo_path = resolve_repo_path_sync(&root, Some(rel_path))?;
            let repo = git_backend::open_repo(&repo_path).map_err(anyhow_error)?;
            git_backend::stage_all(&repo).map_err(anyhow_error)?;
            if git_backend::commit_all(&repo, &message)
                .map_err(anyhow_error)?
                .is_some()
            {
                count += 1;
            }
        }
        Ok::<usize, (StatusCode, Json<ErrorResponse>)>(count)
    })
    .await
    .map_err(|e| anyhow_error(anyhow::anyhow!("spawn_blocking panicked: {:?}", e)));
    match result {
        Ok(Ok(committed)) => { op.complete().await; notify_git_change_all(&state).await; Ok(ok_result(format!("committed in {} repos", committed))) }
        Ok(Err(e)) => { op.fail(handler_err_msg(&e)).await; Err(e) }
        Err(e) => { let err = anyhow_error(anyhow::anyhow!("spawn_blocking panicked: {:?}", e)); op.fail(handler_err_msg(&err)).await; Err(err) }
    }
}

/// POST /api/v1/fleet/push-all — push submodules first, then superproject
pub async fn fleet_push_all(
    State(state): State<WebState>,
) -> Result<Json<GitOpResult>, (StatusCode, Json<ErrorResponse>)> {
    let root = workspace_root()?;
    let root_clone = root.clone();
    let op = state.ops.start_op(OpKind::GitMutation, "Push all repos").await;
    let repos = tokio::task::spawn_blocking(move || collect_repo_statuses(&root_clone))
        .await
        .map_err(|e| anyhow_error(anyhow::anyhow!("spawn_blocking panicked: {:?}", e)))?
        .map_err(internal_error)?;

    // Submodules first, then superproject — ordering matters for git.
    let ordered_paths: Vec<(String, bool)> = repos
        .iter()
        .filter(|r| !r.is_superproject && r.exists)
        .map(|r| (r.path.clone(), r.is_superproject))
        .chain(
            repos
                .iter()
                .filter(|r| r.is_superproject && r.exists)
                .map(|r| (r.path.clone(), r.is_superproject)),
        )
        .collect();

    let result = tokio::task::spawn_blocking(move || {
        let mut count = 0usize;
        for (rel_path, _) in &ordered_paths {
            let repo_path = resolve_repo_path_sync(&root, Some(rel_path))?;
            let repo = git_backend::open_repo(&repo_path).map_err(anyhow_error)?;
            git_backend::push_current_branch(&repo, Some("origin"), true, false)
                .map_err(anyhow_error)?;
            count += 1;
        }
        Ok::<usize, (StatusCode, Json<ErrorResponse>)>(count)
    })
    .await
    .map_err(|e| anyhow_error(anyhow::anyhow!("spawn_blocking panicked: {:?}", e)));
    match result {
        Ok(Ok(pushed)) => { op.complete().await; notify_git_change_all(&state).await; Ok(ok_result(format!("pushed {} repos", pushed))) }
        Ok(Err(e)) => { op.fail(handler_err_msg(&e)).await; Err(e) }
        Err(e) => { let err = anyhow_error(anyhow::anyhow!("spawn_blocking panicked: {:?}", e)); op.fail(handler_err_msg(&err)).await; Err(err) }
    }
}

/// POST /api/v1/fleet/pull-all — pull superproject first, then submodules
pub async fn fleet_pull_all(
    State(state): State<WebState>,
) -> Result<Json<GitOpResult>, (StatusCode, Json<ErrorResponse>)> {
    let root = workspace_root()?;
    let root_clone = root.clone();
    let op = state.ops.start_op(OpKind::GitMutation, "Pull all repos").await;
    let repos = tokio::task::spawn_blocking(move || collect_repo_statuses(&root_clone))
        .await
        .map_err(|e| anyhow_error(anyhow::anyhow!("spawn_blocking panicked: {:?}", e)))?
        .map_err(internal_error)?;

    // Superproject first, then submodules.
    let ordered_paths: Vec<String> = repos
        .iter()
        .filter(|r| r.is_superproject && r.exists)
        .map(|r| r.path.clone())
        .chain(
            repos
                .iter()
                .filter(|r| !r.is_superproject && r.exists)
                .map(|r| r.path.clone()),
        )
        .collect();

    let result = tokio::task::spawn_blocking(move || {
        let mut count = 0usize;
        for rel_path in &ordered_paths {
            let repo_path = resolve_repo_path_sync(&root, Some(rel_path))?;
            let repo = git_backend::open_repo(&repo_path).map_err(anyhow_error)?;
            git_backend::pull(&repo, false).map_err(anyhow_error)?;
            count += 1;
        }
        Ok::<usize, (StatusCode, Json<ErrorResponse>)>(count)
    })
    .await
    .map_err(|e| anyhow_error(anyhow::anyhow!("spawn_blocking panicked: {:?}", e)));
    match result {
        Ok(Ok(pulled)) => { op.complete().await; notify_git_change_all(&state).await; Ok(ok_result(format!("pulled {} repos", pulled))) }
        Ok(Err(e)) => { op.fail(handler_err_msg(&e)).await; Err(e) }
        Err(e) => { let err = anyhow_error(anyhow::anyhow!("spawn_blocking panicked: {:?}", e)); op.fail(handler_err_msg(&err)).await; Err(err) }
    }
}

/// POST /api/v1/fleet/checkout-all — checkout same branch in all repos
pub async fn fleet_checkout_all(
    State(state): State<WebState>,
    Json(body): Json<FleetCheckoutRequest>,
) -> Result<Json<GitOpResult>, (StatusCode, Json<ErrorResponse>)> {
    let root = workspace_root()?;
    let root_clone = root.clone();
    let op = state.ops.start_op(OpKind::GitMutation, format!("Checkout all: {}", body.branch)).await;
    let repos = tokio::task::spawn_blocking(move || collect_repo_statuses(&root_clone))
        .await
        .map_err(|e| anyhow_error(anyhow::anyhow!("spawn_blocking panicked: {:?}", e)))?
        .map_err(internal_error)?;

    let existing_paths: Vec<String> = repos
        .iter()
        .filter(|r| r.exists)
        .map(|r| r.path.clone())
        .collect();
    let branch = body.branch.clone();
    let create = body.create.unwrap_or(false);
    let branch_name = body.branch.clone();

    let result = tokio::task::spawn_blocking(move || {
        let mut count = 0usize;
        for rel_path in &existing_paths {
            let repo_path = resolve_repo_path_sync(&root, Some(rel_path))?;
            let repo = git_backend::open_repo(&repo_path).map_err(anyhow_error)?;
            if create {
                git_backend::create_branch(&repo, &branch, None).map_err(anyhow_error)?;
            } else {
                git_backend::checkout_branch(&repo, &branch).map_err(anyhow_error)?;
            }
            count += 1;
        }
        Ok::<usize, (StatusCode, Json<ErrorResponse>)>(count)
    })
    .await
    .map_err(|e| anyhow_error(anyhow::anyhow!("spawn_blocking panicked: {:?}", e)));
    match result {
        Ok(Ok(switched)) => { op.complete().await; notify_git_change_all(&state).await; Ok(ok_result(format!("checked out '{}' in {} repos", branch_name, switched))) }
        Ok(Err(e)) => { op.fail(handler_err_msg(&e)).await; Err(e) }
        Err(e) => { let err = anyhow_error(anyhow::anyhow!("spawn_blocking panicked: {:?}", e)); op.fail(handler_err_msg(&err)).await; Err(err) }
    }
}

/// POST /api/v1/fleet/merge-back — 5-phase merge-back:
/// stage → commit → push feature, checkout main, ff-merge, push main
pub async fn fleet_merge_back(
    State(state): State<WebState>,
    Json(body): Json<FleetMergeBackRequest>,
) -> Result<Json<GitOpResult>, (StatusCode, Json<ErrorResponse>)> {
    let root = workspace_root()?;
    let root_clone = root.clone();
    let repos = tokio::task::spawn_blocking(move || collect_repo_statuses(&root_clone))
        .await
        .map_err(|e| anyhow_error(anyhow::anyhow!("spawn_blocking panicked: {:?}", e)))?
        .map_err(internal_error)?;

    let commit_msg = body
        .message
        .unwrap_or_else(|| "merge-back: merge feature into main".to_string());

    // Build per-repo inputs: (rel_path, branch, is_clean)
    // Validate branch availability before entering spawn_blocking.
    let mut repo_inputs: Vec<(String, String, bool)> = Vec::new();
    for repo_status in &repos {
        if !repo_status.exists {
            continue;
        }
        let current_branch = match &repo_status.branch {
            Some(b) => b.clone(),
            None => {
                return Err((
                    StatusCode::BAD_REQUEST,
                    Json(ErrorResponse {
                        error: format!(
                            "repo '{}' is in detached HEAD state, cannot merge-back",
                            repo_status.name
                        ),
                    }),
                ));
            }
        };
        repo_inputs.push((
            repo_status.path.clone(),
            current_branch,
            repo_status.is_clean,
        ));
    }

    let op = state.ops.start_op(OpKind::GitMutation, "Merge-back all repos").await;
    let result = tokio::task::spawn_blocking(move || {
        for (rel_path, current_branch, is_clean) in &repo_inputs {
            let repo_path = resolve_repo_path_sync(&root, Some(rel_path))?;
            let repo = git_backend::open_repo(&repo_path).map_err(anyhow_error)?;
            let default_branch = git_backend::get_default_branch(&repo).map_err(anyhow_error)?;

            if current_branch == &default_branch {
                continue;
            }

            if !is_clean {
                git_backend::stage_all(&repo).map_err(anyhow_error)?;
            }
            git_backend::commit_all(&repo, &commit_msg).map_err(anyhow_error)?;
            git_backend::push_current_branch(&repo, Some("origin"), true, false)
                .map_err(anyhow_error)?;
            git_backend::checkout_branch(&repo, &default_branch).map_err(anyhow_error)?;
            git_backend::merge_ff(&repo, current_branch).map_err(anyhow_error)?;
            git_backend::push_current_branch(&repo, Some("origin"), false, false)
                .map_err(anyhow_error)?;
        }
        Ok::<(), (StatusCode, Json<ErrorResponse>)>(())
    })
    .await
    .map_err(|e| anyhow_error(anyhow::anyhow!("spawn_blocking panicked: {:?}", e)));
    match result {
        Ok(Ok(())) => { op.complete().await; notify_git_change_all(&state).await; Ok(ok_result("merge-back complete")) }
        Ok(Err(e)) => { op.fail(handler_err_msg(&e)).await; Err(e) }
        Err(e) => { let err = anyhow_error(anyhow::anyhow!("spawn_blocking panicked: {:?}", e)); op.fail(handler_err_msg(&err)).await; Err(err) }
    }
}

// ---------------------------------------------------------------------------
// Background git watcher
// ---------------------------------------------------------------------------

/// Trigger a targeted re-scan of a single repo after a mutation.
async fn notify_git_change_repo(state: &WebState, repo_path: &str) {
    let _ = state.git_nudge_tx.send(Some(repo_path.to_string()));
}

/// Trigger a full re-scan of all repos after a fleet-wide mutation.
async fn notify_git_change_all(state: &WebState) {
    let _ = state.git_nudge_tx.send(None);
}

/// Spawn a per-repo filesystem watcher on each repo's working tree.
///
/// When any tracked file in a repo's working tree changes (edits, saves,
/// creates, deletes), the OS event fires and we send a targeted nudge
/// `Some(rel_path)` so only that repo is re-scanned — not the entire fleet.
///
/// Design:
/// - One debouncer per repo, each with a 200ms window so rapid saves
///   (autosave, formatter) collapse into a single git status call.
/// - Watches the repo root non-recursively for direct children, plus
///   recursively for all subdirectories, but excludes `.git/` so we don't
///   react to git's own internal writes (refs, index, COMMIT_EDITMSG, etc.) —
///   those are already covered by mutation handlers calling notify_git_change.
/// - The watcher thread parks permanently; the debouncers drive all activity
///   via their own internal threads (notify's design).
pub fn spawn_git_fs_watcher(state: WebState) {
    use notify_debouncer_mini::{new_debouncer, notify::RecursiveMode};
    use std::time::Duration;

    let workspace_root = state.workspace_root.clone();
    let nudge_tx = state.git_nudge_tx.clone();

    std::thread::Builder::new()
        .name("git-fs-watcher".to_string())
        .spawn(move || {
            let repos = match build_repo_list(&workspace_root) {
                Ok(r) => r,
                Err(e) => {
                    tracing::warn!(error = %e, "git fs watcher: failed to build repo list");
                    return;
                }
            };

            if repos.is_empty() {
                tracing::warn!("git fs watcher: no repos found, watcher idle");
                return;
            }

            // One debouncer per repo. We keep them all alive in a Vec so they
            // are not dropped (dropping a debouncer stops the watcher).
            let mut debouncers = Vec::with_capacity(repos.len());

            for repo in &repos {
                let rel_path = repo.rel_path.clone();
                let nudge = nudge_tx.clone();

                let mut debouncer = match new_debouncer(
                    Duration::from_millis(200),
                    move |result: Result<Vec<notify_debouncer_mini::DebouncedEvent>, notify_debouncer_mini::notify::Error>| {
                        match result {
                            Ok(events) => {
                                // Filter out events inside .git/ (internal git
                                // operations) and .wsb/ (wsb's own runtime files:
                                // sessions.db, workbench.json, logs, etc.).
                                // Without this, wsb's own writes trigger a scan
                                // which causes more writes — a feedback loop.
                                let has_workdir_change = events.iter().any(|ev| {
                                    !ev.path.components().any(|c| {
                                        let s = c.as_os_str();
                                        s == ".git" || s == ".wsb"
                                    })
                                });
                                if has_workdir_change {
                                    let _ = nudge.send(Some(rel_path.clone()));
                                }
                            }
                            Err(e) => {
                                tracing::warn!(
                                    repo = %rel_path,
                                    error = %e,
                                    "git fs watcher: notify error"
                                );
                            }
                        }
                    },
                ) {
                    Ok(d) => d,
                    Err(e) => {
                        tracing::warn!(
                            repo = %repo.rel_path,
                            error = %e,
                            "git fs watcher: failed to create debouncer for repo"
                        );
                        continue;
                    }
                };

                // Watch the repo's working tree recursively. notify on macOS
                // uses FSEvents which natively excludes nothing, so we filter
                // .git/ in the event callback above.
                if let Err(e) = debouncer.watcher().watch(&repo.abs_path, RecursiveMode::Recursive) {
                    tracing::warn!(
                        path = %repo.abs_path.display(),
                        error = %e,
                        "git fs watcher: failed to watch repo working tree"
                    );
                    continue;
                }

                debouncers.push(debouncer);
            }

            tracing::info!(
                "git fs watcher: watching working trees of {} repos",
                debouncers.len()
            );

            // Park this thread permanently. The debouncers and their internal
            // notify threads are alive as long as the Vec lives here.
            std::thread::park();
        })
        .expect("git-fs-watcher thread failed to spawn");
}

/// A repo entry descriptor built once from .gitmodules, used to drive
/// the incremental per-repo scan loop.
#[derive(Clone)]
struct RepoDescriptor {
    /// Absolute filesystem path to the repo.
    abs_path: PathBuf,
    /// Repo name (directory / submodule name).
    name: String,
    /// Workspace-relative path used as the key in the status map.
    rel_path: String,
    is_superproject: bool,
}

/// Build the ordered list of repo descriptors for the workspace.
/// Returns superproject first, then submodules in .gitmodules order.
fn build_repo_list(
    repo_root: &Path,
) -> crate::fleet::error::Result<Vec<RepoDescriptor>> {
    let super_name = repo_root
        .file_name()
        .map(|n| n.to_string_lossy().into_owned())
        .unwrap_or_else(|| repo_root.display().to_string());

    let mut list = vec![RepoDescriptor {
        abs_path: repo_root.to_path_buf(),
        name: super_name,
        rel_path: ".".to_string(),
        is_superproject: true,
    }];

    let submodules = match GitModules::parse(repo_root) {
        Ok(gm) => gm.submodules,
        Err(crate::fleet::error::SuperGitError::NoSubmodules) => Vec::new(),
        Err(e) => return Err(e),
    };

    for sub in &submodules {
        list.push(RepoDescriptor {
            abs_path: repo_root.join(&sub.path),
            name: sub.name.clone(),
            rel_path: sub.path.display().to_string(),
            is_superproject: false,
        });
    }

    Ok(list)
}

/// Spawn the background git status watcher.
///
/// Design:
/// - Nudge channel carries `Option<String>`: `None` = full scan all repos,
///   `Some(rel_path)` = scan only that repo (targeted, from FS watcher).
/// - The 30-second idle interval is a fallback for edge cases where the FS
///   watcher misses an event (e.g. network FS, NFS mounts). Normal operation
///   is entirely event-driven via the nudge channel.
/// - Builds the repo list fresh on every full scan so newly added/removed
///   submodules are picked up automatically.
/// - Each scan (full or targeted) runs all relevant repos concurrently via
///   JoinSet — no waiting for one slow repo before starting the next.
/// - Each repo result is broadcast as soon as it finishes — no accumulation.
/// - Only broadcasts a repo when its serialized status actually changed.
/// - RwLock on the snapshot: multiple SSE clients read simultaneously; the
///   watcher only takes a write lock briefly per changed repo.
pub fn spawn_git_watcher(state: WebState) {
    let workspace_root = state.workspace_root.clone();
    let tx = state.git_status_tx.clone();
    let snapshot = state.git_status_snapshot.clone();
    let ops = state.ops.clone();
    let mut focus_rx = state.git_focus_tx.subscribe();
    let mut nudge_rx = state.git_nudge_tx.subscribe();

    tokio::spawn(async move {
        // Fallback interval — only fires if no FS events arrive. Normal
        // operation is entirely event-driven.
        let idle_interval = tokio::time::Duration::from_secs(30);
        // Per-repo last-seen JSON for change detection, keyed by rel_path.
        let mut last_seen: std::collections::HashMap<String, String> =
            std::collections::HashMap::new();
        // Cached repo list — rebuilt on every full scan.
        let mut cached_repos: Vec<RepoDescriptor> = Vec::new();

        loop {
            // Collect targeted repo set from pending nudges.
            // None in the set means full scan; Some(rel_path) means targeted.
            let mut targeted: std::collections::HashSet<String> = std::collections::HashSet::new();
            let mut full_scan = false;

            // Wait for: idle timeout, focus change, or nudge.
            tokio::select! {
                _ = tokio::time::sleep(idle_interval) => {
                    full_scan = true;
                }
                Ok(nudge) = nudge_rx.recv() => {
                    match nudge {
                        None => full_scan = true,
                        Some(rel_path) => { targeted.insert(rel_path); }
                    }
                }
                Ok(_) = focus_rx.recv() => {
                    full_scan = true;
                }
            }
            // Drain all pending nudges and focus changes without blocking.
            while let Ok(nudge) = nudge_rx.try_recv() {
                match nudge {
                    None => full_scan = true,
                    Some(rel_path) => { targeted.insert(rel_path); }
                }
            }
            while focus_rx.try_recv().is_ok() {}

            // Rebuild repo list on full scans (picks up new/removed submodules).
            if full_scan {
                let root = workspace_root.clone();
                cached_repos = match tokio::task::spawn_blocking(move || build_repo_list(&root)).await {
                    Ok(Ok(r)) => r,
                    Ok(Err(e)) => {
                        tracing::warn!(error = %e, "git watcher: failed to list repos");
                        continue;
                    }
                    Err(e) => {
                        tracing::warn!(error = %e, "git watcher: build_repo_list panicked");
                        continue;
                    }
                };
            }

            // Determine which repos to scan this cycle.
            let repos_to_scan: Vec<RepoDescriptor> = if full_scan {
                cached_repos.clone()
            } else {
                // Targeted: only repos whose rel_path was nudged.
                // If the targeted path isn't in the cached list (e.g. new
                // submodule added since last full scan), trigger a full scan
                // instead so the list stays accurate.
                let found: Vec<RepoDescriptor> = cached_repos
                    .iter()
                    .filter(|r| targeted.contains(&r.rel_path))
                    .cloned()
                    .collect();
                if found.len() < targeted.len() {
                    // At least one nudged path is unknown — rebuild list.
                    let root = workspace_root.clone();
                    cached_repos = match tokio::task::spawn_blocking(move || build_repo_list(&root)).await {
                        Ok(Ok(r)) => r,
                        Ok(Err(e)) => {
                            tracing::warn!(error = %e, "git watcher: failed to list repos (targeted fallback)");
                            continue;
                        }
                        Err(e) => {
                            tracing::warn!(error = %e, "git watcher: build_repo_list panicked (targeted fallback)");
                            continue;
                        }
                    };
                    cached_repos.clone()
                } else {
                    found
                }
            };

            if repos_to_scan.is_empty() {
                continue;
            }

            let repo_count = repos_to_scan.len();
            let scan_label = if full_scan {
                format!("Scan {} repos", repo_count)
            } else {
                let names: Vec<&str> = repos_to_scan.iter().map(|r| r.name.as_str()).collect();
                format!("Scan: {}", names.join(", "))
            };
            let scan_op = ops.start_op(OpKind::GitScan, scan_label).await;

            // Scan all repos in this cycle concurrently.
            let mut join_set: tokio::task::JoinSet<(String, Result<RepoDetailedStatus, crate::fleet::error::SuperGitError>)> =
                tokio::task::JoinSet::new();

            for desc in repos_to_scan {
                let root = workspace_root.clone();
                join_set.spawn_blocking(move || {
                    let result = collect_detailed_repo_status(
                        &root,
                        &desc.abs_path,
                        &desc.name,
                        &desc.rel_path,
                        desc.is_superproject,
                    );
                    (desc.rel_path, result)
                });
            }

            // Stream each changed repo immediately as it finishes.
            let mut repos_done = 0usize;

            while let Some(join_result) = join_set.join_next().await {
                repos_done += 1;
                if repo_count > 0 {
                    scan_op.progress(repos_done as f32 / repo_count as f32).await;
                }

                let (rel_path, status_result) = match join_result {
                    Ok(pair) => pair,
                    Err(e) => {
                        tracing::warn!(error = %e, "git watcher: spawn_blocking panicked");
                        continue;
                    }
                };

                let repo_status = match status_result {
                    Ok(s) => s,
                    Err(e) => {
                        tracing::warn!(
                            path = %rel_path,
                            error = %e,
                            "git watcher: failed to collect repo status"
                        );
                        continue;
                    }
                };

                // Only broadcast if this repo's status actually changed.
                let json = match serde_json::to_string(&repo_status) {
                    Ok(j) => j,
                    Err(_) => continue,
                };

                let changed = last_seen
                    .get(&repo_status.path)
                    .map_or(true, |prev| prev != &json);

                if !changed {
                    continue;
                }

                last_seen.insert(repo_status.path.clone(), json);

                {
                    let mut snap = snapshot.write().await;
                    snap.insert(repo_status.path.clone(), repo_status.clone());
                }

                // Broadcast immediately — clients receive each repo update
                // as soon as it is ready, not after all repos finish.
                let _ = tx.send(repo_status);
            }

            scan_op.complete().await;
        }
    });
}

// ---------------------------------------------------------------------------
// Tests
// ---------------------------------------------------------------------------

#[cfg(test)]
mod tests {
    use super::*;
    use std::process::Command;
    use tempfile::TempDir;

    #[test]
    fn workspace_root_resolves_from_nested_directory() {
        let temp_dir = init_repo();
        let nested = temp_dir.path().join("nested").join("dir");
        std::fs::create_dir_all(&nested).unwrap();

        let resolved = GitOps::discover_repo_root(&nested).unwrap();
        assert_eq!(
            resolved.canonicalize().unwrap(),
            temp_dir.path().canonicalize().unwrap()
        );
    }

    #[test]
    fn collect_issue_repos_includes_superproject_and_github_submodules() {
        let temp_dir = init_repo();
        std::fs::write(
            temp_dir.path().join(".gitmodules"),
            r#"[submodule "libs/core"]
	path = libs/core
	url = git@github.com:example/core.git
"#,
        )
        .unwrap();

        let repos = collect_issue_repos(temp_dir.path()).unwrap();
        assert_eq!(repos.len(), 2);
        assert_eq!(repos[0].github.owner, "example");
        assert_eq!(repos[0].github.repo, "workspace");
        assert_eq!(repos[1].github.repo, "core");
    }

    #[test]
    fn collect_issue_repos_skips_non_github_submodule_remotes() {
        let temp_dir = init_repo();
        std::fs::write(
            temp_dir.path().join(".gitmodules"),
            r#"[submodule "core"]
	path = libs/core
	url = git@github.com:example/core.git
[submodule "coco_captions_small"]
	path = automation/test_files/coco_captions_small
	url = https://huggingface.co/datasets/Erland/coco_captions_small
"#,
        )
        .unwrap();

        let repos = collect_issue_repos(temp_dir.path()).unwrap();
        assert_eq!(repos.len(), 2);
        assert_eq!(repos[0].github.repo, "workspace");
        assert_eq!(repos[1].github.repo, "core");
    }

    #[test]
    fn collect_issue_repos_returns_empty_when_workspace_remote_is_not_github() {
        let temp_dir = TempDir::new().unwrap();
        Command::new("git")
            .arg("init")
            .current_dir(temp_dir.path())
            .output()
            .unwrap();
        Command::new("git")
            .args(["config", "user.email", "wsb-tests@example.com"])
            .current_dir(temp_dir.path())
            .output()
            .unwrap();
        Command::new("git")
            .args(["config", "user.name", "wsb tests"])
            .current_dir(temp_dir.path())
            .output()
            .unwrap();
        std::fs::write(temp_dir.path().join("README.md"), "workspace\n").unwrap();
        Command::new("git")
            .args(["add", "README.md"])
            .current_dir(temp_dir.path())
            .output()
            .unwrap();
        Command::new("git")
            .args(["commit", "-m", "init"])
            .current_dir(temp_dir.path())
            .output()
            .unwrap();
        Command::new("git")
            .args([
                "remote",
                "add",
                "origin",
                "https://huggingface.co/datasets/example/workspace",
            ])
            .current_dir(temp_dir.path())
            .output()
            .unwrap();

        let repos = collect_issue_repos(temp_dir.path()).unwrap();
        assert!(repos.is_empty());
    }

    #[test]
    fn collect_repo_statuses_returns_repos_shape() {
        let temp_dir = init_repo();
        std::fs::write(
            temp_dir.path().join(".gitmodules"),
            r#"[submodule "libs/core"]
	path = libs/core
	url = git@github.com:example/core.git
"#,
        )
        .unwrap();
        std::fs::create_dir_all(temp_dir.path().join("libs/core")).unwrap();
        Command::new("git")
            .arg("init")
            .arg(temp_dir.path().join("libs/core"))
            .output()
            .unwrap();
        Command::new("git")
            .args(["config", "user.email", "wsb-tests@example.com"])
            .current_dir(temp_dir.path().join("libs/core"))
            .output()
            .unwrap();
        Command::new("git")
            .args(["config", "user.name", "wsb tests"])
            .current_dir(temp_dir.path().join("libs/core"))
            .output()
            .unwrap();
        std::fs::write(temp_dir.path().join("libs/core/README.md"), "core\n").unwrap();
        Command::new("git")
            .args(["add", "README.md"])
            .current_dir(temp_dir.path().join("libs/core"))
            .output()
            .unwrap();
        Command::new("git")
            .args(["commit", "-m", "init"])
            .current_dir(temp_dir.path().join("libs/core"))
            .output()
            .unwrap();

        let repos = collect_repo_statuses(temp_dir.path()).unwrap();
        let json = serde_json::to_value(&repos).unwrap();
        assert!(json.is_array());
        assert_eq!(json.as_array().unwrap().len(), 2);
        assert!(json.as_array().unwrap()[0]
            .get("is_superproject")
            .unwrap()
            .as_bool()
            .unwrap());
        // Verify files field exists and is an array
        assert!(json.as_array().unwrap()[0]
            .get("files")
            .unwrap()
            .is_array());
    }

    #[test]
    fn detailed_status_classifies_file_states() {
        let temp_dir = init_repo();

        // Create an untracked file
        std::fs::write(temp_dir.path().join("untracked.txt"), "new file\n").unwrap();

        // Modify a tracked file
        std::fs::write(temp_dir.path().join("README.md"), "modified\n").unwrap();

        let repos = collect_repo_statuses(temp_dir.path()).unwrap();
        let super_repo = &repos[0];
        assert!(!super_repo.is_clean);
        assert!(super_repo.files.len() >= 2);

        // Find the untracked file
        let untracked = super_repo
            .files
            .iter()
            .find(|f| f.path == "untracked.txt")
            .expect("untracked.txt should be in status");
        assert_eq!(untracked.status, FileStatusKind::Untracked);

        // Find the modified file
        let modified = super_repo
            .files
            .iter()
            .find(|f| f.path == "README.md")
            .expect("README.md should be in status");
        assert_eq!(modified.status, FileStatusKind::Modified);
    }

    #[test]
    fn detailed_status_classifies_staged_files() {
        let temp_dir = init_repo();

        // Modify and stage a file
        std::fs::write(temp_dir.path().join("README.md"), "staged change\n").unwrap();
        Command::new("git")
            .args(["add", "README.md"])
            .current_dir(temp_dir.path())
            .output()
            .unwrap();

        let repos = collect_repo_statuses(temp_dir.path()).unwrap();
        let super_repo = &repos[0];
        assert!(!super_repo.is_clean);

        let staged = super_repo
            .files
            .iter()
            .find(|f| f.path == "README.md")
            .expect("README.md should be in status");
        assert_eq!(staged.status, FileStatusKind::Staged);
    }

    #[test]
    fn detailed_status_keeps_ignored_files_out_of_dirty_state() {
        let temp_dir = init_repo();

        std::fs::write(temp_dir.path().join(".gitignore"), ".wsb/\nignored.log\n").unwrap();
        Command::new("git")
            .args(["add", ".gitignore"])
            .current_dir(temp_dir.path())
            .output()
            .unwrap();
        Command::new("git")
            .args(["commit", "-m", "add ignore rules"])
            .current_dir(temp_dir.path())
            .output()
            .unwrap();

        std::fs::create_dir_all(temp_dir.path().join(".wsb")).unwrap();
        std::fs::write(temp_dir.path().join(".wsb/cache.db"), "cache\n").unwrap();
        std::fs::write(temp_dir.path().join("ignored.log"), "log\n").unwrap();

        let repos = collect_repo_statuses(temp_dir.path()).unwrap();
        let super_repo = &repos[0];
        assert!(super_repo.is_clean);

        let ignored = super_repo
            .files
            .iter()
            .map(|f| (f.path.clone(), f.status))
            .collect::<std::collections::BTreeMap<_, _>>();
        assert_eq!(ignored.get("ignored.log"), Some(&FileStatusKind::Ignored));
        assert!(ignored
            .iter()
            .any(|(path, status)| (path == ".wsb" || path.starts_with(".wsb/")) && *status == FileStatusKind::Ignored));
    }

    #[test]
    fn untracked_dirs_with_only_ignored_or_empty_content_are_excluded() {
        let temp_dir = init_repo();

        // Set up gitignore rules
        std::fs::write(
            temp_dir.path().join(".gitignore"),
            "Package.resolved\n*.xcuserdatad/\n",
        )
        .unwrap();
        Command::new("git")
            .args(["add", ".gitignore"])
            .current_dir(temp_dir.path())
            .output()
            .unwrap();
        Command::new("git")
            .args(["commit", "-m", "add gitignore"])
            .current_dir(temp_dir.path())
            .output()
            .unwrap();

        // Create a directory whose only file is gitignored (mirrors
        // xcshareddata/swiftpm/Package.resolved in Xcode projects)
        std::fs::create_dir_all(
            temp_dir.path().join("Foo.xcworkspace/xcshareddata/swiftpm"),
        )
        .unwrap();
        std::fs::write(
            temp_dir
                .path()
                .join("Foo.xcworkspace/xcshareddata/swiftpm/Package.resolved"),
            "{}",
        )
        .unwrap();

        // Create a directory tree that is entirely empty (mirrors .swiftpm
        // directories with only empty subdirs)
        std::fs::create_dir_all(
            temp_dir.path().join(".swiftpm/xcode/configuration"),
        )
        .unwrap();

        // Create a directory with only ignored content nested deeper
        std::fs::create_dir_all(
            temp_dir
                .path()
                .join("deep/nested.xcuserdatad/schemes"),
        )
        .unwrap();
        std::fs::write(
            temp_dir
                .path()
                .join("deep/nested.xcuserdatad/schemes/data.plist"),
            "<plist/>",
        )
        .unwrap();

        let repos = collect_repo_statuses(temp_dir.path()).unwrap();
        let super_repo = &repos[0];

        // None of these phantom directories should appear as untracked
        let untracked_paths: Vec<&str> = super_repo
            .files
            .iter()
            .filter(|f| f.status == FileStatusKind::Untracked)
            .map(|f| f.path.as_str())
            .collect();

        assert!(
            !untracked_paths.iter().any(|p| p.contains("xcshareddata")),
            "directory with only gitignored files should not appear as untracked, got: {:?}",
            untracked_paths,
        );
        assert!(
            !untracked_paths.iter().any(|p| p.contains(".swiftpm")),
            "empty directory tree should not appear as untracked, got: {:?}",
            untracked_paths,
        );

        // The repo should be clean (only ignored content was added)
        assert!(
            super_repo.is_clean,
            "repo should be clean when only ignored/empty dirs exist, but found: {:?}",
            super_repo
                .files
                .iter()
                .filter(|f| f.status != FileStatusKind::Ignored)
                .map(|f| format!("{}: {:?}", f.path, f.status))
                .collect::<Vec<_>>(),
        );
    }

    #[test]
    fn resolve_repo_path_rejects_escape() {
        let temp_dir = TempDir::new().unwrap();
        let root = temp_dir.path();
        std::fs::create_dir_all(root.join("sub")).unwrap();
        // Create a directory outside root to escape to
        let outside = temp_dir.path().join("../escape_target");
        std::fs::create_dir_all(&outside).unwrap();

        let result = resolve_repo_path(root, Some("../escape_target"));
        assert!(result.is_err());
    }

    fn init_repo() -> TempDir {
        let temp_dir = TempDir::new().unwrap();
        Command::new("git")
            .arg("init")
            .current_dir(temp_dir.path())
            .output()
            .unwrap();
        Command::new("git")
            .args(["config", "user.email", "wsb-tests@example.com"])
            .current_dir(temp_dir.path())
            .output()
            .unwrap();
        Command::new("git")
            .args(["config", "user.name", "wsb tests"])
            .current_dir(temp_dir.path())
            .output()
            .unwrap();
        std::fs::write(temp_dir.path().join("README.md"), "workspace\n").unwrap();
        Command::new("git")
            .args(["add", "README.md"])
            .current_dir(temp_dir.path())
            .output()
            .unwrap();
        Command::new("git")
            .args(["commit", "-m", "init"])
            .current_dir(temp_dir.path())
            .output()
            .unwrap();
        Command::new("git")
            .arg("remote")
            .arg("add")
            .arg("origin")
            .arg("git@github.com:example/workspace.git")
            .current_dir(temp_dir.path())
            .output()
            .unwrap();
        temp_dir
    }
}