xbp_cli/commands/

worktree_watch.rs

use crate::commands::cli_session::{cli_request_client, resolve_cli_access_token};
use crate::config::{resolve_device_identity, ApiConfig};
use chrono::{DateTime, Utc};
use notify::event::{CreateKind, ModifyKind, RemoveKind, RenameMode};
use notify::{Config, Event, EventKind, RecommendedWatcher, RecursiveMode, Watcher};
use reqwest::StatusCode;
use serde::{Deserialize, Serialize};
use serde_json::{json, Value};
use sha2::{Digest, Sha256};
use std::collections::{BTreeMap, BTreeSet};
use std::fs::{self, File, OpenOptions};
use std::io::{BufRead, BufReader, Write};
use std::path::{Path, PathBuf};
use std::process::{Command, Stdio};
use std::sync::mpsc;
use std::time::{Duration, Instant};
use sysinfo::{Pid, System};
use uuid::Uuid;

const BACKGROUND_CHILD_ENV: &str = "XBP_WORKTREE_WATCH_BACKGROUND_CHILD";
const DEFAULT_REMOTE: &str = "origin";
const EVENT_DEDUPE_WINDOW_SECONDS: i64 = 5;

#[cfg(windows)]
const CREATE_NO_WINDOW: u32 = 0x08000000;

#[derive(Debug, Clone)]
pub struct WorktreeWatchTargetOptions {
    pub repo: Option<PathBuf>,
    pub parent: Option<PathBuf>,
}

#[derive(Debug, Clone)]
pub struct WorktreeWatchStartOptions {
    pub target: WorktreeWatchTargetOptions,
    pub detach: bool,
    pub sync_interval_seconds: u64,
    pub once: bool,
}

#[derive(Debug, Clone)]
pub struct WorktreeWatchSyncOptions {
    pub target: WorktreeWatchTargetOptions,
    pub dry_run: bool,
    pub resync: bool,
}

#[derive(Debug, Clone)]
pub struct WorktreeWatchStatusOptions {
    pub target: WorktreeWatchTargetOptions,
    pub json: bool,
    pub records: bool,
    pub record_limit: usize,
    pub stats: bool,
    pub stats_gap_minutes: u64,
}

#[derive(Debug, Clone)]
struct RepoIdentity {
    owner: String,
    name: String,
    branch: String,
    root: PathBuf,
    head_sha: Option<String>,
}

#[derive(Debug, Clone)]
struct SpoolLayout {
    root: PathBuf,
    events_file: PathBuf,
    commits_file: PathBuf,
    stats_file: PathBuf,
    watcher_state_file: PathBuf,
    sync_log_file: PathBuf,
    event_dedupe_file: PathBuf,
}

#[derive(Debug, Serialize, Deserialize, Clone)]
#[serde(rename_all = "camelCase")]
struct WorktreeMutationEvent {
    id: String,
    repo_owner: String,
    repo_name: String,
    branch_name: String,
    repo_root: String,
    head_sha: Option<String>,
    event_kind: String,
    paths: Vec<String>,
    primary_path: Option<String>,
    old_path: Option<String>,
    new_path: Option<String>,
    added_lines: Option<u64>,
    removed_lines: Option<u64>,
    file_created: bool,
    file_removed: bool,
    folder_created: bool,
    folder_removed: bool,
    renamed_or_moved: bool,
    raw_kind: String,
    occurred_at: DateTime<Utc>,
}

#[derive(Debug, Serialize, Deserialize, Clone)]
#[serde(rename_all = "camelCase")]
struct WorktreeCommitEvent {
    id: String,
    repo_owner: String,
    repo_name: String,
    branch_name: String,
    repo_root: String,
    previous_head_sha: Option<String>,
    head_sha: String,
    subject: Option<String>,
    author_name: Option<String>,
    author_email: Option<String>,
    committed_at: Option<String>,
    occurred_at: DateTime<Utc>,
}

#[derive(Debug, Serialize)]
#[serde(rename_all = "camelCase")]
struct WorktreeMutationIngestPayload {
    device: WorktreeMutationDevicePayload,
    repository: WorktreeMutationRepositoryPayload,
    events: Vec<WorktreeMutationEvent>,
    commits: Vec<WorktreeCommitEvent>,
}

#[derive(Debug, Serialize)]
#[serde(rename_all = "camelCase")]
struct WorktreeMutationDevicePayload {
    hardware_id: String,
    device_name: Option<String>,
    hostname: Option<String>,
    platform: String,
}

#[derive(Debug, Serialize)]
#[serde(rename_all = "camelCase")]
struct WorktreeMutationRepositoryPayload {
    owner: String,
    name: String,
    branch_name: String,
    repo_root: String,
}

#[derive(Debug)]
struct SyncCandidate {
    path: PathBuf,
    kind: SyncFileKind,
    records: Vec<Value>,
}

#[derive(Debug, Clone, Copy)]
enum SyncFileKind {
    Events,
    Commits,
}

#[derive(Debug, Serialize, Clone)]
#[serde(rename_all = "camelCase")]
struct WorktreeWatchStatsSummary {
    generated_at: DateTime<Utc>,
    repository_owner: String,
    repository_name: String,
    branch_name: String,
    repo_root: String,
    spool_root: String,
    session_gap_minutes: u64,
    total_events: u64,
    total_commits: u64,
    first_event_at: Option<DateTime<Utc>>,
    last_event_at: Option<DateTime<Utc>>,
    estimated_coding_seconds: u64,
    added_lines: u64,
    removed_lines: u64,
    by_file: Vec<WorktreeWatchFileStats>,
    by_filetype: Vec<WorktreeWatchBucketStats>,
    by_event_kind: Vec<WorktreeWatchBucketStats>,
}

#[derive(Debug, Serialize, Clone, Default)]
#[serde(rename_all = "camelCase")]
struct WorktreeWatchFileStats {
    path: String,
    filetype: String,
    event_count: u64,
    estimated_coding_seconds: u64,
    added_lines: u64,
    removed_lines: u64,
}

#[derive(Debug, Serialize, Clone, Default)]
#[serde(rename_all = "camelCase")]
struct WorktreeWatchBucketStats {
    name: String,
    event_count: u64,
    estimated_coding_seconds: u64,
    added_lines: u64,
    removed_lines: u64,
}

#[derive(Debug, Serialize, Deserialize, Clone)]
#[serde(rename_all = "camelCase")]
struct WorktreeWatcherState {
    pid: u32,
    repo_root: String,
    repository_owner: String,
    repository_name: String,
    branch_name: String,
    executable: String,
    started_at: DateTime<Utc>,
}

#[derive(Debug, Serialize, Deserialize, Clone)]
#[serde(rename_all = "camelCase")]
struct WorktreeWatchSyncLogEntry {
    id: String,
    synced_at: DateTime<Utc>,
    endpoint: String,
    repository_owner: String,
    repository_name: String,
    branch_name: String,
    repo_root: String,
    resync: bool,
    status_code: u16,
    event_count: usize,
    commit_count: usize,
    spool_file_count: usize,
    spool_files: Vec<String>,
}

#[derive(Debug, Serialize, Deserialize, Clone)]
#[serde(rename_all = "camelCase")]
struct WorktreeWatchEventDedupeRecord {
    fingerprint: String,
    first_seen_at: DateTime<Utc>,
    event_kind: String,
    primary_path: Option<String>,
}

#[derive(Debug, Default)]
struct RecentEventDedupe {
    fingerprints: BTreeSet<String>,
}

pub async fn run_worktree_watch_start(options: WorktreeWatchStartOptions) -> Result<(), String> {
    if options.target.parent.is_some() {
        let identities = resolve_target_identities(&options.target)?;
        if identities.is_empty() {
            println!("No git repositories found under parent folder.");
            return Ok(());
        }

        if options.detach {
            for identity in &identities {
                let path = spawn_detached_worktree_watch_for_identity(identity)?;
                println!("Started background worktree watcher for {}", path.display());
            }
            println!(
                "Started {} background worktree watcher(s).",
                identities.len()
            );
            return Ok(());
        }

        if options.once {
            for identity in &identities {
                let layout = prepare_spool_layout(identity)?;
                record_commit_snapshot(identity, &layout, None)?;
                println!("Recorded current git state for {}", identity.root.display());
            }
            return Ok(());
        }

        return Err(
            "Parent-folder watching starts one watcher per repository. Pass `--detach` to run them in the background, or `--once` to snapshot each repo."
                .to_string(),
        );
    }

    if options.detach {
        return spawn_detached_worktree_watch(options.target.repo.as_deref()).map(|path| {
            println!("Started background worktree watcher for {}", path.display());
        });
    }

    let identity = resolve_repo_identity(options.target.repo.as_deref())?;
    let layout = prepare_spool_layout(&identity)?;
    println!(
        "Watching {} and spooling mutations to {}",
        identity.root.display(),
        layout.root.display()
    );

    if options.once {
        record_commit_snapshot(&identity, &layout, None)?;
        return Ok(());
    }

    watch_foreground(identity, layout, options.sync_interval_seconds).await
}

