wsb-daemon 0.248.6265

//! Every project and worktree on this machine, and what `prj` runs in them.
//!
//! `$PRJ` holds the projects, each a folder of worktrees; `prj` keeps the
//! Claude Code slots that run in them. This is what the machine home shows: a
//! card per project, a row per worktree with its git state and its slots.
//!
//! # Read here, acted on by prj
//!
//! Slots are `prj`'s. They are read through `prj ls --json` and every action
//! on them — resume, stop, start, making or removing a worktree — runs `prj`
//! itself, so there is one owner of what a slot is and one place its rules
//! live. A slot list that cannot be read is said so, beside worktrees that
//! can: the layout is wsb's to read, the slots are not.

use std::collections::BTreeMap;
use std::path::{Path, PathBuf};
use std::sync::Arc;

use axum::{
    extract::{Path as UrlPath, State},
    http::StatusCode,
    response::sse::{Event, KeepAlive, Sse},
    routing::{delete, get, post},
    Json, Router,
};
use serde::{Deserialize, Serialize};

use crate::machine::MachineState;

type Refusal = (StatusCode, Json<serde_json::Value>);

fn refuse(status: StatusCode, error: impl std::fmt::Display) -> Refusal {
    (status, Json(serde_json::json!({ "error": error.to_string() })))
}

pub fn routes() -> Router<Arc<MachineState>> {
    Router::new()
        .route("/api/v1/projects", get(list))
        .route("/api/v1/projects/events", get(events))
        .route("/api/v1/projects/:project", get(project))
        .route("/api/v1/prj/slots/:name/resume", post(resume_slot))
        .route("/api/v1/prj/slots/:name/stop", post(stop_slot))
        .route("/api/v1/prj/start", post(start))
        .route("/api/v1/prj/worktrees", post(add_worktree))
        .route("/api/v1/prj/worktrees/:id", delete(remove_worktree))
}

/// `GET /api/v1/projects`.
#[derive(Debug, Clone, PartialEq, Serialize)]
pub struct Projects {
    /// `$PRJ`, or none on a machine without one.
    pub prj_root: Option<String>,
    pub projects: Vec<Project>,
    /// Worktrees this machine has opened that are in no project.
    pub standalone: Vec<Worktree>,
    /// Why the slots could not be read, when they could not. The worktrees
    /// are still listed; their slots are then empty for that reason, not
    /// because none run.
    pub slots_error: Option<String>,
}

#[derive(Debug, Clone, PartialEq, Serialize)]
pub struct Project {
    pub name: String,
    pub path: String,
    pub worktrees: Vec<Worktree>,
}

#[derive(Debug, Clone, PartialEq, Serialize)]
pub struct Worktree {
    pub id: String,
    pub project: Option<String>,
    pub name: String,
    pub path: String,
    /// The superproject's branch; none when it is detached.
    pub branch: Option<String>,
    /// The commit it is on; none before its first commit.
    pub head: Option<String>,
    /// How many paths have uncommitted changes, untracked files and
    /// submodules included; 0 is clean.
    pub dirty: u64,
    pub upstream: Option<String>,
    pub ahead: Option<u64>,
    pub behind: Option<u64>,
    /// Unix seconds of the commit it is on.
    pub last_commit_at: Option<i64>,
    /// Whether this daemon has it open.
    pub open: bool,
    /// Why its git state could not be read, when it could not.
    pub error: Option<String>,
    pub slots: Vec<Slot>,
}

/// One `prj` slot, as `prj ls --json` reports it.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct Slot {
    pub key: String,
    pub name: String,
    pub status: String,
    #[serde(default)]
    pub detail: Option<String>,
    #[serde(default)]
    pub session_id: Option<String>,
    pub activity_at: Option<i64>,
}

/// A row of `prj ls --json`: a slot, and where it runs.
#[derive(Debug, Deserialize)]
struct SlotRow {
    project: String,
    worktree: String,
    #[serde(flatten)]
    slot: Slot,
}

