semora-cli 0.4.0

//! `semora watch`: push once, then re-push whenever files change.

use std::path::{Path, PathBuf};
use std::time::Duration;

use anyhow::Context;
use notify::{EventKind, RecursiveMode, Watcher};
use tokio::sync::mpsc::{UnboundedReceiver, unbounded_channel};

use crate::args::WatchArgs;
use crate::client::PushQuery;
use crate::commands::push::{hint_outside_project, note_outside_project, push_path, tags};
use crate::{Ctx, exit, output, project_link};

pub const QUIET: Duration = Duration::from_millis(500);

/// Waits for a change, then keeps absorbing changes until `quiet` passes with none.
/// Returns `None` when the sender is gone.
pub async fn debounce(rx: &mut UnboundedReceiver<()>, quiet: Duration) -> Option<()> {
    rx.recv().await?;
    loop {
        match tokio::time::timeout(quiet, rx.recv()).await {
            Ok(Some(())) => continue,
            Ok(None) | Err(_) => return Some(()),
        }
    }
}

/// True when a changed path should trigger a push.
pub fn relevant(root: &Path, single_file: Option<&Path>, changed: &Path) -> bool {
    if let Some(f) = single_file {
        return changed == f;
    }
    let Ok(rel) = changed.strip_prefix(root) else {
        return false;
    };
    !rel.components().any(|c| {
        let s = c.as_os_str().to_string_lossy();
        s.starts_with('.') || s == "node_modules"
    })
}

fn print_line(ctx: &Ctx, r: &semora_proto::PushResponse) {
    if ctx.json {
        println!("{}", serde_json::to_string(r).expect("serializable"));
    } else {
        let now = time::OffsetDateTime::now_utc();
        let fmt = time::macros::format_description!("[hour]:[minute]:[second]");
        println!(
            "[{}] r{} pushed  {}",
            now.format(&fmt).unwrap_or_default(),
            r.rev,
            r.path_url.as_deref().unwrap_or(&r.url)
        );
    }
    for b in &r.broken_links {
        output::warn(format!("broken link in {}: {}", b.from, b.href));
    }
}

pub async fn run(ctx: &Ctx, args: WatchArgs) -> anyhow::Result<i32> {
    let client = ctx.client()?;
    let path: PathBuf = std::fs::canonicalize(&args.path)
        .with_context(|| format!("cannot read {}", args.path.display()))?;
    // Decided once, before the first push, from the arguments and the folder, not from any
    // response: `watch --id X` keeps pushing to X even inside a project.
    let keyed = args.id.is_none() && project_link::find(&path)?.is_some();
    note_outside_project(&path, args.id.as_deref(), false);
    let query = PushQuery {
        title: args.title.clone(),
        tags: tags(&args.tags, &ctx.settings.default_tags),
        entry: args.entry.clone(),
        ..Default::default()
    };
    let first = push_path(ctx, &client, &path, args.id.as_deref(), false, &query)
        .await
        .map_err(|e| hint_outside_project(e, "--id"))?;
    print_line(ctx, &first);
    if !ctx.json {
        eprintln!("Watching {} (Ctrl+C to stop)", args.path.display());
    }
    // A keyed watch pushes by key every time. Otherwise the first document is replaced by id.
    let id = if keyed { None } else { Some(first.id.clone()) };

    let (single, root) = if path.is_file() {
        (
            Some(path.clone()),
            path.parent()
                .map(Path::to_path_buf)
                .unwrap_or_else(|| PathBuf::from(".")),
        )
    } else {
        (None, path.clone())
    };
    let (tx, mut rx) = unbounded_channel();
    let (root_c, single_c) = (root.clone(), single.clone());
    let mut watcher = notify::recommended_watcher(move |res: notify::Result<notify::Event>| {
        let Ok(ev) = res else { return };
        if matches!(ev.kind, EventKind::Access(_)) {
            return;
        }
        if ev
            .paths
            .iter()
            .any(|p| relevant(&root_c, single_c.as_deref(), p))
        {
            let _ = tx.send(());
        }
    })?;
    let mode = if single.is_some() {
        RecursiveMode::NonRecursive
    } else {
        RecursiveMode::Recursive
    };
    watcher
        .watch(&root, mode)
        .with_context(|| format!("watching {}", root.display()))?;

    loop {
        tokio::select! {
            _ = tokio::signal::ctrl_c() => break,
            got = debounce(&mut rx, QUIET) => {
                if got.is_none() {
                    break;
                }
                match push_path(ctx, &client, &path, id.as_deref(), false, &query).await {
                    Ok(r) => print_line(ctx, &r),
                    Err(e) => output::warn(format!("push failed, still watching: {e:#}")),
                }
            }
        }
    }
    Ok(exit::OK)
}

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

    #[tokio::test(start_paused = true)]
    async fn a_burst_of_changes_is_one_push() {
        let (tx, mut rx) = unbounded_channel();
        let sender = tokio::spawn(async move {
            for _ in 0..5 {
                tx.send(()).unwrap();
                tokio::time::sleep(Duration::from_millis(100)).await;
            }
            tokio::time::sleep(Duration::from_secs(5)).await;
            drop(tx);
        });
        let start = tokio::time::Instant::now();
        assert_eq!(debounce(&mut rx, QUIET).await, Some(()));
        let waited = start.elapsed();
        assert!(
            waited >= Duration::from_millis(900) && waited < Duration::from_millis(1000),
            "{waited:?}"
        );
        assert_eq!(debounce(&mut rx, QUIET).await, None, "nothing else arrives");
        sender.await.unwrap();
    }

    #[test]
    fn relevance_rules() {
        let root = Path::new("/r");
        assert!(relevant(root, None, Path::new("/r/index.md")));
        assert!(relevant(root, None, Path::new("/r/a/b.png")));
        assert!(!relevant(root, None, Path::new("/r/.env")));
        assert!(!relevant(root, None, Path::new("/r/.git/index")));
        assert!(!relevant(root, None, Path::new("/r/node_modules/x.js")));
        assert!(!relevant(root, None, Path::new("/elsewhere/x.md")));
        let f = Path::new("/r/doc.md");
        assert!(relevant(root, Some(f), f));
        assert!(!relevant(root, Some(f), Path::new("/r/.doc.md.swp")));
    }
}