pub async fn run_worktree_watch_sync(options: WorktreeWatchSyncOptions) -> Result<(), String> {
    let identities = resolve_target_identities(&options.target)?;
    let repo_count = identities.len();
    let mut plans = Vec::new();
    let mut total_files = 0usize;
    let mut total_records = 0usize;

    for identity in identities {
        let layout = prepare_spool_layout(&identity)?;
        let candidates = collect_spool_candidates(&layout, options.resync)?;
        total_files += candidates.len();
        total_records += candidates
            .iter()
            .map(|candidate| candidate.records.len())
            .sum::<usize>();
        if !candidates.is_empty() {
            plans.push((identity, candidates));
        }
    }

    if options.dry_run {
        println!(
            "Would sync {} record(s) from {} spool file(s) across {} repo(s).",
            total_records, total_files, repo_count
        );
        if options.resync {
            println!("Resync mode includes files already marked `.synced.*.jsonl`.");
        }
        return Ok(());
    }

    if plans.is_empty() {
        print_no_sync_candidates_message(&options.target, options.resync)?;
        return Ok(());
    }

    let upload_repo_count = plans.len();
    for (identity, candidates) in plans {
        sync_candidates(&identity, candidates, options.resync).await?;
    }
    println!(
        "Synced {} worktree mutation record(s) across {} repo(s).",
        total_records, upload_repo_count
    );
    Ok(())
}

pub async fn run_worktree_watch_status(options: WorktreeWatchStatusOptions) -> Result<(), String> {
    let identities = resolve_target_identities(&options.target)?;
    let mut payloads = Vec::new();
    let mut total_files = 0usize;
    let mut total_records = 0usize;

    for identity in identities {
        let status =
            worktree_watch_status_payload(&identity, options.records, options.record_limit)?;
        let stats = if options.stats {
            Some(generate_and_store_stats(
                &identity,
                options.stats_gap_minutes,
            )?)
        } else {
            None
        };
        let status = attach_optional_stats(status, stats)?;
        total_files += status
            .get("unsyncedFiles")
            .and_then(Value::as_u64)
            .unwrap_or(0) as usize;
        total_records += status
            .get("unsyncedRecords")
            .and_then(Value::as_u64)
            .unwrap_or(0) as usize;
        payloads.push(status);
    }

    if options.json {
        let payload = if options.target.parent.is_some() {
            json!({
                "repositories": payloads,
                "repositoryCount": payloads.len(),
                "unsyncedFiles": total_files,
                "unsyncedRecords": total_records,
            })
        } else {
            payloads.into_iter().next().unwrap_or_else(|| json!({}))
        };
        println!(
            "{}",
            serde_json::to_string_pretty(&payload)
                .map_err(|error| format!("Failed to render status JSON: {error}"))?
        );
    } else {
        for (index, payload) in payloads.iter().enumerate() {
            if index > 0 {
                println!();
            }
            println!(
                "repo: {}/{}",
                payload
                    .get("repositoryOwner")
                    .and_then(Value::as_str)
                    .unwrap_or("unknown"),
                payload
                    .get("repositoryName")
                    .and_then(Value::as_str)
                    .unwrap_or("repository")
            );
            println!(
                "branch: {}",
                payload
                    .get("branchName")
                    .and_then(Value::as_str)
                    .unwrap_or("unknown")
            );
            println!(
                "root: {}",
                payload
                    .get("repoRoot")
                    .and_then(Value::as_str)
                    .unwrap_or("")
            );
            println!(
                "spool: {}",
                payload
                    .get("spoolRoot")
                    .and_then(Value::as_str)
                    .unwrap_or("")
            );
            println!(
                "unsynced files: {}",
                payload
                    .get("unsyncedFiles")
                    .and_then(Value::as_u64)
                    .unwrap_or(0)
            );
            println!(
                "unsynced records: {}",
                payload
                    .get("unsyncedRecords")
                    .and_then(Value::as_u64)
                    .unwrap_or(0)
            );
            println!(
                "synced files: {}",
                payload
                    .get("syncedFiles")
                    .and_then(Value::as_u64)
                    .unwrap_or(0)
            );
            println!(
                "synced records: {}",
                payload
                    .get("syncedRecords")
                    .and_then(Value::as_u64)
                    .unwrap_or(0)
            );
            print_last_sync(payload);
            if options.records {
                print_status_records(payload);
            }
            if options.stats {
                print_status_stats(payload);
            }
        }
        if options.target.parent.is_some() {
            println!();
            println!(
                "total: {} repo(s), {} unsynced file(s), {} unsynced record(s)",
                payloads.len(),
                total_files,
                total_records
            );
        }
    }

    Ok(())
}

fn attach_optional_stats(
    mut payload: Value,
    stats: Option<WorktreeWatchStatsSummary>,
) -> Result<Value, String> {
    if let Some(stats) = stats {
        if let Some(object) = payload.as_object_mut() {
            object.insert(
                "stats".to_string(),
                serde_json::to_value(stats)
                    .map_err(|error| format!("Failed to render stats payload: {error}"))?,
            );
        }
    }
    Ok(payload)
}

pub fn spawn_detached_worktree_watch_for_current_repo() -> Result<Option<PathBuf>, String> {
    if is_background_child() {
        return Ok(None);
    }

    let identity = match resolve_repo_identity(None) {
        Ok(identity) => identity,
        Err(_) => return Ok(None),
    };
    spawn_detached_worktree_watch_for_identity(&identity).map(Some)
}

fn resolve_target_identities(
    target: &WorktreeWatchTargetOptions,
) -> Result<Vec<RepoIdentity>, String> {
    if target.repo.is_some() && target.parent.is_some() {
        return Err("Pass either `--repo` or `--parent`, not both.".to_string());
    }

    if let Some(parent) = target.parent.as_deref() {
        return discover_repo_identities_under(parent);
    }

    resolve_repo_identity(target.repo.as_deref()).map(|identity| vec![identity])
}

fn worktree_watch_status_payload(
    identity: &RepoIdentity,
    include_records: bool,
    record_limit: usize,
) -> Result<Value, String> {
    let layout = prepare_spool_layout(identity)?;
    let candidates = collect_sync_candidates(&layout)?;
    let all_candidates = collect_spool_candidates(&layout, true)?;
    let record_count: usize = candidates
        .iter()
        .map(|candidate| candidate.records.len())
        .sum();
    let all_record_count: usize = all_candidates
        .iter()
        .map(|candidate| candidate.records.len())
        .sum();
    let synced_file_count = all_candidates.len().saturating_sub(candidates.len());
    let synced_record_count = all_record_count.saturating_sub(record_count);

    let mut payload = json!({
        "repoRoot": identity.root.display().to_string(),
        "repositoryOwner": identity.owner.clone(),
        "repositoryName": identity.name.clone(),
        "branchName": identity.branch.clone(),
        "spoolRoot": layout.root.display().to_string(),
        "unsyncedFiles": candidates.len(),
        "unsyncedRecords": record_count,
        "syncedFiles": synced_file_count,
        "syncedRecords": synced_record_count,
        "localSpoolFiles": all_candidates.len(),
        "localSpoolRecords": all_record_count,
    });

    if let Some(last_sync) = read_last_sync_log_entry(&layout.sync_log_file)? {
        if let Some(object) = payload.as_object_mut() {
            object.insert(
                "lastSync".to_string(),
                serde_json::to_value(last_sync)
                    .map_err(|error| format!("Failed to render sync log payload: {error}"))?,
            );
        }
    }

    if include_records {
        if let Some(object) = payload.as_object_mut() {
            object.insert(
                "records".to_string(),
                collect_status_records(&candidates, record_limit)?,
            );
        }
    }

    Ok(payload)
}

fn collect_status_records(candidates: &[SyncCandidate], limit: usize) -> Result<Value, String> {
    let available: usize = candidates
        .iter()
        .map(|candidate| candidate.records.len())
        .sum();
    let mut shown = 0usize;
    let mut events = Vec::new();
    let mut commits = Vec::new();

    'candidates: for candidate in candidates {
        for record in &candidate.records {
            if shown >= limit {
                break 'candidates;
            }

            match candidate.kind {
                SyncFileKind::Events => {
                    let event = serde_json::from_value::<WorktreeMutationEvent>(record.clone())
                        .map_err(|error| {
                            format!(
                                "Failed to decode event from {}: {error}",
                                candidate.path.display()
                            )
                        })?;
                    events.push(json!({
                        "sourceFile": candidate.path.display().to_string(),
                        "occurredAt": event.occurred_at,
                        "eventKind": event.event_kind,
                        "primaryPath": event.primary_path,
                        "oldPath": event.old_path,
                        "newPath": event.new_path,
                        "paths": event.paths,
                        "addedLines": event.added_lines,
                        "removedLines": event.removed_lines,
                        "headSha": event.head_sha,
                        "rawKind": event.raw_kind,
                    }));
                }
                SyncFileKind::Commits => {
                    let commit = serde_json::from_value::<WorktreeCommitEvent>(record.clone())
                        .map_err(|error| {
                            format!(
                                "Failed to decode commit from {}: {error}",
                                candidate.path.display()
                            )
                        })?;
                    commits.push(json!({
                        "sourceFile": candidate.path.display().to_string(),
                        "occurredAt": commit.occurred_at,
                        "headSha": commit.head_sha,
                        "previousHeadSha": commit.previous_head_sha,
                        "subject": commit.subject,
                        "authorName": commit.author_name,
                        "authorEmail": commit.author_email,
                        "committedAt": commit.committed_at,
                    }));
                }
            }

            shown += 1;
        }
    }

    Ok(json!({
        "available": available,
        "shown": shown,
        "truncated": shown < available,
        "events": events,
        "commits": commits,
    }))
}