/// The machine's projects as last read, kept current by the daemon.
///
/// Reading every worktree's git state takes seconds on a machine with large
/// superprojects, and a page browsing projects must not wait seconds per
/// click. So the daemon reads it when something moves — a folder, a slot, a
/// worktree opening, an action — and every half minute for the changes no
/// watcher sees (an edit inside a worktree), and pages are told only when
/// what it read differs from what they have.
pub struct Listing {
    /// The `$PRJ` it lists; none on a machine without one.
    prj: Option<wsb_prj::Prj>,
    current: tokio::sync::RwLock<Option<Arc<Result<Projects, String>>>>,
    /// One reading at a time.
    reading: tokio::sync::Mutex<()>,
    wanted: tokio::sync::Notify,
    /// Raised when a reading differs from the one before it.
    pub changed: tokio::sync::broadcast::Sender<()>,
}

/// How often the listing is read again with nothing having asked.
const RELIST_EVERY: std::time::Duration = std::time::Duration::from_secs(30);

impl Listing {
    pub fn new(prj: Option<wsb_prj::Prj>) -> Self {
        Self {
            prj,
            current: tokio::sync::RwLock::new(None),
            reading: tokio::sync::Mutex::new(()),
            wanted: tokio::sync::Notify::new(),
            changed: tokio::sync::broadcast::channel(16).0,
        }
    }

    /// Ask for the listing to be read again.
    pub fn want(&self) {
        self.wanted.notify_one();
    }

    /// The last reading, or a first one now when there is none.
    async fn get(&self, machine: &MachineState) -> Arc<Result<Projects, String>> {
        if let Some(current) = self.current.read().await.clone() {
            return current;
        }
        self.read(machine).await
    }

    /// Read the listing, keep it, and say whether it changed.
    async fn read(&self, machine: &MachineState) -> Arc<Result<Projects, String>> {
        let _one = self.reading.lock().await;
        let open = open_ids(machine).await;
        let reading = match machine.store.worktrees() {
            Ok(recorded) => {
                let prj = self.prj.clone();
                tokio::task::spawn_blocking(move || read_projects(prj.as_ref(), &open, &recorded))
                .await
                    .unwrap_or_else(|e| Err(format!("reading the projects stopped: {e}")))
            }
            Err(error) => Err(error),
        };
        let reading = Arc::new(reading);
        let mut current = self.current.write().await;
        let differs = current.as_deref() != Some(&*reading);
        *current = Some(reading.clone());
        drop(current);
        if differs {
            // No page listening is an ordinary state, not an error.
            let _ = self.changed.send(());
        }
        reading
    }
}

/// Keep the listing current: read it again whenever it is wanted, and every
/// [`RELIST_EVERY`] regardless.
pub fn spawn_listing(machine: Arc<MachineState>) {
    tokio::spawn(async move {
        loop {
            machine.listing.read(&machine).await;
            tokio::select! {
                _ = machine.listing.wanted.notified() => {}
                _ = tokio::time::sleep(RELIST_EVERY) => {}
            }
        }
    });
}

async fn list(State(machine): State<Arc<MachineState>>) -> Result<Json<Projects>, Refusal> {
    match &*machine.listing.get(&machine).await {
        Ok(projects) => Ok(Json(projects.clone())),
        Err(error) => Err(refuse(StatusCode::INTERNAL_SERVER_ERROR, error)),
    }
}

async fn open_ids(machine: &MachineState) -> std::collections::BTreeSet<String> {
    machine
        .all()
        .await
        .into_iter()
        .map(|runtime| runtime.registered.id.to_string())
        .collect()
}

