branchfs 0.1.1

use std::collections::HashMap;
use std::fs;
use std::io::{BufRead, BufReader, Write};
use std::os::unix::net::{UnixListener, UnixStream};
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use std::time::Duration;

use fuser::{BackgroundSession, MountOption};
use parking_lot::Mutex;
use serde::{Deserialize, Serialize};
use std::process::{Command, Stdio};

use crate::branch::BranchManager;
use crate::error::Result;
use crate::fs::BranchFs;

#[derive(Debug, Serialize, Deserialize)]
#[serde(tag = "cmd", rename_all = "snake_case")]
pub enum Request {
    Mount {
        branch: String,
        mountpoint: String,
        #[serde(default)]
        passthrough: bool,
    },
    Unmount {
        mountpoint: String,
    },
    Create {
        name: String,
        parent: String,
    },
    GetMountBranch {
        mountpoint: String,
    },
    List,
    Shutdown,
}

#[derive(Debug, Serialize, Deserialize)]
pub struct Response {
    pub ok: bool,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub error: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub data: Option<serde_json::Value>,
}

impl Response {
    pub fn success() -> Self {
        Self {
            ok: true,
            error: None,
            data: None,
        }
    }

    pub fn success_with_data(data: serde_json::Value) -> Self {
        Self {
            ok: true,
            error: None,
            data: Some(data),
        }
    }

    pub fn error(msg: &str) -> Self {
        Self {
            ok: false,
            error: Some(msg.to_string()),
            data: None,
        }
    }
}

/// Per-mount state (FUSE session handle — kept alive until unmount)
pub struct MountInfo {
    #[allow(dead_code)]
    session: BackgroundSession,
}

pub struct Daemon {
    manager: Arc<BranchManager>,
    mounts: Mutex<HashMap<PathBuf, MountInfo>>,
    socket_path: PathBuf,
    shutdown: AtomicBool,
}

impl Daemon {
    pub fn new(
        base_path: PathBuf,
        storage_path: PathBuf,
        _workspace_path: PathBuf,
        max_storage: Option<u64>,
    ) -> Result<Self> {
        let socket_path = storage_path.join("daemon.sock");

        // Clean up branches from previous daemon run for fresh state
        let branches_dir = storage_path.join("branches");
        if branches_dir.exists() {
            if let Err(e) = fs::remove_dir_all(&branches_dir) {
                log::warn!("Failed to clean up branches directory: {}", e);
            }
        }

        // Also clean up legacy mounts directory if present
        let mounts_dir = storage_path.join("mounts");
        if mounts_dir.exists() {
            if let Err(e) = fs::remove_dir_all(&mounts_dir) {
                log::warn!("Failed to clean up orphaned mounts directory: {}", e);
            }
        }

        // Store base_path for later use (simple file, not state.json)
        let base_file = storage_path.join("base_path");
        fs::create_dir_all(&storage_path)?;
        fs::write(&base_file, base_path.to_string_lossy().as_bytes())?;

        // Create the single shared BranchManager
        let manager = Arc::new(BranchManager::new(
            storage_path.clone(),
            base_path.clone(),
            base_path.clone(),
            max_storage,
        )?);

        Ok(Self {
            manager,
            mounts: Mutex::new(HashMap::new()),
            socket_path,
            shutdown: AtomicBool::new(false),
        })
    }

    pub fn socket_path(&self) -> &PathBuf {
        &self.socket_path
    }

    pub fn spawn_mount(
        &self,
        branch_name: &str,
        mountpoint: &Path,
        passthrough: bool,
    ) -> Result<()> {
        // Register the mount branch *before* creating BranchFs so get_branch_name() works
        self.manager.set_mount_branch(mountpoint, branch_name);

        let fs = BranchFs::new(self.manager.clone(), mountpoint.to_path_buf(), passthrough);
        let mut options = vec![MountOption::FSName("branchfs".to_string())];
        options.extend(crate::platform::get_mount_options());

        log::info!(
            "Spawning mount for branch '{}' at {:?}",
            branch_name,
            mountpoint,
        );

        let session = match fuser::spawn_mount2(fs, mountpoint, &options) {
            Ok(s) => s,
            Err(e) => {
                // Clean up on failure
                self.manager.unregister_mount(mountpoint);
                return Err(crate::error::BranchError::Io(e));
            }
        };

        // Get the notifier for cache invalidation and register it with the manager
        let notifier = Arc::new(session.notifier());
        self.manager
            .register_notifier(branch_name, mountpoint.to_path_buf(), notifier);

        let mount_info = MountInfo { session };

        self.mounts
            .lock()
            .insert(mountpoint.to_path_buf(), mount_info);

        Ok(())
    }