fn print_status_records(payload: &Value) {
    let Some(records) = payload.get("records") else {
        return;
    };
    let shown = records.get("shown").and_then(Value::as_u64).unwrap_or(0);
    if shown == 0 {
        println!("records: none");
        return;
    }

    println!("records:");
    if let Some(events) = records.get("events").and_then(Value::as_array) {
        if !events.is_empty() {
            println!("  mutations:");
            for event in events {
                print_status_event_record(event);
            }
        }
    }
    if let Some(commits) = records.get("commits").and_then(Value::as_array) {
        if !commits.is_empty() {
            println!("  commits:");
            for commit in commits {
                print_status_commit_record(commit);
            }
        }
    }
    if records
        .get("truncated")
        .and_then(Value::as_bool)
        .unwrap_or(false)
    {
        let available = records
            .get("available")
            .and_then(Value::as_u64)
            .unwrap_or(shown);
        println!("  showing {shown} of {available} record(s)");
    }
}

fn print_status_stats(payload: &Value) {
    let Some(stats) = payload.get("stats") else {
        return;
    };
    println!("stats:");
    println!(
        "  coding time: {}",
        format_duration(
            stats
                .get("estimatedCodingSeconds")
                .and_then(Value::as_u64)
                .unwrap_or(0)
        )
    );
    println!(
        "  events: {}",
        stats
            .get("totalEvents")
            .and_then(Value::as_u64)
            .unwrap_or(0)
    );
    println!(
        "  commits: {}",
        stats
            .get("totalCommits")
            .and_then(Value::as_u64)
            .unwrap_or(0)
    );
    println!(
        "  lines: +{} -{}",
        stats.get("addedLines").and_then(Value::as_u64).unwrap_or(0),
        stats
            .get("removedLines")
            .and_then(Value::as_u64)
            .unwrap_or(0)
    );
    if let Some(spool_root) = value_str(stats, "spoolRoot") {
        println!(
            "  stored: {}",
            Path::new(spool_root).join("stats.json").display()
        );
    }
    print_stats_bucket_section(stats, "byFiletype", "  by filetype:", 10);
    print_stats_bucket_section(stats, "byEventKind", "  by event kind:", 10);
    print_stats_file_section(stats, 10);
}

fn print_last_sync(payload: &Value) {
    let Some(last_sync) = payload.get("lastSync") else {
        return;
    };
    let synced_at = value_str(last_sync, "syncedAt").unwrap_or("unknown-time");
    let status = last_sync
        .get("statusCode")
        .and_then(Value::as_u64)
        .unwrap_or(0);
    let events = last_sync
        .get("eventCount")
        .and_then(Value::as_u64)
        .unwrap_or(0);
    let commits = last_sync
        .get("commitCount")
        .and_then(Value::as_u64)
        .unwrap_or(0);
    let files = last_sync
        .get("spoolFileCount")
        .and_then(Value::as_u64)
        .unwrap_or(0);
    let resync = last_sync
        .get("resync")
        .and_then(Value::as_bool)
        .unwrap_or(false);
    println!(
        "last sync: {synced_at}, HTTP {status}, {events} event(s), {commits} commit(s), {files} file(s), resync={resync}"
    );
}

fn print_stats_bucket_section(stats: &Value, key: &str, title: &str, limit: usize) {
    let Some(items) = stats.get(key).and_then(Value::as_array) else {
        return;
    };
    if items.is_empty() {
        return;
    }
    println!("{title}");
    for item in items.iter().take(limit) {
        let name = value_str(item, "name").unwrap_or("unknown");
        let events = item.get("eventCount").and_then(Value::as_u64).unwrap_or(0);
        let seconds = item
            .get("estimatedCodingSeconds")
            .and_then(Value::as_u64)
            .unwrap_or(0);
        let added = item.get("addedLines").and_then(Value::as_u64).unwrap_or(0);
        let removed = item
            .get("removedLines")
            .and_then(Value::as_u64)
            .unwrap_or(0);
        println!(
            "    {name}: {}, {} event(s), +{} -{}",
            format_duration(seconds),
            events,
            added,
            removed
        );
    }
}

fn print_stats_file_section(stats: &Value, limit: usize) {
    let Some(items) = stats.get("byFile").and_then(Value::as_array) else {
        return;
    };
    if items.is_empty() {
        return;
    }
    println!("  by file:");
    for item in items.iter().take(limit) {
        let path = value_str(item, "path").unwrap_or("(no path)");
        let filetype = value_str(item, "filetype").unwrap_or("(none)");
        let events = item.get("eventCount").and_then(Value::as_u64).unwrap_or(0);
        let seconds = item
            .get("estimatedCodingSeconds")
            .and_then(Value::as_u64)
            .unwrap_or(0);
        let added = item.get("addedLines").and_then(Value::as_u64).unwrap_or(0);
        let removed = item
            .get("removedLines")
            .and_then(Value::as_u64)
            .unwrap_or(0);
        println!(
            "    {path} [{filetype}]: {}, {} event(s), +{} -{}",
            format_duration(seconds),
            events,
            added,
            removed
        );
    }
}

fn print_status_event_record(event: &Value) {
    let occurred_at = value_str(event, "occurredAt").unwrap_or("unknown-time");
    let kind = value_str(event, "eventKind").unwrap_or("event");
    let primary_path = event_display_path(event);
    let added = event.get("addedLines").and_then(Value::as_u64).unwrap_or(0);
    let removed = event
        .get("removedLines")
        .and_then(Value::as_u64)
        .unwrap_or(0);
    println!("    {occurred_at} {kind} {primary_path} (+{added} -{removed})");

    if let Some(head_sha) = value_str(event, "headSha") {
        println!("      head: {}", short_sha(head_sha));
    }
    if let Some(paths) = event.get("paths").and_then(Value::as_array) {
        if paths.len() > 1 {
            let rendered = paths
                .iter()
                .filter_map(Value::as_str)
                .collect::<Vec<_>>()
                .join(", ");
            println!("      paths: {rendered}");
        }
    }
}

fn print_status_commit_record(commit: &Value) {
    let occurred_at = value_str(commit, "occurredAt").unwrap_or("unknown-time");
    let head_sha = value_str(commit, "headSha")
        .map(short_sha)
        .unwrap_or_else(|| "unknown".to_string());
    let subject = value_str(commit, "subject").unwrap_or("(no subject)");
    println!("    {occurred_at} {head_sha} {subject}");

    if let Some(author) = value_str(commit, "authorName") {
        println!("      author: {author}");
    }
}

fn event_display_path(event: &Value) -> String {
    let old_path = value_str(event, "oldPath");
    let new_path = value_str(event, "newPath");
    if let (Some(old_path), Some(new_path)) = (old_path, new_path) {
        return format!("{old_path} -> {new_path}");
    }
    value_str(event, "primaryPath")
        .map(str::to_string)
        .or_else(|| {
            event
                .get("paths")
                .and_then(Value::as_array)
                .and_then(|paths| paths.first())
                .and_then(Value::as_str)
                .map(str::to_string)
        })
        .unwrap_or_else(|| "(no path)".to_string())
}

fn value_str<'a>(value: &'a Value, key: &str) -> Option<&'a str> {
    value.get(key).and_then(Value::as_str)
}

fn short_sha(value: &str) -> String {
    value.chars().take(12).collect()
}

fn generate_and_store_stats(
    identity: &RepoIdentity,
    session_gap_minutes: u64,
) -> Result<WorktreeWatchStatsSummary, String> {
    let layout = prepare_spool_layout(identity)?;
    let candidates = collect_spool_candidates(&layout, true)?;
    let summary = build_stats_summary(identity, &layout, &candidates, session_gap_minutes)?;
    let rendered = serde_json::to_string_pretty(&summary)
        .map_err(|error| format!("Failed to serialize stats: {error}"))?;
    fs::write(&layout.stats_file, format!("{rendered}\n")).map_err(|error| {
        format!(
            "Failed to write stats file {}: {error}",
            layout.stats_file.display()
        )
    })?;
    Ok(summary)
}