/// Every project and worktree, read now.
fn read_projects(
    prj: Option<&wsb_prj::Prj>,
    open: &std::collections::BTreeSet<String>,
    recorded: &[wsb::daemon::RegisteredWorktree],
) -> Result<Projects, String> {
    let (mut slots, slots_error) = match prj {
        Some(_) => match read_slots() {
            Ok(slots) => (slots, None),
            Err(error) => (BTreeMap::new(), Some(error)),
        },
        None => (BTreeMap::new(), None),
    };

    let mut projects = Vec::new();
    if let Some(prj) = prj {
        for project in prj.projects().map_err(|e| e.to_string())? {
            let mut worktrees = Vec::new();
            for worktree in project.worktrees().map_err(|e| e.to_string())? {
                let id = format!("{}:{}", project.name(), worktree.name);
                let here = slots
                    .remove(&(project.name().to_string(), worktree.name.clone()))
                    .unwrap_or_default();
                worktrees.push(describe(
                    id.clone(),
                    Some(project.name().to_string()),
                    worktree.name,
                    &worktree.path,
                    open.contains(&id),
                    here,
                ));
            }
            projects.push(Project {
                name: project.name().to_string(),
                path: project.dir().display().to_string(),
                worktrees: in_parallel(worktrees),
            });
        }
    }
    let standalone = recorded
        .iter()
        .filter(|worktree| worktree.project.is_none() && worktree.available)
        .map(|worktree| {
            describe(
                worktree.id.to_string(),
                None,
                worktree.name.clone(),
                &worktree.super_root,
                open.contains(worktree.id.as_str()),
                Vec::new(),
            )
        })
        .collect();
    Ok(Projects {
        prj_root: prj.map(|prj| prj.root().display().to_string()),
        projects,
        standalone: in_parallel(standalone),
        slots_error,
    })
}

/// A worktree whose git state is still to be read.
struct Pending {
    worktree: Worktree,
    path: PathBuf,
}

fn describe(
    id: String,
    project: Option<String>,
    name: String,
    path: &Path,
    open: bool,
    slots: Vec<Slot>,
) -> Pending {
    Pending {
        worktree: Worktree {
            id,
            project,
            name,
            path: path.display().to_string(),
            branch: None,
            head: None,
            dirty: 0,
            upstream: None,
            ahead: None,
            behind: None,
            last_commit_at: None,
            open,
            error: None,
            slots,
        },
        path: path.to_path_buf(),
    }
}

/// Read every worktree's git state at once: one `git status` each, and a
/// machine with twenty worktrees should not wait for twenty in a row.
fn in_parallel(pending: Vec<Pending>) -> Vec<Worktree> {
    std::thread::scope(|scope| {
        let handles: Vec<_> = pending
            .into_iter()
            .map(|Pending { mut worktree, path }| {
                scope.spawn(move || {
                    match git_state(&path) {
                        Ok(state) => {
                            worktree.branch = state.branch;
                            worktree.head = state.head;
                            worktree.dirty = state.dirty;
                            worktree.upstream = state.upstream;
                            worktree.ahead = state.ahead;
                            worktree.behind = state.behind;
                            worktree.last_commit_at = state.last_commit_at;
                        }
                        Err(error) => worktree.error = Some(error),
                    }
                    worktree
                })
            })
            .collect();
        handles
            .into_iter()
            .map(|handle| handle.join().expect("reading a worktree's git state panicked"))
            .collect()
    })
}

/// A superproject's state, as `git status --porcelain=v2 --branch` gives it.
#[derive(Debug, Default, PartialEq, Eq)]
pub struct GitState {
    pub branch: Option<String>,
    pub head: Option<String>,
    pub dirty: u64,
    pub upstream: Option<String>,
    pub ahead: Option<u64>,
    pub behind: Option<u64>,
    pub last_commit_at: Option<i64>,
}

