Queue autosaves

1 files changed, 93 insertions, 8 deletions

diff --git a/src/bin/git-autosave-daemon.rs b/src/bin/git-autosave-daemon.rs
index 80edb34..d8c23fc 100644
--- a/src/bin/git-autosave-daemon.rs
+++ b/src/bin/git-autosave-daemon.rs
@@ -1,4 +1,8 @@
-use std::{collections::HashSet, time::Duration};
+use std::collections::HashSet;
+use std::path::PathBuf;
+use std::sync::mpsc::Receiver;
+use std::sync::mpsc::Sender;
+use std::time::Duration;
 
 use auth_git2::GitAuthenticator;
 use git_autosave::{Config, authenticate::Inquirer, commit_autosave, push_autosaves};
@@ -12,7 +16,67 @@ struct ConfigWatcher<Cache: FileIdCache + 'static> {
 	repo_watcher: &'static mut Debouncer<INotifyWatcher, Cache>,
 }
 
-struct Watcher(&'static Mutex<Config>);
+struct RepoWatcher {
+	config: &'static Mutex<Config>,
+	push_queue: Sender<PathBuf>,
+}
+
+fn push_queue(
+	consumer: Receiver<PathBuf>,
+	config: &'static Mutex<Config>,
+	mut key: ThreadKey,
+) -> ! {
+	let mut queue = HashSet::new();
+
+	loop {
+		while ping::new("8.8.8.8".parse().unwrap())
+			.socket_type(ping::SocketType::DGRAM)
+			.timeout(Duration::from_secs(60))
+			.ttl(128)
+			.send()
+			.is_err()
+		{
+			let mut added_item = false;
+			while let Ok(workdir) = consumer.try_recv() {
+				added_item = true;
+				queue.insert(workdir);
+			}
+			if added_item {
+				log::warn!(
+					"There is no Internet connection, so pushes to the repository have been queued"
+				);
+			}
+			std::thread::yield_now();
+		}
+
+		while let Ok(workdir) = consumer.try_recv() {
+			queue.insert(workdir);
+		}
+
+		config.scoped_lock(&mut key, |config| {
+			for workdir in &queue {
+				log::info!("Pushing {}...", workdir.to_string_lossy());
+				let Ok(repository) = Repository::open(workdir) else {
+					log::error!("Failed to open repository: {}", workdir.to_string_lossy());
+					continue;
+				};
+				let Ok(gitconfig) = repository.config() else {
+					log::error!("Failed to load gitconfig for repository: {:?}", workdir);
+					return;
+				};
+				let auth = GitAuthenticator::new().set_prompter(Inquirer(&*config));
+				let mut callbacks = RemoteCallbacks::new();
+				callbacks.credentials(auth.credentials(&gitconfig));
+				if let Err(e) = push_autosaves(&repository, callbacks) {
+					log::error!("Failed to push autosaves: {e}");
+				}
+
+				log::info!("Successfully pushed {}", workdir.to_string_lossy());
+			}
+		});
+		queue.clear();
+	}
+}
 
 impl<Cache: FileIdCache + Send + 'static> DebounceEventHandler for ConfigWatcher<Cache> {
 	fn handle_event(&mut self, events: DebounceEventResult) {
@@ -68,7 +132,7 @@ impl<Cache: FileIdCache + Send + 'static> DebounceEventHandler for ConfigWatcher
 	}
 }
 
-impl DebounceEventHandler for Watcher {
+impl DebounceEventHandler for RepoWatcher {
 	fn handle_event(&mut self, events: DebounceEventResult) {
 		let Some(mut key) = ThreadKey::get() else {
 			log::error!("Failed to acquire thread key when autosaving repository. This is a bug!");
@@ -134,7 +198,7 @@ impl DebounceEventHandler for Watcher {
 			workdirs_to_autosave.insert(workdir.to_path_buf());
 			repositories_to_autosave.push(repository);
 		}
-		self.0.scoped_lock(&mut key, |config| {
+		self.config.scoped_lock(&mut key, |config| {
 			for repository in repositories_to_autosave {
 				let workdir = repository
 					.workdir()
@@ -152,8 +216,11 @@ impl DebounceEventHandler for Watcher {
 				if let Err(e) = commit_autosave(&repository) {
 					log::error!("Failed to commit autosave: {e}");
 				}
-				if let Err(e) = push_autosaves(&repository, callbacks) {
-					log::error!("Failed to push autosaves: {e}");
+				if let Err(e) = self
+					.push_queue
+					.send(repository.workdir().unwrap().to_path_buf())
+				{
+					log::error!("Failed to add repository to push queue: {e}");
 				}
 
 				log::info!("Successfully autosaved {:?}", workdir);
@@ -163,18 +230,36 @@ impl DebounceEventHandler for Watcher {
 }
 
 fn main() -> Result<(), anyhow::Error> {
-	let key = ThreadKey::get().expect("Could not get ThreadKey on startup. This is a bug!");
+	let mut key = ThreadKey::get().expect("Could not get ThreadKey on startup. This is a bug!");
 	colog::init();
 
 	log::info!("Loading autosave config...");
 	let config: &'static Mutex<Config> = Box::leak(Box::new(Mutex::new(Config::load()?)));
 	log::info!("Loaded autosave config");
 
+	log::info!("Starting push queue...");
+	let (sender, receiver) = std::sync::mpsc::channel();
+	std::thread::spawn(move || {
+		let key = ThreadKey::get().unwrap();
+		push_queue(receiver, config, key);
+	});
+	config.scoped_lock(&mut key, |config| {
+		for repository in config.repositories() {
+			if let Err(e) = sender.send(repository.clone()) {
+				log::error!("Failed to queue {}: {e}", repository.to_string_lossy());
+			}
+		}
+	});
+	log::info!("Started push queue");
+
 	log::info!("Starting repository watcher...");
 	let repo_watcher = Box::leak(Box::new(notify_debouncer_full::new_debouncer(
 		Duration::from_secs(5),
 		None,
-		Watcher(config),
+		RepoWatcher {
+			config,
+			push_queue: sender,
+		},
 	)?));
 	config.scoped_lock(key, |config| {
 		log::info!("Adding repositories to watch...");