    pub fn unmount(&self, mountpoint: &Path) -> Result<()> {
        let should_shutdown = {
            let mut mounts = self.mounts.lock();
            if mounts.remove(mountpoint).is_some() {
                log::info!("Unmounted {:?}", mountpoint);
                // The BackgroundSession drop will handle FUSE cleanup
                mounts.is_empty()
            } else {
                return Err(crate::error::BranchError::MountNotFound(format!(
                    "{:?}",
                    mountpoint
                )));
            }
        };

        self.manager.unregister_mount(mountpoint);

        if should_shutdown {
            log::info!("All mounts removed, daemon will exit");
            self.shutdown.store(true, Ordering::SeqCst);
        }

        Ok(())
    }

    fn cleanup_all_mounts(&self) {
        let mut mounts = self.mounts.lock();
        let mountpoints: Vec<PathBuf> = mounts.keys().cloned().collect();
        for mountpoint in &mountpoints {
            if mounts.remove(mountpoint).is_some() {
                self.manager.unregister_mount(mountpoint);
                // BackgroundSession dropped here → FUSE unmount
                log::info!("Cleaned up mount at {:?}", mountpoint);
            }
        }
    }

    pub fn mount_count(&self) -> usize {
        self.mounts.lock().len()
    }

    pub fn create_branch(&self, name: &str, parent: &str) -> Result<()> {
        self.manager.create_branch(name, parent)
    }

    pub fn list_branches(&self) -> Vec<(String, Option<String>)> {
        self.manager.list_branches()
    }

    pub fn get_manager(&self) -> Arc<BranchManager> {
        self.manager.clone()
    }

    pub fn run(&self) -> Result<()> {
        if self.socket_path.exists() {
            std::fs::remove_file(&self.socket_path)?;
        }

        let listener =
            UnixListener::bind(&self.socket_path).map_err(crate::error::BranchError::Io)?;

        listener
            .set_nonblocking(true)
            .map_err(crate::error::BranchError::Io)?;

        log::info!("Daemon listening on {:?}", self.socket_path);

        loop {
            if self.shutdown.load(Ordering::SeqCst) {
                log::info!("Shutdown flag set, exiting");
                break;
            }

            match listener.accept() {
                Ok((stream, _)) => {
                    stream.set_nonblocking(false).ok();
                    if let Err(e) = self.handle_client(stream) {
                        log::error!("Client error: {}", e);
                    }
                }
                Err(ref e) if e.kind() == std::io::ErrorKind::WouldBlock => {
                    std::thread::sleep(Duration::from_millis(100));
                }
                Err(e) => {
                    log::error!("Accept error: {}", e);
                }
            }
        }

        if self.socket_path.exists() {
            std::fs::remove_file(&self.socket_path).ok();
        }

        Ok(())
    }

    fn handle_client(&self, mut stream: UnixStream) -> Result<()> {
        let reader = BufReader::new(stream.try_clone()?);

        for line in reader.lines() {
            let line = line?;
            let request: Request = match serde_json::from_str(&line) {
                Ok(req) => req,
                Err(e) => {
                    let resp = Response::error(&format!("Invalid request: {}", e));
                    writeln!(stream, "{}", serde_json::to_string(&resp)?)?;
                    continue;
                }
            };

            let response = self.handle_request(request);
            writeln!(stream, "{}", serde_json::to_string(&response)?)?;

            if self.shutdown.load(Ordering::SeqCst) {
                break;
            }
        }

        Ok(())
    }