fn git_state(root: &Path) -> Result<GitState, String> {
    let output = wsb::git_backend::git_command()
        .arg("-C")
        .arg(root)
        // A reading, not a refresh: without this every listing rewrote each
        // worktree's index, and the listings of worktrees sharing main's
        // repositories queued behind each other's locks.
        .args(["--no-optional-locks", "status", "--porcelain=v2", "--branch"])
        .output()
        .map_err(|e| format!("cannot run git in {}: {e}", root.display()))?;
    if !output.status.success() {
        return Err(format!(
            "git status in {} failed: {}",
            root.display(),
            String::from_utf8_lossy(&output.stderr).trim()
        ));
    }
    let mut state = parse_status(&String::from_utf8_lossy(&output.stdout))?;
    if state.head.is_some() {
        let output = wsb::git_backend::git_command()
            .arg("-C")
            .arg(root)
            .args(["log", "-1", "--format=%ct"])
            .output()
            .map_err(|e| format!("cannot run git in {}: {e}", root.display()))?;
        let text = String::from_utf8_lossy(&output.stdout);
        state.last_commit_at = Some(text.trim().parse().map_err(|_| {
            format!("git log in {} gave no commit time: {:?}", root.display(), text.trim())
        })?);
    }
    Ok(state)
}

/// Read `git status --porcelain=v2 --branch`.
pub fn parse_status(text: &str) -> Result<GitState, String> {
    let mut state = GitState::default();
    for line in text.lines() {
        if let Some(header) = line.strip_prefix("# ") {
            let (key, value) = header.split_once(' ').unwrap_or((header, ""));
            match key {
                "branch.oid" if value != "(initial)" => state.head = Some(value.to_string()),
                "branch.head" if value != "(detached)" => state.branch = Some(value.to_string()),
                "branch.upstream" => state.upstream = Some(value.to_string()),
                "branch.ab" => {
                    let (ahead, behind) = value
                        .split_once(' ')
                        .ok_or_else(|| format!("git status gave a malformed branch.ab: {value:?}"))?;
                    let count = |raw: &str, sign: char| {
                        raw.strip_prefix(sign)
                            .and_then(|n| n.parse::<u64>().ok())
                            .ok_or_else(|| format!("git status gave a malformed branch.ab: {value:?}"))
                    };
                    state.ahead = Some(count(ahead, '+')?);
                    state.behind = Some(count(behind, '-')?);
                }
                _ => {}
            }
        } else if !line.is_empty() {
            // `1`, `2`, `u` and `?` lines are changes, conflicts and
            // untracked files: each one a path of uncommitted work.
            state.dirty += 1;
        }
    }
    Ok(state)
}

/// Every slot `prj` keeps, by `(project, worktree)`.
fn read_slots() -> Result<BTreeMap<(String, String), Vec<Slot>>, String> {
    let output = std::process::Command::new(prj_program()?)
        .args(["ls", "--json"])
        .output()
        .map_err(|e| format!("cannot run prj ls --json: {e}"))?;
    if !output.status.success() {
        return Err(format!(
            "prj ls --json failed: {}",
            String::from_utf8_lossy(&output.stderr).trim()
        ));
    }
    parse_slots(&String::from_utf8_lossy(&output.stdout))
}

/// Read `prj ls --json`: an array of slot rows.
pub fn parse_slots(text: &str) -> Result<BTreeMap<(String, String), Vec<Slot>>, String> {
    let rows: Vec<SlotRow> =
        serde_json::from_str(text).map_err(|e| format!("prj ls --json is not a list of slots: {e}"))?;
    let mut slots: BTreeMap<(String, String), Vec<Slot>> = BTreeMap::new();
    for row in rows {
        slots.entry((row.project, row.worktree)).or_default().push(row.slot);
    }
    Ok(slots)
}

/// The `prj` program, found on `PATH` as a person's shell finds it.
fn prj_program() -> Result<PathBuf, String> {
    wsb::report::which("prj").map(PathBuf::from).ok_or_else(|| {
        "prj is not on PATH; the toolset installs it (`dx install` in the toolset repository)"
            .to_string()
    })
}