fn build_stats_summary(
    identity: &RepoIdentity,
    layout: &SpoolLayout,
    candidates: &[SyncCandidate],
    session_gap_minutes: u64,
) -> Result<WorktreeWatchStatsSummary, String> {
    let mut events = Vec::new();
    let mut total_commits = 0u64;
    for candidate in candidates {
        match candidate.kind {
            SyncFileKind::Events => {
                for record in &candidate.records {
                    events.push(
                        serde_json::from_value::<WorktreeMutationEvent>(record.clone()).map_err(
                            |error| {
                                format!(
                                    "Failed to decode event from {}: {error}",
                                    candidate.path.display()
                                )
                            },
                        )?,
                    );
                }
            }
            SyncFileKind::Commits => {
                total_commits += candidate.records.len() as u64;
            }
        }
    }
    events.sort_by_key(|event| event.occurred_at);

    let gap_seconds = session_gap_minutes.saturating_mul(60).max(60);
    let mut previous_at: Option<DateTime<Utc>> = None;
    let mut total_seconds = 0u64;
    let mut total_added = 0u64;
    let mut total_removed = 0u64;
    let mut by_file: BTreeMap<String, WorktreeWatchFileStats> = BTreeMap::new();
    let mut by_filetype: BTreeMap<String, WorktreeWatchBucketStats> = BTreeMap::new();
    let mut by_event_kind: BTreeMap<String, WorktreeWatchBucketStats> = BTreeMap::new();

    for event in &events {
        let seconds = coding_seconds_for_event(previous_at, event.occurred_at, gap_seconds);
        previous_at = Some(event.occurred_at);
        let added = event.added_lines.unwrap_or(0);
        let removed = event.removed_lines.unwrap_or(0);
        let path = stats_event_path(event);
        let filetype = filetype_for_path(&path);

        total_seconds += seconds;
        total_added += added;
        total_removed += removed;
        update_file_stats(&mut by_file, &path, &filetype, seconds, added, removed);
        update_bucket_stats(&mut by_filetype, &filetype, seconds, added, removed);
        update_bucket_stats(
            &mut by_event_kind,
            &event.event_kind,
            seconds,
            added,
            removed,
        );
    }

    Ok(WorktreeWatchStatsSummary {
        generated_at: Utc::now(),
        repository_owner: identity.owner.clone(),
        repository_name: identity.name.clone(),
        branch_name: identity.branch.clone(),
        repo_root: identity.root.display().to_string(),
        spool_root: layout.root.display().to_string(),
        session_gap_minutes,
        total_events: events.len() as u64,
        total_commits,
        first_event_at: events.first().map(|event| event.occurred_at),
        last_event_at: events.last().map(|event| event.occurred_at),
        estimated_coding_seconds: total_seconds,
        added_lines: total_added,
        removed_lines: total_removed,
        by_file: sorted_file_stats(by_file),
        by_filetype: sorted_bucket_stats(by_filetype),
        by_event_kind: sorted_bucket_stats(by_event_kind),
    })
}

fn coding_seconds_for_event(
    previous_at: Option<DateTime<Utc>>,
    occurred_at: DateTime<Utc>,
    gap_seconds: u64,
) -> u64 {
    let Some(previous_at) = previous_at else {
        return 60;
    };
    let delta = occurred_at.signed_duration_since(previous_at).num_seconds();
    if delta <= 0 {
        0
    } else {
        (delta as u64).min(gap_seconds)
    }
}

fn update_file_stats(
    by_file: &mut BTreeMap<String, WorktreeWatchFileStats>,
    path: &str,
    filetype: &str,
    seconds: u64,
    added: u64,
    removed: u64,
) {
    let stats = by_file
        .entry(path.to_string())
        .or_insert_with(|| WorktreeWatchFileStats {
            path: path.to_string(),
            filetype: filetype.to_string(),
            ..WorktreeWatchFileStats::default()
        });
    stats.event_count += 1;
    stats.estimated_coding_seconds += seconds;
    stats.added_lines += added;
    stats.removed_lines += removed;
}

fn update_bucket_stats(
    buckets: &mut BTreeMap<String, WorktreeWatchBucketStats>,
    name: &str,
    seconds: u64,
    added: u64,
    removed: u64,
) {
    let stats = buckets
        .entry(name.to_string())
        .or_insert_with(|| WorktreeWatchBucketStats {
            name: name.to_string(),
            ..WorktreeWatchBucketStats::default()
        });
    stats.event_count += 1;
    stats.estimated_coding_seconds += seconds;
    stats.added_lines += added;
    stats.removed_lines += removed;
}

fn sorted_file_stats(
    by_file: BTreeMap<String, WorktreeWatchFileStats>,
) -> Vec<WorktreeWatchFileStats> {
    let mut stats = by_file.into_values().collect::<Vec<_>>();
    stats.sort_by(|left, right| {
        right
            .estimated_coding_seconds
            .cmp(&left.estimated_coding_seconds)
            .then_with(|| right.event_count.cmp(&left.event_count))
            .then_with(|| left.path.cmp(&right.path))
    });
    stats
}

fn sorted_bucket_stats(
    buckets: BTreeMap<String, WorktreeWatchBucketStats>,
) -> Vec<WorktreeWatchBucketStats> {
    let mut stats = buckets.into_values().collect::<Vec<_>>();
    stats.sort_by(|left, right| {
        right
            .estimated_coding_seconds
            .cmp(&left.estimated_coding_seconds)
            .then_with(|| right.event_count.cmp(&left.event_count))
            .then_with(|| left.name.cmp(&right.name))
    });
    stats
}

fn stats_event_path(event: &WorktreeMutationEvent) -> String {
    event
        .new_path
        .as_deref()
        .or(event.primary_path.as_deref())
        .or_else(|| event.paths.first().map(String::as_str))
        .unwrap_or("(no path)")
        .to_string()
}

fn filetype_for_path(path: &str) -> String {
    Path::new(path)
        .extension()
        .and_then(|value| value.to_str())
        .map(|value| value.to_ascii_lowercase())
        .filter(|value| !value.is_empty())
        .unwrap_or_else(|| "(none)".to_string())
}

fn format_duration(seconds: u64) -> String {
    let hours = seconds / 3600;
    let minutes = (seconds % 3600) / 60;
    let remaining_seconds = seconds % 60;
    if hours > 0 {
        format!("{hours}h {minutes}m")
    } else if minutes > 0 {
        format!("{minutes}m {remaining_seconds}s")
    } else {
        format!("{remaining_seconds}s")
    }
}

fn watch_foreground(
    identity: RepoIdentity,
    layout: SpoolLayout,
    sync_interval_seconds: u64,
) -> impl std::future::Future<Output = Result<(), String>> {
    async move {
        let (tx, rx) = mpsc::channel();
        let mut watcher = RecommendedWatcher::new(
            move |result| {
                let _ = tx.send(result);
            },
            Config::default(),
        )
        .map_err(|error| format!("Failed to create filesystem watcher: {error}"))?;

        watcher
            .watch(&identity.root, RecursiveMode::Recursive)
            .map_err(|error| {
                format!(
                    "Failed to watch repository root {}: {error}",
                    identity.root.display()
                )
            })?;

        let mut last_head = identity.head_sha.clone();
        let mut last_commit_check = Instant::now();
        let mut last_sync = Instant::now();
        let mut event_dedupe = RecentEventDedupe {
            fingerprints: load_event_dedupe_fingerprints(&layout.event_dedupe_file)?,
        };

        loop {
            match rx.recv_timeout(Duration::from_millis(750)) {
                Ok(Ok(event)) => {
                    if let Some(mutation) = mutation_event_from_notify(&identity, event) {
                        append_unique_mutation_event(&layout, &mutation, &mut event_dedupe)?;
                    }
                }
                Ok(Err(error)) => {
                    eprintln!("worktree watcher error: {error}");
                }
                Err(mpsc::RecvTimeoutError::Timeout) => {}
                Err(mpsc::RecvTimeoutError::Disconnected) => {
                    return Err("Filesystem watcher disconnected.".to_string());
                }
            }

            if last_commit_check.elapsed() >= Duration::from_secs(3) {
                last_head = record_commit_snapshot(&identity, &layout, last_head.as_deref())?;
                last_commit_check = Instant::now();
            }

            if sync_interval_seconds > 0
                && last_sync.elapsed() >= Duration::from_secs(sync_interval_seconds)
            {
                if let Err(error) = sync_spool_for_identity(&identity, &layout).await {
                    eprintln!("worktree mutation sync failed: {error}");
                }
                last_sync = Instant::now();
            }
        }
    }
}

