semora-cli 1.0.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::{Client, PushQuery};
use crate::commands::push::{hint_outside_project, note_outside_project, push_path, tags};
use crate::config::KeySource;
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"
    })
}

/// `[12:00:00] r3 pushed  URL`, or `unchanged` when the push made no revision.
fn push_line(stamp: &str, r: &semora_proto::PushResponse) -> String {
    format!(
        "[{stamp}] r{} {}  {}",
        r.rev,
        if r.unchanged { "unchanged" } else { "pushed" },
        r.path_url.as_deref().unwrap_or(&r.url)
    )
}

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!("{}", push_line(&now.format(&fmt).unwrap_or_default(), r));
    }
    for b in &r.broken_links {
        output::warn(format!("broken link in {}: {}", b.from, b.href));
    }
}

/// A client with the key that is stored now, when the push failed because the server refused
/// the key and another one has been stored since. `semora login` revokes the key it replaces,
/// so a watch that started before the login holds a dead key.
fn client_after_login(ctx: &Ctx, client: &Client, failure: &anyhow::Error) -> Option<Client> {
    if exit::code_of(failure) != exit::AUTH || ctx.settings.key_source != KeySource::Stored {
        return None;
    }
    let key = ctx.store.get(&client.origin())?;
    if key == client.key() {
        return None;
    }
    let mut settings = ctx.settings.clone();
    settings.api_key = Some(key);
    Client::new(&settings).ok()
}

pub async fn run(ctx: &Ctx, args: WatchArgs) -> anyhow::Result<i32> {
    let mut 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 (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()))?;

    // The watcher is armed before the first push, so an edit made while that push runs, or the
    // instant its line prints, is queued for the loop below instead of lost. If the first push
    // already carried that edit, the queued change makes one more push that reports `unchanged`.
    let first = push_path(
        ctx,
        &client,
        &path,
        args.id.as_deref(),
        false,
        false,
        &query,
    )
    .await
    .map_err(|e| hint_outside_project(e, "--id"))?;
    print_line(ctx, &first);
    if !ctx.json {
        anstream::eprintln!(
            "{DIM}Watching {} (Ctrl+C to stop){DIM:#}",
            args.path.display(),
            DIM = output::DIM
        );
    }
    // 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()) };

    loop {
        tokio::select! {
            _ = tokio::signal::ctrl_c() => break,
            got = debounce(&mut rx, QUIET) => {
                if got.is_none() {
                    break;
                }
                let mut pushed = push_path(ctx, &client, &path, id.as_deref(), false, false, &query).await;
                if let Err(e) = &pushed
                    && let Some(fresh) = client_after_login(ctx, &client, e)
                {
                    client = fresh;
                    pushed = push_path(ctx, &client, &path, id.as_deref(), false, false, &query).await;
                }
                match pushed {
                    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 the_watch_line_says_when_a_push_made_no_revision() {
        let mut r = semora_proto::PushResponse {
            id: "k3F9xQ2mLp".into(),
            rev: 3,
            kind: semora_proto::DocKind::Md,
            url: "https://s.example/d/k3F9xQ2mLp/".into(),
            ws_url: "w".into(),
            files: 1,
            title: "T".into(),
            unchanged: false,
            broken_links: vec![],
            warnings: vec![],
            project: None,
            path: None,
            path_url: None,
        };
        assert_eq!(
            push_line("12:00:00", &r),
            "[12:00:00] r3 pushed  https://s.example/d/k3F9xQ2mLp/"
        );
        r.unchanged = true;
        assert_eq!(
            push_line("12:00:00", &r),
            "[12:00:00] r3 unchanged  https://s.example/d/k3F9xQ2mLp/"
        );
    }

    #[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")));
    }
}