    fn handle_request(&self, request: Request) -> Response {
        match request {
            Request::Mount {
                branch,
                mountpoint,
                passthrough,
            } => {
                let path = PathBuf::from(&mountpoint);
                if let Err(e) = fs::create_dir_all(&path) {
                    return Response::error(&format!("Failed to create mountpoint: {}", e));
                }
                match self.spawn_mount(&branch, &path, passthrough) {
                    Ok(()) => Response::success(),
                    Err(e) => Response::error(&format!("{}", e)),
                }
            }
            Request::Unmount { mountpoint } => {
                let path = PathBuf::from(&mountpoint);
                match self.unmount(&path) {
                    Ok(()) => Response::success(),
                    Err(e) => Response::error(&format!("{}", e)),
                }
            }
            Request::Create { name, parent } => match self.create_branch(&name, &parent) {
                Ok(()) => Response::success(),
                Err(e) => Response::error(&format!("{}", e)),
            },
            Request::GetMountBranch { mountpoint } => {
                let path = PathBuf::from(&mountpoint);
                if let Some(branch) = self.manager.get_mount_branch(&path) {
                    Response::success_with_data(serde_json::json!(branch))
                } else {
                    Response::error(&format!("Mount not found: {:?}", path))
                }
            }
            Request::List => {
                let branches: Vec<_> = self
                    .list_branches()
                    .into_iter()
                    .map(|(name, parent)| {
                        serde_json::json!({
                            "name": name,
                            "parent": parent
                        })
                    })
                    .collect();
                Response::success_with_data(serde_json::json!(branches))
            }
            Request::Shutdown => {
                log::info!("Shutdown requested, cleaning up all mounts");
                self.cleanup_all_mounts();
                self.shutdown.store(true, Ordering::SeqCst);
                Response::success()
            }
        }
    }
}

pub fn send_request(socket_path: &Path, request: &Request) -> std::io::Result<Response> {
    let mut stream = UnixStream::connect(socket_path)?;
    let request_str = serde_json::to_string(request)?;
    writeln!(stream, "{}", request_str)?;
    stream.flush()?;

    let mut reader = BufReader::new(stream);
    let mut response_str = String::new();
    reader.read_line(&mut response_str)?;

    let response: Response = serde_json::from_str(&response_str)?;
    Ok(response)
}

pub fn is_daemon_running(socket_path: &Path) -> bool {
    if !socket_path.exists() {
        return false;
    }
    UnixStream::connect(socket_path).is_ok()
}

pub fn start_daemon_background(
    base_path: &Path,
    storage_path: &Path,
    max_storage: Option<u64>,
) -> std::io::Result<()> {
    let socket_path = storage_path.join("daemon.sock");

    // Spawn the daemon as a detached child process.  The env var tells
    // main() to skip CLI parsing and run the daemon loop directly.
    // Using Command (instead of fork) avoids unsafe code and gives us
    // proper fd inheritance control — stdin/stdout/stderr are /dev/null
    // so callers that capture output won't block.
    let exe = std::env::current_exe()?;

    let mut cmd = Command::new(exe);
    cmd.args([
        "run-daemon",
        "--base",
        &base_path.to_string_lossy(),
        "--storage",
        &storage_path.to_string_lossy(),
    ]);
    if let Some(max) = max_storage {
        cmd.args(["--max-storage", &max.to_string()]);
    }
    cmd.stdin(Stdio::null())
        .stdout(Stdio::null())
        .stderr(Stdio::null())
        .spawn()?;

    // Wait for daemon to be ready
    for _ in 0..50 {
        std::thread::sleep(Duration::from_millis(100));
        if is_daemon_running(&socket_path) {
            return Ok(());
        }
    }

    Err(std::io::Error::new(
        std::io::ErrorKind::TimedOut,
        "Daemon failed to start",
    ))
}

pub fn ensure_daemon(
    base_path: Option<&Path>,
    storage_path: &Path,
    max_storage: Option<u64>,
) -> std::io::Result<()> {
    let socket_path = storage_path.join("daemon.sock");

    if is_daemon_running(&socket_path) {
        return Ok(());
    }

    let base_path = match base_path {
        Some(p) => p.to_path_buf(),
        None => {
            // Try to load from base_path file
            let base_file = storage_path.join("base_path");
            if base_file.exists() {
                let content = fs::read_to_string(&base_file)?;
                PathBuf::from(content.trim())
            } else {
                return Err(std::io::Error::new(
                    std::io::ErrorKind::NotFound,
                    "No daemon running and --base not specified. Use --base on first mount.",
                ));
            }
        }
    };

    start_daemon_background(&base_path, storage_path, max_storage)
}