fn mutation_event_from_notify(
    identity: &RepoIdentity,
    event: Event,
) -> Option<WorktreeMutationEvent> {
    if event.kind.is_access() || event.kind.is_other() {
        return None;
    }

    let paths: Vec<String> = event
        .paths
        .iter()
        .filter(|path| !is_ignored_path(&identity.root, path))
        .filter_map(|path| relative_slash_path(&identity.root, path))
        .collect();

    if paths.is_empty() {
        return None;
    }

    let primary_path = paths.first().cloned();
    let (old_path, new_path) = rename_paths(&event.kind, &paths);
    let line_counts = primary_path
        .as_deref()
        .and_then(|path| git_numstat_for_path(&identity.root, path).ok());
    let event_kind = classify_event_kind(&event.kind);
    let is_dir = event
        .paths
        .first()
        .and_then(|path| fs::metadata(path).ok())
        .map(|metadata| metadata.is_dir())
        .unwrap_or(false);

    Some(WorktreeMutationEvent {
        id: Uuid::new_v4().to_string(),
        repo_owner: identity.owner.clone(),
        repo_name: identity.name.clone(),
        branch_name: identity.branch.clone(),
        repo_root: identity.root.display().to_string(),
        head_sha: current_head(&identity.root).ok().flatten(),
        event_kind: event_kind.to_string(),
        paths,
        primary_path,
        old_path,
        new_path,
        added_lines: line_counts.map(|counts| counts.0),
        removed_lines: line_counts.map(|counts| counts.1),
        file_created: matches!(
            event.kind,
            EventKind::Create(CreateKind::File | CreateKind::Any)
        ) && !is_dir,
        file_removed: matches!(
            event.kind,
            EventKind::Remove(RemoveKind::File | RemoveKind::Any)
        ) && !is_dir,
        folder_created: matches!(
            event.kind,
            EventKind::Create(CreateKind::Folder | CreateKind::Any)
        ) && is_dir,
        folder_removed: matches!(
            event.kind,
            EventKind::Remove(RemoveKind::Folder | RemoveKind::Any)
        ) && is_dir,
        renamed_or_moved: is_rename_or_move(&event.kind),
        raw_kind: format!("{:?}", event.kind),
        occurred_at: Utc::now(),
    })
}

fn append_unique_mutation_event(
    layout: &SpoolLayout,
    mutation: &WorktreeMutationEvent,
    dedupe: &mut RecentEventDedupe,
) -> Result<bool, String> {
    let fingerprint = mutation_event_fingerprint(mutation);
    if dedupe.fingerprints.contains(&fingerprint)
        || event_dedupe_file_contains(&layout.event_dedupe_file, &fingerprint)?
    {
        dedupe.fingerprints.insert(fingerprint);
        return Ok(false);
    }

    append_json_line(&layout.events_file, mutation)?;
    append_json_line(
        &layout.event_dedupe_file,
        &WorktreeWatchEventDedupeRecord {
            fingerprint: fingerprint.clone(),
            first_seen_at: Utc::now(),
            event_kind: mutation.event_kind.clone(),
            primary_path: mutation.primary_path.clone(),
        },
    )?;
    dedupe.fingerprints.insert(fingerprint);
    Ok(true)
}

fn mutation_event_fingerprint(mutation: &WorktreeMutationEvent) -> String {
    let bucket = mutation.occurred_at.timestamp() / EVENT_DEDUPE_WINDOW_SECONDS;
    let payload = json!({
        "repoOwner": mutation.repo_owner,
        "repoName": mutation.repo_name,
        "branchName": mutation.branch_name,
        "repoRoot": mutation.repo_root,
        "headSha": mutation.head_sha,
        "eventKind": mutation.event_kind,
        "paths": mutation.paths,
        "primaryPath": mutation.primary_path,
        "oldPath": mutation.old_path,
        "newPath": mutation.new_path,
        "addedLines": mutation.added_lines,
        "removedLines": mutation.removed_lines,
        "fileCreated": mutation.file_created,
        "fileRemoved": mutation.file_removed,
        "folderCreated": mutation.folder_created,
        "folderRemoved": mutation.folder_removed,
        "renamedOrMoved": mutation.renamed_or_moved,
        "timeBucket": bucket,
    });
    let encoded = serde_json::to_vec(&payload).unwrap_or_default();
    let mut hasher = Sha256::new();
    hasher.update(encoded);
    format!("{:x}", hasher.finalize())
}

fn load_event_dedupe_fingerprints(path: &Path) -> Result<BTreeSet<String>, String> {
    let mut fingerprints = BTreeSet::new();
    if !path.exists() {
        return Ok(fingerprints);
    }

    for value in read_jsonl_values(path)? {
        if let Some(fingerprint) = value.get("fingerprint").and_then(Value::as_str) {
            fingerprints.insert(fingerprint.to_string());
        }
    }
    Ok(fingerprints)
}

fn event_dedupe_file_contains(path: &Path, fingerprint: &str) -> Result<bool, String> {
    if !path.exists() {
        return Ok(false);
    }
    for value in read_jsonl_values(path)? {
        if value
            .get("fingerprint")
            .and_then(Value::as_str)
            .map(|value| value == fingerprint)
            .unwrap_or(false)
        {
            return Ok(true);
        }
    }
    Ok(false)
}

fn classify_event_kind(kind: &EventKind) -> &'static str {
    match kind {
        EventKind::Create(CreateKind::File) => "file_create",
        EventKind::Create(CreateKind::Folder) => "folder_create",
        EventKind::Create(_) => "create",
        EventKind::Remove(RemoveKind::File) => "file_remove",
        EventKind::Remove(RemoveKind::Folder) => "folder_remove",
        EventKind::Remove(_) => "remove",
        EventKind::Modify(ModifyKind::Name(
            RenameMode::From | RenameMode::To | RenameMode::Both,
        )) => "rename_or_move",
        EventKind::Modify(ModifyKind::Data(_)) => "file_modify",
        EventKind::Modify(ModifyKind::Metadata(_)) => "metadata_modify",
        EventKind::Modify(_) => "modify",
        _ => "other",
    }
}

fn is_rename_or_move(kind: &EventKind) -> bool {
    matches!(
        kind,
        EventKind::Modify(ModifyKind::Name(
            RenameMode::From | RenameMode::To | RenameMode::Both
        ))
    )
}

fn rename_paths(kind: &EventKind, paths: &[String]) -> (Option<String>, Option<String>) {
    if !is_rename_or_move(kind) {
        return (None, None);
    }

    match paths {
        [old_path, new_path, ..] => (Some(old_path.clone()), Some(new_path.clone())),
        [path] => match kind {
            EventKind::Modify(ModifyKind::Name(RenameMode::From)) => (Some(path.clone()), None),
            EventKind::Modify(ModifyKind::Name(RenameMode::To)) => (None, Some(path.clone())),
            _ => (None, Some(path.clone())),
        },
        _ => (None, None),
    }
}

fn record_commit_snapshot(
    identity: &RepoIdentity,
    layout: &SpoolLayout,
    previous_head: Option<&str>,
) -> Result<Option<String>, String> {
    let head = current_head(&identity.root)?;
    let Some(head_sha) = head else {
        return Ok(None);
    };

    if previous_head == Some(head_sha.as_str()) {
        return Ok(Some(head_sha));
    }

    let commit = WorktreeCommitEvent {
        id: Uuid::new_v4().to_string(),
        repo_owner: identity.owner.clone(),
        repo_name: identity.name.clone(),
        branch_name: identity.branch.clone(),
        repo_root: identity.root.display().to_string(),
        previous_head_sha: previous_head.map(str::to_string),
        head_sha: head_sha.clone(),
        subject: git_output(&identity.root, &["log", "-1", "--pretty=%s"]).ok(),
        author_name: git_output(&identity.root, &["log", "-1", "--pretty=%an"]).ok(),
        author_email: git_output(&identity.root, &["log", "-1", "--pretty=%ae"]).ok(),
        committed_at: git_output(&identity.root, &["log", "-1", "--pretty=%cI"]).ok(),
        occurred_at: Utc::now(),
    };
    append_json_line(&layout.commits_file, &commit)?;
    Ok(Some(head_sha))
}

async fn sync_spool_for_identity(
    identity: &RepoIdentity,
    layout: &SpoolLayout,
) -> Result<(), String> {
    let candidates = collect_sync_candidates(layout)?;
    if candidates.is_empty() {
        return Ok(());
    }

    sync_candidates(identity, candidates, false).await
}

