watchers 1.4.0

Git integrated file watcher
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150

use crate::{
    config::{Config, get_watchers_config_dir},
    debouncer::Debouncer,
    file_utils::was_modification,
    git::{EventContext, handle_event},
    systemd::SystemdContext,
};

use anyhow::Result;
use inquire::Text;
use log::debug;
use notify::{Event, RecursiveMode};
use std::{fs, path::PathBuf, sync::mpsc, time::Duration};

pub struct Watcher<'a, F> {
    pub config: &'a Config,
    pub debouncer: Debouncer<F>,
}

impl<'a, F> Watcher<'a, F>
where
    F: FnMut(EventContext) + Send + 'static,
{
    pub fn new(config: &'a Config, debouncer_cb: F) -> Self {
        let debouncer = Debouncer::new(
            debouncer_cb,
            Duration::from_secs(config.commit_delay_secs as u64),
        );
        Self { config, debouncer }
    }

    pub fn trigger_debouncer(&mut self) {
        debug!("triggering debouncer");
        let context = EventContext {
            repo_path: self.config.watch_dir.clone(),
            config: self.config.clone(),
        };
        self.debouncer.on_event(context);
    }
}

pub fn watch_repo<F>(watcher: &mut Watcher<F>) -> Result<()>
where
    F: FnMut(EventContext) + Send + 'static,
{
    let (tx, rx) = mpsc::channel::<notify::Result<Event>>();
    let mut fs_watcher = notify::recommended_watcher(tx)?;
    notify::Watcher::watch(
        &mut fs_watcher,
        &watcher.config.watch_dir,
        RecursiveMode::Recursive,
    )?;

    loop {
        match rx.recv() {
            Err(e) => println!("watch error: {:?}", e),
            Ok(ev) => {
                if let Ok(ev) = ev
                    && was_modification(&ev)
                // TODO: && not_ignored(&ev)
                {
                    debug!("got modification: {:?}", ev);
                    watcher.trigger_debouncer();
                }
            }
        }
    }
}

fn get_watcher_config(name: &str) -> Result<Config> {
    let path = Config::get_watcher_config_path(name)?;
    anyhow::ensure!(path.is_file(), "Could not find config for '{}'", name);
    Config::from_file(path)
}

pub async fn start_watcher(name: &str) -> Result<()> {
    let systemd_ctx = SystemdContext::new().await?;

    systemd_ctx.start_and_enable_service(name).await?;
    Ok(())
}

pub async fn create_watcher(name: &str) -> Result<()> {
    let path_input = Text::new("Path to directory to watch:").prompt()?;
    let path = PathBuf::from(shellexpand::tilde(&path_input).to_string());

    anyhow::ensure!(
        path.is_dir(),
        "Path '{}' is not a directory",
        path.display()
    );

    let config = Config::new(name, path);
    fs::write(Config::get_watcher_config_path(name)?, config.dump()?)?;

    let systemd_ctx = SystemdContext::new().await?;
    systemd_ctx.start_and_enable_service(name).await?;

    Ok(())
}

pub async fn stop_watcher(name: &str) -> Result<()> {
    let systemd_ctx = SystemdContext::new().await?;
    systemd_ctx.stop_and_disable_service(name).await?;

    Ok(())
}

pub fn delete_watcher(name: &str) -> Result<()> {
    let config_path = Config::get_watcher_config_path(name)?;
    anyhow::ensure!(config_path.is_file(), "Couldn't find watcher '{}'", name);
    fs::remove_file(config_path)?;
    Ok(())
}

pub fn list_watchers() -> Result<()> {
    println!("Watchers:");
    let config_dir = get_watchers_config_dir();
    for file in fs::read_dir(config_dir)? {
        let path = file?.path();
        if !path.is_file() {
            continue;
        }

        println!(
            "  {}",
            path.file_stem()
                .and_then(|s| s.to_str())
                .ok_or_else(|| anyhow::anyhow!("Failed to read config directory"))?
        );
    }

    Ok(())
}

pub async fn run_daemon(name: &str) -> Result<()> {
    let config = get_watcher_config(name)?;
    debug!(
        "Config:\n{}",
        config.dump().unwrap_or("failed to read config".to_string())
    );

    let mut watcher = Watcher::new(&config, |context| {
        handle_event(context);
    });

    watch_repo(&mut watcher)?;

    anyhow::bail!("Should never finish watching");
}