pitchfork-cli 2.30.0

//! Autostop logic and boot daemon startup
//!
//! Handles automatic stopping of daemons when shells leave directories,
//! and starting daemons configured with `boot_start = true`.

use super::{SUPERVISOR, Supervisor};
use crate::Result;
use crate::daemon_id::DaemonId;
use crate::ipc::IpcResponse;
use crate::pitchfork_toml::PitchforkToml;
use crate::settings::settings;
use log::LevelFilter::Info;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
use tokio::time;

/// Whether `path` is equal to or nested inside `base`.
///
/// Tries a direct `starts_with` first (fast path for paths that share the same
/// representation). Falls back to canonicalizing both sides to bridge
/// representation differences: the Windows verbatim (`\\?\`) prefix that
/// `canonicalize()` adds versus bare paths coming from `env::CWD` / config-file
/// parents, and symlink resolution (e.g. macOS `/tmp` -> `/private/tmp`).
/// `canonicalize` is only invoked when the cheap string comparison already
/// fails, so the common case stays allocation-free.
pub(super) fn is_within(base: &Path, path: &Path) -> bool {
    if path.starts_with(base) {
        return true;
    }
    canonicalize_pair(base, path).is_some_and(|(b, p)| p.starts_with(&b))
}

/// Whether `a` and `b` are equal, or one contains the other (bidirectional).
fn dirs_overlap(a: &Path, b: &Path) -> bool {
    is_within(a, b) || is_within(b, a)
}

/// Canonicalize both paths, returning `None` if either fails (e.g. the path no
/// longer exists). Callers treat `None` as "no ancestry relationship known".
fn canonicalize_pair(a: &Path, b: &Path) -> Option<(PathBuf, PathBuf)> {
    Some((a.canonicalize().ok()?, b.canonicalize().ok()?))
}

impl Supervisor {
    /// Handle shell leaving a directory - schedule autostops for daemons
    pub(crate) async fn leave_dir(&self, dir: &Path) -> Result<()> {
        debug!("left dir {}", dir.display());
        let active_dirs = self.get_active_directories().await;
        debug!("active directories after leaving {dir:?}: {active_dirs:?}");
        let autostop_delay = settings().general_autostop_delay();

        for daemon in self.active_daemons().await {
            if !daemon.autostop {
                continue;
            }
            // if this daemon's dir is within the left dir
            // and no other active directory is within the daemon's dir
            // schedule the daemon for autostop
            if let Some(daemon_dir) = daemon.dir.as_ref() {
                let starts = is_within(dir, daemon_dir);
                let still_active = active_dirs.iter().any(|d| is_within(daemon_dir, d));
                debug!(
                    "leave_dir daemon={} daemon_dir={daemon_dir:?} starts_with_left={starts} still_active={still_active}",
                    daemon.id
                );
                if starts && !still_active {
                    if autostop_delay.is_zero() {
                        // No delay configured, stop immediately. Spawned: a stop
                        // now waits for the daemon's whole process group to exit
                        // (up to its stop budget), and this path runs inside the
                        // UpdateShellDir handler — blocking here would stall the
                        // user's `cd` shell hook until the stop completes.
                        info!("autostopping {daemon}");
                        self.spawn_autostop(daemon.id.clone(), daemon.to_string())
                            .await;
                    } else {
                        // Schedule autostop with delay
                        let stop_at = time::Instant::now() + autostop_delay;
                        let mut pending = self.pending_autostops.lock().await;
                        if !pending.contains_key(&daemon.id) {
                            info!(
                                "scheduling autostop for {} in {:?}",
                                daemon.id, autostop_delay
                            );
                            pending.insert(daemon.id.clone(), stop_at);
                        }
                    }
                }
            }
        }
        Ok(())
    }