/// `GET /api/v1/projects/:project` — one project, with the repositories its
/// worktrees push to and the boards it references.
async fn project(
    State(machine): State<Arc<MachineState>>,
    UrlPath(name): UrlPath<String>,
) -> Result<Json<serde_json::Value>, Refusal> {
    let listing = match &*machine.listing.get(&machine).await {
        Ok(projects) => projects.clone(),
        Err(error) => return Err(refuse(StatusCode::INTERNAL_SERVER_ERROR, error)),
    };
    let wanted = name.clone();
    let found = tokio::task::spawn_blocking(move || -> Result<(Project, Vec<String>, serde_json::Value, Option<String>), Refusal> {
        let project = listing
            .projects
            .into_iter()
            .find(|project| project.name == wanted)
            .ok_or_else(|| refuse(StatusCode::NOT_FOUND, format!("there is no project {wanted}")))?;
        let prj = wsb_prj::Prj::from_env().map_err(|e| refuse(StatusCode::INTERNAL_SERVER_ERROR, e))?;
        let located = prj.project(&wanted).map_err(|e| refuse(StatusCode::NOT_FOUND, e))?;
        let repositories = wsb_github::checkout::repositories(&located.main_worktree())
            .map_err(|e| refuse(StatusCode::INTERNAL_SERVER_ERROR, format!("{e:#}")))?
            .into_iter()
            .map(|repository| repository.name)
            .collect();
        let boards = serde_json::to_value(
            wsb_github::boards::read(&located)
                .map_err(|e| refuse(StatusCode::INTERNAL_SERVER_ERROR, format!("{e:#}")))?
                .boards,
        )
        .map_err(|e| refuse(StatusCode::INTERNAL_SERVER_ERROR, e))?;
        Ok((project, repositories, boards, listing.slots_error))
    })
    .await
    .map_err(|e| refuse(StatusCode::INTERNAL_SERVER_ERROR, e))??;
    let (project, repositories, boards, slots_error) = found;
    Ok(Json(serde_json::json!({
        "project": project,
        "repositories": repositories,
        "boards": boards,
        "slots_error": slots_error,
    })))
}

/// `GET /api/v1/projects/events` — `projects_changed` whenever the listing
/// differs from the last one read: a project or worktree came or went, a
/// slot changed, a worktree's branch or changes moved.
async fn events(
    State(machine): State<Arc<MachineState>>,
) -> Sse<impl futures::Stream<Item = Result<Event, std::convert::Infallible>>> {
    let mut changed = machine.listing.changed.subscribe();
    let stream = async_stream::stream! {
        yield Ok(Event::default().event("connected").data("{}"));
        loop {
            match changed.recv().await {
                Ok(()) | Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => {
                    yield Ok(Event::default().event("projects_changed").data("{}"));
                }
                Err(tokio::sync::broadcast::error::RecvError::Closed) => break,
            }
        }
    };
    Sse::new(stream).keep_alive(KeepAlive::default())
}

/// Run `prj` with `arguments`, and answer with what it said — or refuse with
/// what it said when it failed.
async fn run_prj(
    machine: &MachineState,
    arguments: Vec<String>,
) -> Result<Json<serde_json::Value>, Refusal> {
    let program = prj_program().map_err(|e| refuse(StatusCode::SERVICE_UNAVAILABLE, e))?;
    let shown = format!("prj {}", arguments.join(" "));
    let output = tokio::task::spawn_blocking(move || {
        std::process::Command::new(program).args(&arguments).output()
    })
    .await
    .map_err(|e| refuse(StatusCode::INTERNAL_SERVER_ERROR, e))?
    .map_err(|e| refuse(StatusCode::INTERNAL_SERVER_ERROR, format!("cannot run {shown}: {e}")))?;
    let stdout = String::from_utf8_lossy(&output.stdout).trim().to_string();
    let stderr = String::from_utf8_lossy(&output.stderr).trim().to_string();
    // Whatever it did, the slots or worktrees may have moved.
    machine.note_projects_changed();
    if output.status.success() {
        Ok(Json(serde_json::json!({ "command": shown, "output": stdout, "errors": stderr })))
    } else {
        Err(refuse(
            StatusCode::UNPROCESSABLE_ENTITY,
            format!("{shown} failed: {}", if stderr.is_empty() { stdout } else { stderr }),
        ))
    }
}