async fn sync_candidates(
    identity: &RepoIdentity,
    candidates: Vec<SyncCandidate>,
    resync: bool,
) -> Result<(), String> {
    let mut events = Vec::new();
    let mut commits = Vec::new();
    for candidate in &candidates {
        match candidate.kind {
            SyncFileKind::Events => {
                for record in &candidate.records {
                    events.push(
                        serde_json::from_value::<WorktreeMutationEvent>(record.clone()).map_err(
                            |error| {
                                format!(
                                    "Failed to decode event from {}: {error}",
                                    candidate.path.display()
                                )
                            },
                        )?,
                    );
                }
            }
            SyncFileKind::Commits => {
                for record in &candidate.records {
                    commits.push(
                        serde_json::from_value::<WorktreeCommitEvent>(record.clone()).map_err(
                            |error| {
                                format!(
                                    "Failed to decode commit from {}: {error}",
                                    candidate.path.display()
                                )
                            },
                        )?,
                    );
                }
            }
        }
    }

    let token = resolve_cli_access_token()?;
    let device = resolve_device_identity()?;
    let client = cli_request_client()?;
    let api = ApiConfig::from_env();
    let endpoint = api.cli_worktree_mutations_endpoint();
    let event_count = events.len();
    let commit_count = commits.len();
    let payload = WorktreeMutationIngestPayload {
        device: WorktreeMutationDevicePayload {
            hardware_id: device.hardware_id,
            device_name: current_hostname(),
            hostname: current_hostname(),
            platform: std::env::consts::OS.to_string(),
        },
        repository: WorktreeMutationRepositoryPayload {
            owner: identity.owner.clone(),
            name: identity.name.clone(),
            branch_name: identity.branch.clone(),
            repo_root: identity.root.display().to_string(),
        },
        events,
        commits,
    };

    let response = client
        .post(endpoint.clone())
        .bearer_auth(token)
        .json(&payload)
        .send()
        .await
        .map_err(|error| format!("Failed to upload worktree mutations: {error}"))?;

    if response.status() == StatusCode::UNAUTHORIZED {
        return Err(
            "Your stored CLI session is no longer valid. Run `xbp login` again.".to_string(),
        );
    }

    if !response.status().is_success() {
        let status = response.status();
        let body = response.text().await.unwrap_or_default();
        return Err(format!(
            "Worktree mutation upload failed with {status}: {body}"
        ));
    }

    let status_code = response.status().as_u16();
    let layout = prepare_spool_layout(identity)?;
    append_sync_log(
        identity,
        &layout,
        &endpoint,
        status_code,
        event_count,
        commit_count,
        &candidates,
        resync,
    )?;

    for candidate in candidates {
        if !is_synced_spool_path(&candidate.path) {
            mark_synced(&candidate.path)?;
        }
    }

    Ok(())
}

fn collect_sync_candidates(layout: &SpoolLayout) -> Result<Vec<SyncCandidate>, String> {
    collect_spool_candidates(layout, false)
}

fn print_no_sync_candidates_message(
    target: &WorktreeWatchTargetOptions,
    resync: bool,
) -> Result<(), String> {
    if resync {
        println!("No local worktree mutation JSONL files found.");
        return Ok(());
    }

    let identities = resolve_target_identities(target)?;
    let mut synced_files = 0usize;
    let mut synced_records = 0usize;
    for identity in identities {
        let layout = prepare_spool_layout(&identity)?;
        let all_candidates = collect_spool_candidates(&layout, true)?;
        let unsynced_candidates = collect_sync_candidates(&layout)?;
        synced_files += all_candidates
            .len()
            .saturating_sub(unsynced_candidates.len());
        let all_records = all_candidates
            .iter()
            .map(|candidate| candidate.records.len())
            .sum::<usize>();
        let unsynced_records = unsynced_candidates
            .iter()
            .map(|candidate| candidate.records.len())
            .sum::<usize>();
        synced_records += all_records.saturating_sub(unsynced_records);
    }

    if synced_files > 0 {
        println!(
            "No unsynced worktree mutation files found. Found {synced_files} already-synced local spool file(s) with {synced_records} record(s). Use `xbp worktree-watch sync --resync` to upload them again."
        );
    } else {
        println!("No unsynced worktree mutation files found.");
    }
    Ok(())
}

fn collect_spool_candidates(
    layout: &SpoolLayout,
    include_synced: bool,
) -> Result<Vec<SyncCandidate>, String> {
    let mut candidates = Vec::new();
    if !layout.root.exists() {
        return Ok(candidates);
    }

    for entry in fs::read_dir(&layout.root).map_err(|error| {
        format!(
            "Failed to read spool directory {}: {error}",
            layout.root.display()
        )
    })? {
        let path = entry
            .map_err(|error| format!("Failed to read spool entry: {error}"))?
            .path();
        let Some(file_name) = path.file_name().and_then(|value| value.to_str()) else {
            continue;
        };
        if !file_name.ends_with(".jsonl") || (!include_synced && file_name.contains(".synced.")) {
            continue;
        }

        let kind = if file_name.starts_with("events-") {
            SyncFileKind::Events
        } else if file_name.starts_with("commits-") {
            SyncFileKind::Commits
        } else {
            continue;
        };
        let records = read_jsonl_values(&path)?;
        if records.is_empty() {
            continue;
        }
        candidates.push(SyncCandidate {
            path,
            kind,
            records,
        });
    }

    Ok(candidates)
}

fn append_sync_log(
    identity: &RepoIdentity,
    layout: &SpoolLayout,
    endpoint: &str,
    status_code: u16,
    event_count: usize,
    commit_count: usize,
    candidates: &[SyncCandidate],
    resync: bool,
) -> Result<(), String> {
    let entry = WorktreeWatchSyncLogEntry {
        id: Uuid::new_v4().to_string(),
        synced_at: Utc::now(),
        endpoint: endpoint.to_string(),
        repository_owner: identity.owner.clone(),
        repository_name: identity.name.clone(),
        branch_name: identity.branch.clone(),
        repo_root: identity.root.display().to_string(),
        resync,
        status_code,
        event_count,
        commit_count,
        spool_file_count: candidates.len(),
        spool_files: candidates
            .iter()
            .map(|candidate| candidate.path.display().to_string())
            .collect(),
    };
    append_json_line(&layout.sync_log_file, &entry)
}

fn read_last_sync_log_entry(path: &Path) -> Result<Option<WorktreeWatchSyncLogEntry>, String> {
    if !path.exists() {
        return Ok(None);
    }
    let values = read_jsonl_values(path)?;
    let Some(value) = values.into_iter().last() else {
        return Ok(None);
    };
    serde_json::from_value(value)
        .map(Some)
        .map_err(|error| format!("Failed to decode sync log {}: {error}", path.display()))
}

fn is_synced_spool_path(path: &Path) -> bool {
    path.file_name()
        .and_then(|value| value.to_str())
        .map(|value| value.contains(".synced."))
        .unwrap_or(false)
}

fn mark_synced(path: &Path) -> Result<(), String> {
    let Some(file_name) = path.file_name().and_then(|value| value.to_str()) else {
        return Ok(());
    };
    let synced_name = file_name.replace(
        ".jsonl",
        &format!(".synced.{}.jsonl", Utc::now().timestamp()),
    );
    let synced_path = path.with_file_name(synced_name);
    fs::rename(path, &synced_path).map_err(|error| {
        format!(
            "Failed to mark spool file {} as synced: {error}",
            path.display()
        )
    })
}

fn read_jsonl_values(path: &Path) -> Result<Vec<Value>, String> {
    let file = File::open(path)
        .map_err(|error| format!("Failed to open spool file {}: {error}", path.display()))?;
    let reader = BufReader::new(file);
    let mut values = Vec::new();
    for line in reader.lines() {
        let line =
            line.map_err(|error| format!("Failed to read spool file {}: {error}", path.display()))?;
        if line.trim().is_empty() {
            continue;
        }
        values.push(serde_json::from_str(&line).map_err(|error| {
            format!(
                "Failed to parse JSONL record in {}: {error}",
                path.display()
            )
        })?);
    }
    Ok(values)
}

fn spawn_detached_worktree_watch(repo: Option<&Path>) -> Result<PathBuf, String> {
    let identity = resolve_repo_identity(repo)?;
    spawn_detached_worktree_watch_for_identity(&identity)
}

fn spawn_detached_worktree_watch_for_identity(identity: &RepoIdentity) -> Result<PathBuf, String> {
    let layout = prepare_spool_layout(identity)?;
    replace_existing_watcher(identity, &layout)?;
    let executable = std::env::current_exe()
        .map_err(|error| format!("Failed to resolve current XBP executable: {error}"))?;
    let mut command = Command::new(&executable);
    command
        .arg("worktree-watch")
        .arg("start")
        .arg("--repo")
        .arg(&identity.root)
        .env(BACKGROUND_CHILD_ENV, "1")
        .current_dir(&identity.root)
        .stdin(Stdio::null())
        .stdout(Stdio::null())
        .stderr(Stdio::null());

    #[cfg(windows)]
    {
        use std::os::windows::process::CommandExt;
        command.creation_flags(CREATE_NO_WINDOW);
    }

    let child = command
        .spawn()
        .map_err(|error| format!("Failed to spawn background worktree watcher: {error}"))?;
    write_watcher_state(identity, &layout, child.id(), &executable)?;
    Ok(identity.root.clone())
}

fn replace_existing_watcher(identity: &RepoIdentity, layout: &SpoolLayout) -> Result<(), String> {
    let Some(state) = read_watcher_state(&layout.watcher_state_file)? else {
        return Ok(());
    };

    if !same_repo_watcher_state(identity, &state) {
        return Ok(());
    }

    if is_matching_watcher_process(&state) {
        stop_process(&state)?;
        println!(
            "Replaced existing background worktree watcher {} for {}",
            state.pid,
            identity.root.display()
        );
    }

    let _ = fs::remove_file(&layout.watcher_state_file);
    Ok(())
}