    /// Cancel any pending autostop for daemons in the given directory
    /// Also cancels autostops for daemons in parent directories (e.g., entering /project/subdir
    /// cancels pending autostop for daemon in /project)
    pub(crate) async fn cancel_pending_autostops_for_dir(&self, dir: &Path) {
        let mut pending = self.pending_autostops.lock().await;
        let daemons_to_cancel: Vec<DaemonId> = {
            let state_file = self.state_file.lock().await;
            state_file
                .daemons
                .iter()
                .filter(|(_id, d)| {
                    d.dir.as_ref().is_some_and(|daemon_dir| {
                        // Cancel if entering a directory inside or equal to daemon's directory
                        // OR if daemon is in a subdirectory of the entered directory
                        dirs_overlap(dir, daemon_dir)
                    })
                })
                .map(|(id, _)| id.clone())
                .collect()
        };

        for daemon_id in daemons_to_cancel {
            if pending.remove(&daemon_id).is_some() {
                info!("cancelled pending autostop for {daemon_id}");
            }
            // Also call off any stop task already spawned but not yet stopping.
            // The entry is removed so a later leave can schedule a fresh
            // autostop; the task keeps its own clone of the flag.
            if let Some(flag) = self.in_flight_autostops.lock().await.remove(&daemon_id) {
                flag.store(true, Ordering::SeqCst);
                info!("cancelled in-flight autostop for {daemon_id}");
            }
        }
    }

    /// Spawn a detached stop task for an autostopped daemon.
    ///
    /// The task is registered in `in_flight_autostops` so that
    /// `cancel_pending_autostops_for_dir` can call it off if a shell
    /// re-enters the directory before the stop begins, and it revalidates
    /// directory activity right before stopping since the scheduling check
    /// ran before the task was spawned.
    async fn spawn_autostop(&self, daemon_id: DaemonId, label: String) {
        let cancelled = Arc::new(AtomicBool::new(false));
        {
            let mut in_flight = self.in_flight_autostops.lock().await;
            if in_flight.contains_key(&daemon_id) {
                debug!("autostop already in flight for {daemon_id}");
                return;
            }
            in_flight.insert(daemon_id.clone(), cancelled.clone());
        }
        tokio::spawn(async move {
            SUPERVISOR
                .run_autostop(&daemon_id, &label, &cancelled)
                .await;
            // Only remove our own registration: a cancelled entry may have
            // been replaced by a newer in-flight autostop for the same daemon.
            let mut in_flight = SUPERVISOR.in_flight_autostops.lock().await;
            if in_flight
                .get(&daemon_id)
                .is_some_and(|f| Arc::ptr_eq(f, &cancelled))
            {
                in_flight.remove(&daemon_id);
            }
        });
    }

    /// Body of a detached autostop task: revalidate, then stop.
    async fn run_autostop(&self, daemon_id: &DaemonId, label: &str, cancelled: &AtomicBool) {
        if cancelled.load(Ordering::SeqCst) {
            debug!("autostop of {daemon_id} cancelled before stopping");
            return;
        }
        // Held from the recheck through the stop, so a start that finishes in
        // between — which may no longer ask to be autostopped — is not the
        // process that gets stopped.
        let lock = self.stop_lock(daemon_id).await;
        let _guard = lock.lock().await;
        // Revalidate that the daemon's directory is still inactive: a shell
        // may have entered it between the scheduling check and this task
        // running.
        if let Some(daemon) = self.get_daemon(daemon_id).await
            && daemon.autostop
            && daemon.status.is_running()
        {
            if let Some(daemon_dir) = daemon.dir.as_ref() {
                let active_dirs = self.get_active_directories().await;
                if active_dirs.iter().any(|d| is_within(daemon_dir, d)) {
                    debug!("autostop of {daemon_id} skipped: directory active again");
                    return;
                }
            }
        } else {
            debug!("autostop of {daemon_id} skipped: no longer running");
            return;
        }
        if cancelled.load(Ordering::SeqCst) {
            debug!("autostop of {daemon_id} cancelled before stopping");
            return;
        }
        match self.stop_locked(daemon_id).await {
            Ok(_) => {
                self.add_notification(Info, format!("autostopped {label}"))
                    .await;
            }
            Err(e) => error!("failed to autostop {daemon_id}: {e}"),
        }
    }