/// A slot name, as `prj` gives them, or a refusal: it becomes an argument.
fn slot_name(name: &str) -> Result<String, Refusal> {
    if wsb_prj::is_valid_name(name) {
        Ok(name.to_string())
    } else {
        Err(refuse(StatusCode::BAD_REQUEST, format!("{name:?} is not a slot name")))
    }
}

async fn resume_slot(
    State(machine): State<Arc<MachineState>>,
    UrlPath(name): UrlPath<String>,
) -> Result<Json<serde_json::Value>, Refusal> {
    // In a window of its own: without one the session runs attached to the
    // caller's terminal, and the daemon has none — prj would not return until
    // the session ended.
    run_prj(&machine, vec!["resume".into(), slot_name(&name)?, "--new-window".into()]).await
}

async fn stop_slot(
    State(machine): State<Arc<MachineState>>,
    UrlPath(name): UrlPath<String>,
) -> Result<Json<serde_json::Value>, Refusal> {
    run_prj(&machine, vec!["stop".into(), slot_name(&name)?]).await
}

#[derive(Debug, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct StartRequest {
    /// `<project>` or `<project>/<worktree>`.
    pub target: String,
}

/// `<project>[/<worktree>]`, checked, or a refusal.
fn target(raw: &str) -> Result<String, Refusal> {
    let (project, worktree) = match raw.split_once('/') {
        Some((project, worktree)) => (project, Some(worktree)),
        None => (raw, None),
    };
    let valid = wsb_prj::is_valid_name(project) && worktree.map_or(true, wsb_prj::is_valid_name);
    if valid {
        Ok(raw.to_string())
    } else {
        Err(refuse(StatusCode::BAD_REQUEST, format!("{raw:?} is not <project> or <project>/<worktree>")))
    }
}

async fn start(
    State(machine): State<Arc<MachineState>>,
    Json(body): Json<StartRequest>,
) -> Result<Json<serde_json::Value>, Refusal> {
    run_prj(&machine, vec!["start".into(), target(&body.target)?, "--new-window".into()]).await
}

#[derive(Debug, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct AddWorktreeRequest {
    pub project: String,
    pub name: String,
    pub branch: Option<String>,
    pub from: Option<String>,
}

/// A git ref as an argument: no option a program could mistake it for.
fn git_ref(raw: &str) -> Result<String, Refusal> {
    let refused = raw.is_empty()
        || raw.starts_with('-')
        || raw.chars().any(|c| c.is_whitespace() || c.is_control());
    if refused {
        Err(refuse(StatusCode::BAD_REQUEST, format!("{raw:?} is not a branch or commit")))
    } else {
        Ok(raw.to_string())
    }
}

async fn add_worktree(
    State(machine): State<Arc<MachineState>>,
    Json(body): Json<AddWorktreeRequest>,
) -> Result<Json<serde_json::Value>, Refusal> {
    let mut arguments = vec![
        "wt".to_string(),
        "add".to_string(),
        target(&format!("{}/{}", body.project, body.name))?,
    ];
    if let Some(branch) = &body.branch {
        arguments.extend(["--branch".to_string(), git_ref(branch)?]);
    }
    if let Some(from) = &body.from {
        arguments.extend(["--from".to_string(), git_ref(from)?]);
    }
    run_prj(&machine, arguments).await
}

