xagent-pi 0.2.10

//! xagent-pi local agent runtime and command-line lifecycle management.

mod brain;
mod config;
mod pi;
mod session;
mod store;
mod tunnel;
mod web;

use anyhow::Context;
use clap::{Parser, Subcommand};
use config::Settings;
use std::fs::OpenOptions;
use std::io::ErrorKind;
use std::os::unix::process::CommandExt;
use std::path::Path;
use std::process::{Command, Stdio};
use std::time::{Duration, Instant};

#[derive(Parser)]
#[command(version, about = "XAGENT local Pi agent")]
struct Cli {
    #[command(subcommand)]
    command: CommandKind,
}

#[derive(Subcommand)]
enum CommandKind {
    /// Start xagent-pi in the background.
    Start,
    /// Run xagent-pi in the foreground.
    StartSync,
    /// Stop the background process.
    Stop,
    /// Show whether the background process is running.
    Status,
}

pub fn main_entry() -> anyhow::Result<()> {
    match Cli::parse().command {
        CommandKind::Start => start(),
        CommandKind::StartSync => start_sync(),
        CommandKind::Stop => stop(),
        CommandKind::Status => status(),
    }
}

fn start() -> anyhow::Result<()> {
    let settings = Settings::load()?;
    std::fs::create_dir_all(config::data_home()?)?;
    if let Some(pid) = running_pid()? {
        println!("xagent-pi 已在运行 (PID {pid})");
        return Ok(());
    }
    clear_stale_pid()?;
    let log_path = config::log_path()?;
    let stdout = OpenOptions::new()
        .create(true)
        .append(true)
        .open(&log_path)?;
    let stderr = stdout.try_clone()?;
    let mut command = Command::new(std::env::current_exe()?);
    command
        .arg("start-sync")
        .current_dir(&settings.cwd)
        .stdin(Stdio::null())
        .stdout(Stdio::from(stdout))
        .stderr(Stdio::from(stderr));
    command.process_group(0);
    let mut child = command.spawn().context("启动后台进程失败")?;
    let deadline = Instant::now() + Duration::from_secs(5);
    loop {
        if let Some(pid) = running_pid()? {
            println!("xagent-pi 已启动 (PID {pid})");
            println!("日志: {}", log_path.display());
            return Ok(());
        }
        if let Some(exit) = child.try_wait()? {
            anyhow::bail!("xagent-pi 启动失败 ({exit})，请查看 {}", log_path.display());
        }
        if Instant::now() >= deadline {
            anyhow::bail!("等待后台进程启动超时，请查看 {}", log_path.display());
        }
        std::thread::sleep(Duration::from_millis(100));
    }
}

fn start_sync() -> anyhow::Result<()> {
    env_logger::Builder::from_env(env_logger::Env::default().default_filter_or("info")).init();
    let settings = Settings::load()?;
    std::fs::create_dir_all(config::data_home()?)?;
    let _guard = PidGuard::acquire()?;
    tokio::runtime::Builder::new_multi_thread()
        .enable_all()
        .build()?
        .block_on(run(settings))
}

async fn run(settings: Settings) -> anyhow::Result<()> {
    let account = settings.account();
    let tunnel_token = uuid::Uuid::new_v4().to_string();
    log::info!(
        "config: service={} brain={} data={} pi={} cwd={}",
        settings.service_url,
        settings.brain_addr,
        settings.data_dir().display(),
        settings.pi_path,
        settings.cwd.display()
    );
    let state_dir = config::data_home()?;
    let store = store::Store::new(settings.data_dir().to_path_buf())?;
    let manager = brain::Manager::new(
        store,
        settings.pi_path.clone(),
        settings.cwd.clone(),
        state_dir,
        settings.username.clone(),
        settings.password.clone(),
        tunnel_token.clone(),
    )?;
    {
        let url = settings.service_url.clone();
        let origin = format!("http://{}", settings.brain_addr);
        tokio::spawn(async move {
            loop {
                match tunnel::register_account(&url, &account).await {
                    Ok(()) => {
                        if let Err(error) = tunnel::run(
                            url.clone(),
                            account.clone(),
                            origin.clone(),
                            tunnel_token.clone(),
                        ).await {
                            log::warn!("tunnel unavailable: {error}");
                        }
                    }
                    Err(error) => log::warn!(
                        "remote service unavailable: {error}; local access remains available at {origin}"
                    ),
                }
                tokio::time::sleep(Duration::from_secs(2)).await;
            }
        });
    }
    manager.serve(&settings.brain_addr).await
}

fn stop() -> anyhow::Result<()> {
    let Some(pid) = running_pid()? else {
        clear_stale_pid()?;
        println!("xagent-pi 未运行");
        return Ok(());
    };
    let result = unsafe { libc::kill(pid, libc::SIGTERM) };
    if result != 0 {
        return Err(std::io::Error::last_os_error().into());
    }
    for _ in 0..50 {
        if !process_exists(pid) {
            clear_stale_pid()?;
            println!("xagent-pi 已停止");
            return Ok(());
        }
        std::thread::sleep(Duration::from_millis(100));
    }
    anyhow::bail!("停止超时 (PID {pid})")
}

fn status() -> anyhow::Result<()> {
    match running_pid()? {
        Some(pid) => {
            println!("xagent-pi 正在运行 (PID {pid})");
            Ok(())
        }
        None => {
            println!("xagent-pi 未运行");
            std::process::exit(1)
        }
    }
}

fn read_pid(path: &Path) -> anyhow::Result<Option<i32>> {
    match std::fs::read_to_string(path) {
        Ok(value) => Ok(value.trim().parse().ok()),
        Err(error) if error.kind() == ErrorKind::NotFound => Ok(None),
        Err(error) => Err(error.into()),
    }
}

fn process_exists(pid: i32) -> bool {
    if std::fs::read_to_string(format!("/proc/{pid}/stat"))
        .ok()
        .and_then(|stat| stat.rsplit_once(") ").map(|(_, fields)| fields.starts_with('Z')))
        == Some(true)
    {
        return false;
    }
    let result = unsafe { libc::kill(pid, 0) };
    result == 0 || std::io::Error::last_os_error().raw_os_error() == Some(libc::EPERM)
}

fn running_pid() -> anyhow::Result<Option<i32>> {
    Ok(read_pid(&config::pid_path()?)?.filter(|pid| process_exists(*pid)))
}

fn clear_stale_pid() -> anyhow::Result<()> {
    let path = config::pid_path()?;
    if path.exists() && running_pid()?.is_none() {
        std::fs::remove_file(path)?;
    }
    Ok(())
}

struct PidGuard {
    path: std::path::PathBuf,
}

impl PidGuard {
    fn acquire() -> anyhow::Result<Self> {
        let path = config::pid_path()?;
        if let Some(pid) = running_pid()? {
            anyhow::bail!("xagent-pi 已在运行 (PID {pid})");
        }
        clear_stale_pid()?;
        let mut file = OpenOptions::new()
            .write(true)
            .create_new(true)
            .open(&path)
            .context("无法创建 PID 文件")?;
        use std::io::Write;
        writeln!(file, "{}", std::process::id())?;
        Ok(Self { path })
    }
}

impl Drop for PidGuard {
    fn drop(&mut self) {
        let _ = std::fs::remove_file(&self.path);
    }
}