fn read_watcher_state(path: &Path) -> Result<Option<WorktreeWatcherState>, String> {
    if !path.exists() {
        return Ok(None);
    }
    let raw = fs::read_to_string(path)
        .map_err(|error| format!("Failed to read watcher state {}: {error}", path.display()))?;
    serde_json::from_str(&raw)
        .map(Some)
        .map_err(|error| format!("Failed to parse watcher state {}: {error}", path.display()))
}

fn write_watcher_state(
    identity: &RepoIdentity,
    layout: &SpoolLayout,
    pid: u32,
    executable: &Path,
) -> Result<(), String> {
    let state = WorktreeWatcherState {
        pid,
        repo_root: identity.root.display().to_string(),
        repository_owner: identity.owner.clone(),
        repository_name: identity.name.clone(),
        branch_name: identity.branch.clone(),
        executable: executable.display().to_string(),
        started_at: Utc::now(),
    };
    let rendered = serde_json::to_string_pretty(&state)
        .map_err(|error| format!("Failed to serialize watcher state: {error}"))?;
    fs::write(&layout.watcher_state_file, format!("{rendered}\n")).map_err(|error| {
        format!(
            "Failed to write watcher state {}: {error}",
            layout.watcher_state_file.display()
        )
    })
}

fn same_repo_watcher_state(identity: &RepoIdentity, state: &WorktreeWatcherState) -> bool {
    path_identity_key(&identity.root) == path_identity_key(Path::new(&state.repo_root))
        && identity.owner == state.repository_owner
        && identity.name == state.repository_name
        && identity.branch == state.branch_name
}

fn is_matching_watcher_process(state: &WorktreeWatcherState) -> bool {
    if state.pid == std::process::id() {
        return false;
    }
    let system = System::new_all();
    let Some(process) = system.process(Pid::from_u32(state.pid)) else {
        return false;
    };
    let command_line = process
        .cmd()
        .iter()
        .map(|part| part.to_string_lossy())
        .collect::<Vec<_>>()
        .join(" ");
    command_line.contains("worktree-watch")
        && command_line.contains("start")
        && command_line.contains(&state.repo_root)
}

fn stop_process(state: &WorktreeWatcherState) -> Result<(), String> {
    let system = System::new_all();
    let Some(process) = system.process(Pid::from_u32(state.pid)) else {
        return Ok(());
    };
    if !is_matching_watcher_process(state) {
        return Ok(());
    }
    if process.kill() {
        Ok(())
    } else {
        Err(format!(
            "Failed to stop existing watcher process {}",
            state.pid
        ))
    }
}

fn discover_repo_identities_under(parent: &Path) -> Result<Vec<RepoIdentity>, String> {
    let parent = normalize_windows_verbatim_path(
        fs::canonicalize(parent).unwrap_or_else(|_| parent.to_path_buf()),
    );
    if !parent.is_dir() {
        return Err(format!(
            "Parent folder {} does not exist or is not a directory.",
            parent.display()
        ));
    }

    let mut identities = Vec::new();
    let mut seen = BTreeSet::new();
    discover_repo_identities_in_dir(&parent, &parent, &mut seen, &mut identities)?;
    identities.sort_by(|left, right| left.root.cmp(&right.root));
    Ok(identities)
}

fn discover_repo_identities_in_dir(
    parent: &Path,
    dir: &Path,
    seen: &mut BTreeSet<String>,
    identities: &mut Vec<RepoIdentity>,
) -> Result<(), String> {
    if dir != parent && is_skipped_discovery_dir(dir) {
        return Ok(());
    }

    if dir.join(".git").exists() {
        if let Ok(identity) = resolve_repo_identity(Some(dir)) {
            let key = path_identity_key(&identity.root);
            if seen.insert(key) {
                identities.push(identity);
            }
            return Ok(());
        }
    }

    let entries = fs::read_dir(dir)
        .map_err(|error| format!("Failed to read folder {}: {error}", dir.display()))?;
    for entry in entries {
        let entry = entry.map_err(|error| {
            format!(
                "Failed to read folder entry under {}: {error}",
                dir.display()
            )
        })?;
        let file_type = entry
            .file_type()
            .map_err(|error| format!("Failed to inspect {}: {error}", entry.path().display()))?;
        if file_type.is_dir() {
            discover_repo_identities_in_dir(parent, &entry.path(), seen, identities)?;
        }
    }

    Ok(())
}

fn resolve_repo_identity(repo: Option<&Path>) -> Result<RepoIdentity, String> {
    let start = match repo {
        Some(path) => path.to_path_buf(),
        None => std::env::current_dir()
            .map_err(|error| format!("Failed to resolve current directory: {error}"))?,
    };
    let root_raw = git_output(&start, &["rev-parse", "--show-toplevel"])?;
    let root = normalize_windows_verbatim_path(
        fs::canonicalize(root_raw.trim()).unwrap_or_else(|_| PathBuf::from(root_raw.trim())),
    );
    let branch = git_output(&root, &["rev-parse", "--abbrev-ref", "HEAD"])
        .unwrap_or_else(|_| "unknown".to_string());
    let remote = git_output(&root, &["remote", "get-url", DEFAULT_REMOTE]).unwrap_or_default();
    let (owner, name) = parse_remote_owner_repo(&remote).unwrap_or_else(|| {
        let name = root
            .file_name()
            .and_then(|value| value.to_str())
            .unwrap_or("repository")
            .to_string();
        ("unknown".to_string(), name)
    });
    let head_sha = current_head(&root)?;

    Ok(RepoIdentity {
        owner,
        name,
        branch: sanitize_path_component(branch.trim()),
        root,
        head_sha,
    })
}

fn prepare_spool_layout(identity: &RepoIdentity) -> Result<SpoolLayout, String> {
    let home = dirs::home_dir().ok_or_else(|| "Failed to resolve home directory.".to_string())?;
    let run_id = Uuid::new_v4().to_string();
    let root = home
        .join(".xbp")
        .join("mutations")
        .join(sanitize_path_component(&identity.owner))
        .join(sanitize_path_component(&identity.name))
        .join(sanitize_path_component(&identity.branch));
    fs::create_dir_all(&root).map_err(|error| {
        format!(
            "Failed to create worktree mutation spool {}: {error}",
            root.display()
        )
    })?;
    Ok(SpoolLayout {
        events_file: root.join(format!("events-{run_id}.jsonl")),
        commits_file: root.join(format!("commits-{run_id}.jsonl")),
        stats_file: root.join("stats.json"),
        watcher_state_file: root.join("watcher-state.json"),
        sync_log_file: root.join("sync-log.jsonl"),
        event_dedupe_file: root.join("event-dedupe.jsonl"),
        root,
    })
}

fn append_json_line<T: Serialize>(path: &Path, value: &T) -> Result<(), String> {
    let mut file = OpenOptions::new()
        .create(true)
        .append(true)
        .open(path)
        .map_err(|error| format!("Failed to open spool file {}: {error}", path.display()))?;
    let line = serde_json::to_string(value)
        .map_err(|error| format!("Failed to serialize worktree event: {error}"))?;
    writeln!(file, "{line}")
        .map_err(|error| format!("Failed to append spool file {}: {error}", path.display()))
}

fn git_numstat_for_path(repo_root: &Path, relative_path: &str) -> Result<(u64, u64), String> {
    let output = Command::new("git")
        .args(["diff", "--numstat", "--"])
        .arg(relative_path)
        .current_dir(repo_root)
        .output()
        .map_err(|error| format!("Failed to run git diff --numstat: {error}"))?;
    if !output.status.success() {
        return Ok((0, 0));
    }
    let stdout = String::from_utf8_lossy(&output.stdout);
    let mut added = 0;
    let mut removed = 0;
    for line in stdout.lines() {
        let mut parts = line.split_whitespace();
        added += parse_numstat_count(parts.next());
        removed += parse_numstat_count(parts.next());
    }
    Ok((added, removed))
}

fn parse_numstat_count(value: Option<&str>) -> u64 {
    value.and_then(|raw| raw.parse::<u64>().ok()).unwrap_or(0)
}

fn current_head(repo_root: &Path) -> Result<Option<String>, String> {
    match git_output(repo_root, &["rev-parse", "HEAD"]) {
        Ok(value) => Ok(Some(value)),
        Err(error)
            if error.contains("unknown revision") || error.contains("ambiguous argument") =>
        {
            Ok(None)
        }
        Err(error) => Err(error),
    }
}

fn git_output(repo_root: &Path, args: &[&str]) -> Result<String, String> {
    let output = Command::new("git")
        .args(args)
        .current_dir(repo_root)
        .output()
        .map_err(|error| format!("Failed to run git {}: {error}", args.join(" ")))?;
    if !output.status.success() {
        return Err(String::from_utf8_lossy(&output.stderr).trim().to_string());
    }
    Ok(String::from_utf8_lossy(&output.stdout).trim().to_string())
}