    /// Process any pending autostops that have reached their scheduled time
    pub(crate) async fn process_pending_autostops(&self) -> Result<()> {
        let now = time::Instant::now();
        let to_stop: Vec<DaemonId> = {
            let pending = self.pending_autostops.lock().await;
            pending
                .iter()
                .filter(|(_, stop_at)| now >= **stop_at)
                .map(|(id, _)| id.clone())
                .collect()
        };

        for daemon_id in to_stop {
            // Remove from pending first
            {
                let mut pending = self.pending_autostops.lock().await;
                pending.remove(&daemon_id);
            }

            // Check if daemon is still running and should be stopped
            if let Some(daemon) = self.get_daemon(&daemon_id).await
                && daemon.autostop
                && daemon.status.is_running()
            {
                // Verify no active directory is in the daemon's directory
                let active_dirs = self.get_active_directories().await;
                if let Some(daemon_dir) = daemon.dir.as_ref() {
                    let still_active = active_dirs.iter().any(|d| is_within(daemon_dir, d));
                    debug!(
                        "process_pending_autostops daemon={daemon_id} daemon_dir={daemon_dir:?} active_dirs={active_dirs:?} still_active={still_active}"
                    );
                    if still_active {
                        debug!(
                            "process_pending_autostops: daemon={daemon_id} still has active directory, skipping"
                        );
                        continue;
                    }
                    // Spawned: a stop now waits for the daemon's whole process
                    // group to exit (up to its stop budget). This path runs from
                    // the interval watcher and the UpdateShellDir handler —
                    // blocking here would stall the `cd` shell hook and delay
                    // other watcher duties (e.g. retries) behind each stop.
                    info!("autostopping {daemon_id} (after delay)");
                    let label = daemon_id.to_string();
                    self.spawn_autostop(daemon_id, label).await;
                }
            }
        }
        Ok(())
    }

    /// Start daemons configured with `boot_start = true`
    pub(crate) async fn start_boot_daemons(&self) -> Result<()> {
        info!("Scanning for boot_start daemons");
        let pt = PitchforkToml::all_merged_all_namespaces().await?;
        let disabled = self.state_file.lock().await.disabled.clone();

        let boot_daemons: Vec<_> = pt
            .daemons
            .iter()
            .filter(|(id, d)| {
                if !d.boot_start.unwrap_or(false) {
                    return false;
                }
                if disabled.contains(*id) {
                    info!("Skipping boot daemon {id}: it is disabled");
                    return false;
                }
                true
            })
            .collect();

        if boot_daemons.is_empty() {
            info!("No daemons configured with boot_start = true");
            return Ok(());
        }

        info!("Found {} daemon(s) to start at boot", boot_daemons.len());

        // Dependencies first, so a template naming another boot daemon's port
        // is rendered after that daemon has one. Only boot daemons are
        // started; `depends` just orders them.
        let boot_ids: Vec<DaemonId> = boot_daemons.iter().map(|(id, _)| (*id).clone()).collect();
        let order: Vec<DaemonId> = match crate::deps::resolve_dependencies(&boot_ids, &pt.daemons) {
            Ok(order) => order
                .levels
                .into_iter()
                .flatten()
                .filter(|id| boot_ids.contains(id))
                .collect(),
            Err(e) => {
                warn!("cannot order boot daemons by their dependencies ({e}); using config order");
                boot_ids
            }
        };

        for id in &order {
            let Some(daemon) = pt.daemons.get(id) else {
                continue;
            };
            info!("Starting boot daemon: {id}");

            // Built as `pitchfork start` would, just before starting, so
            // templates and the top-level `[env]` apply at boot too.
            let mut run_opts = match self.run_options_from_config(id, daemon, &pt).await {
                Ok(opts) => opts,
                Err(e) => {
                    error!("failed to start boot daemon {id}: {e}");
                    continue;
                }
            };
            run_opts.autostop = false; // Boot daemons should not autostop
            run_opts.wait_ready = false; // Don't block on boot daemons

            match self.run(run_opts).await {
                Ok(IpcResponse::DaemonStart { .. }) | Ok(IpcResponse::DaemonReady { .. }) => {
                    info!("Successfully started boot daemon: {id}");
                }
                Ok(IpcResponse::DaemonAlreadyRunning) => {
                    info!("Boot daemon already running: {id}");
                }
                Ok(other) => {
                    warn!("Unexpected response when starting boot daemon {id}: {other:?}");
                }
                Err(e) => {
                    error!("Failed to start boot daemon {id}: {e}");
                }
            }
        }

        Ok(())
    }
}