async fn remove_worktree(
    State(machine): State<Arc<MachineState>>,
    UrlPath(id): UrlPath<String>,
) -> Result<Json<serde_json::Value>, Refusal> {
    let id = wsb::daemon::WorktreeId::parse(&id).map_err(|e| refuse(StatusCode::BAD_REQUEST, e))?;
    let (project, worktree) = id
        .in_project()
        .ok_or_else(|| refuse(StatusCode::BAD_REQUEST, format!("{id} is not a worktree of a project")))?;
    if worktree == wsb_prj::MAIN {
        return Err(refuse(StatusCode::BAD_REQUEST, "a project's main worktree is not removed"));
    }
    run_prj(&machine, vec!["wt".into(), "rm".into(), format!("{project}/{worktree}")]).await
}

/// Follow `$PRJ` and `prj`'s slots, and say when either moves.
///
/// What is watched is exactly what the listing is made of: `$PRJ` itself
/// (a project added or removed), each project folder (a worktree added or
/// removed), and `prj`'s `state.json` and `run/` in this machine's folder (a
/// slot started, stopped, parked). Not recursive — a build inside a worktree
/// writes thousands of files, and none of them changes the list.
pub fn spawn_watcher(machine: Arc<MachineState>) {
    use notify_debouncer_mini::{new_debouncer, notify::RecursiveMode, DebouncedEvent};

    let prj = match wsb_prj::Prj::from_env() {
        Ok(prj) => prj,
        Err(wsb_prj::PrjError::NotSet) => return,
        Err(error) => {
            tracing::warn!(%error, "projects are not followed");
            return;
        }
    };
    let host = machine.machine_dir.clone();
    let runtime = tokio::runtime::Handle::current();
    std::thread::spawn(move || {
        let notify = machine.clone();
        let watched_host = host.clone();
        let mut debouncer = match new_debouncer(
            std::time::Duration::from_millis(400),
            move |events: Result<Vec<DebouncedEvent>, notify::Error>| match events {
                Ok(events) => {
                    let relevant = events.iter().any(|event| {
                        // In the machine's folder only prj's files count:
                        // the daemon's own database lives beside them.
                        match event.path.strip_prefix(&watched_host) {
                            Ok(rest) => {
                                rest.starts_with("state.json") || rest.starts_with("run")
                            }
                            Err(_) => true,
                        }
                    });
                    if relevant {
                        notify.note_projects_changed();
                    }
                }
                Err(error) => tracing::warn!(%error, "the projects watcher reported an error"),
            },
        ) {
            Ok(debouncer) => debouncer,
            Err(error) => {
                tracing::warn!(%error, "projects are not followed");
                return;
            }
        };
        let mut watched: std::collections::BTreeSet<PathBuf> = Default::default();
        let mut changed = machine.listing.changed.subscribe();
        loop {
            let mut wanted = vec![prj.root().to_path_buf(), host.clone(), host.join("run")];
            match prj.projects() {
                Ok(projects) => wanted.extend(projects.iter().map(|p| p.dir().to_path_buf())),
                Err(error) => tracing::warn!(%error, "cannot list the projects to follow"),
            }
            for dir in wanted {
                if dir.is_dir() && !watched.contains(&dir) {
                    match debouncer.watcher().watch(&dir, RecursiveMode::NonRecursive) {
                        Ok(()) => {
                            watched.insert(dir);
                        }
                        Err(error) => {
                            tracing::warn!(%error, dir = %dir.display(), "cannot follow")
                        }
                    }
                }
            }
            watched.retain(|dir| dir.is_dir());
            // A change may be a new project folder to follow; look again
            // after each one.
            if runtime.block_on(changed.recv()).is_err_and(|e| {
                matches!(e, tokio::sync::broadcast::error::RecvError::Closed)
            }) {
                return;
            }
        }
    });
}

#[cfg(test)]
#[path = "prj_tests.rs"]
mod tests;