fn parse_remote_owner_repo(remote: &str) -> Option<(String, String)> {
    let trimmed = remote.trim().trim_end_matches(".git");
    if trimmed.is_empty() {
        return None;
    }

    let path_part = if !trimmed.contains("://") {
        if let Some((_, path)) = trimmed.rsplit_once(':') {
            path
        } else {
            trimmed
        }
    } else {
        trimmed
            .trim_start_matches("https://")
            .trim_start_matches("http://")
            .trim_start_matches("ssh://")
            .split_once('/')
            .map(|(_, path)| path)
            .unwrap_or(trimmed)
    };
    let mut parts = path_part.rsplitn(2, '/');
    let name = parts.next()?.trim();
    let owner = parts.next()?.trim();
    if owner.is_empty() || name.is_empty() {
        return None;
    }
    Some((
        sanitize_path_component(owner),
        sanitize_path_component(name.trim_end_matches(".git")),
    ))
}

fn sanitize_path_component(value: &str) -> String {
    let sanitized: String = value
        .chars()
        .map(|ch| {
            if ch.is_ascii_alphanumeric() || matches!(ch, '-' | '_' | '.') {
                ch
            } else {
                '-'
            }
        })
        .collect();
    sanitized
        .trim_matches('-')
        .chars()
        .take(160)
        .collect::<String>()
}

fn normalize_windows_verbatim_path(path: PathBuf) -> PathBuf {
    PathBuf::from(strip_windows_verbatim_prefix(&path.to_string_lossy()))
}

fn strip_windows_verbatim_prefix(input: &str) -> &str {
    input.strip_prefix(r"\\?\").unwrap_or(input)
}

fn path_identity_key(path: &Path) -> String {
    let value = path.to_string_lossy().to_string();
    if cfg!(windows) {
        value.to_ascii_lowercase()
    } else {
        value
    }
}

fn is_skipped_discovery_dir(path: &Path) -> bool {
    let Some(name) = path.file_name().and_then(|value| value.to_str()) else {
        return false;
    };
    matches!(
        name,
        ".git"
            | ".hg"
            | ".svn"
            | ".next"
            | ".turbo"
            | ".vercel"
            | "node_modules"
            | "target"
            | "dist"
            | "build"
    )
}

fn relative_slash_path(root: &Path, path: &Path) -> Option<String> {
    let relative = path.strip_prefix(root).ok().unwrap_or(path);
    Some(relative.to_string_lossy().replace('\\', "/"))
}

fn is_ignored_path(root: &Path, path: &Path) -> bool {
    let Some(relative) = relative_slash_path(root, path) else {
        return false;
    };
    relative == ".git"
        || relative.starts_with(".git/")
        || relative == "target"
        || relative.starts_with("target/")
}

fn current_hostname() -> Option<String> {
    std::env::var("COMPUTERNAME")
        .or_else(|_| std::env::var("HOSTNAME"))
        .ok()
        .map(|value| value.trim().to_string())
        .filter(|value| !value.is_empty())
}

fn is_background_child() -> bool {
    std::env::var(BACKGROUND_CHILD_ENV)
        .ok()
        .map(|value| value == "1")
        .unwrap_or(false)
}

#[allow(dead_code)]
fn content_sha256(path: &Path) -> Option<String> {
    let bytes = fs::read(path).ok()?;
    let mut hasher = Sha256::new();
    hasher.update(bytes);
    Some(format!("{:x}", hasher.finalize()))
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn parses_https_and_ssh_remote_urls() {
        assert_eq!(
            parse_remote_owner_repo("https://github.com/xylex-group/xbp.git"),
            Some(("xylex-group".to_string(), "xbp".to_string()))
        );
        assert_eq!(
            parse_remote_owner_repo("git@github.com:xylex-group/xbp.git"),
            Some(("xylex-group".to_string(), "xbp".to_string()))
        );
    }

    #[test]
    fn sanitizes_branch_for_storage_path() {
        assert_eq!(
            sanitize_path_component("feature/worktree watcher"),
            "feature-worktree-watcher"
        );
    }

    #[test]
    fn normalizes_windows_verbatim_repo_roots() {
        assert_eq!(
            normalize_windows_verbatim_path(PathBuf::from(
                r"\\?\C:\Users\floris\Documents\GitHub\mollie-api-rust"
            )),
            PathBuf::from(r"C:\Users\floris\Documents\GitHub\mollie-api-rust")
        );
    }

    #[test]
    fn skips_heavy_discovery_directories() {
        assert!(is_skipped_discovery_dir(Path::new("node_modules")));
        assert!(is_skipped_discovery_dir(Path::new("target")));
        assert!(is_skipped_discovery_dir(Path::new(".next")));
        assert!(!is_skipped_discovery_dir(Path::new("mollie-api-rust")));
    }

    #[test]
    fn builds_coding_stats_by_file_and_filetype() {
        let identity = RepoIdentity {
            owner: "xylex-group".to_string(),
            name: "xbp".to_string(),
            branch: "main".to_string(),
            root: PathBuf::from(r"C:\Users\floris\Documents\GitHub\xbp"),
            head_sha: None,
        };
        let layout = SpoolLayout {
            root: PathBuf::from(r"C:\Users\floris\.xbp\mutations\xylex-group\xbp\main"),
            events_file: PathBuf::from("events.jsonl"),
            commits_file: PathBuf::from("commits.jsonl"),
            stats_file: PathBuf::from("stats.json"),
            watcher_state_file: PathBuf::from("watcher-state.json"),
            sync_log_file: PathBuf::from("sync-log.jsonl"),
            event_dedupe_file: PathBuf::from("event-dedupe.jsonl"),
        };
        let candidates = vec![SyncCandidate {
            path: PathBuf::from("events.jsonl"),
            kind: SyncFileKind::Events,
            records: vec![
                stats_test_event("2026-07-08T10:00:00Z", "src/main.rs", 4, 1),
                stats_test_event("2026-07-08T10:05:00Z", "src/main.rs", 2, 0),
                stats_test_event("2026-07-08T11:00:00Z", "README.md", 1, 3),
            ],
        }];

        let summary = build_stats_summary(&identity, &layout, &candidates, 15).unwrap();

        assert_eq!(summary.total_events, 3);
        assert_eq!(summary.estimated_coding_seconds, 60 + 300 + 900);
        assert_eq!(summary.added_lines, 7);
        assert_eq!(summary.removed_lines, 4);
        assert_eq!(summary.by_file[0].path, "README.md");
        assert_eq!(summary.by_file[0].estimated_coding_seconds, 900);
        assert_eq!(summary.by_filetype[0].name, "md");
        assert_eq!(summary.by_filetype[0].estimated_coding_seconds, 900);
        assert_eq!(summary.by_filetype[1].name, "rs");
        assert_eq!(summary.by_filetype[1].estimated_coding_seconds, 360);
    }

    #[test]
    fn mutation_fingerprint_ignores_random_event_id_but_keeps_time_bucket() {
        let first: WorktreeMutationEvent = serde_json::from_value(stats_test_event(
            "2026-07-08T10:00:00Z",
            "src/main.rs",
            4,
            1,
        ))
        .unwrap();
        let same_bucket: WorktreeMutationEvent = serde_json::from_value(stats_test_event(
            "2026-07-08T10:00:00Z",
            "src/main.rs",
            4,
            1,
        ))
        .unwrap();
        let next_bucket: WorktreeMutationEvent = serde_json::from_value(stats_test_event(
            "2026-07-08T10:00:06Z",
            "src/main.rs",
            4,
            1,
        ))
        .unwrap();

        assert_eq!(
            mutation_event_fingerprint(&first),
            mutation_event_fingerprint(&same_bucket)
        );
        assert_ne!(
            mutation_event_fingerprint(&first),
            mutation_event_fingerprint(&next_bucket)
        );
    }

    #[test]
    fn classifies_create_remove_and_rename_events() {
        assert_eq!(
            classify_event_kind(&EventKind::Create(CreateKind::File)),
            "file_create"
        );
        assert_eq!(
            classify_event_kind(&EventKind::Remove(RemoveKind::Folder)),
            "folder_remove"
        );
        assert_eq!(
            classify_event_kind(&EventKind::Modify(ModifyKind::Name(RenameMode::Both))),
            "rename_or_move"
        );
    }

    fn stats_test_event(
        occurred_at: &str,
        path: &str,
        added_lines: u64,
        removed_lines: u64,
    ) -> Value {
        json!({
            "id": Uuid::new_v4().to_string(),
            "repoOwner": "xylex-group",
            "repoName": "xbp",
            "branchName": "main",
            "repoRoot": r"C:\Users\floris\Documents\GitHub\xbp",
            "headSha": null,
            "eventKind": "file_modify",
            "paths": [path],
            "primaryPath": path,
            "oldPath": null,
            "newPath": null,
            "addedLines": added_lines,
            "removedLines": removed_lines,
            "fileCreated": false,
            "fileRemoved": false,
            "folderCreated": false,
            "folderRemoved": false,
            "renamedOrMoved": false,
            "rawKind": "Modify(Data(Content))",
            "occurredAt": occurred_at,
        })
    }
}