benchmark-repository-rs/src/lib.rs

extern crate git2;
extern crate pretty_env_logger;
#[macro_use]
extern crate log;
extern crate indicatif;
extern crate notify;
extern crate tempfile;
extern crate walkdir;

use git2::{Cred, Error, FetchOptions, Index, IndexEntry, IndexTime, Progress, PushOptions, RemoteCallbacks, ResetType, Repository, Signature, TreeBuilder};
use git2::build::{CheckoutBuilder, RepoBuilder};
use std::fs::{File, create_dir_all, read, remove_file};
use std::io::Cursor;
use std::path::{Path, PathBuf};
use std::string::String;
use std::str;
use std::sync::{Arc, Mutex};
use std::sync::mpsc::{channel, Sender, Receiver};
use std::thread::{spawn, JoinHandle};
use indicatif::{ProgressBar, ProgressStyle};
use notify::{raw_watcher, RecommendedWatcher, RawEvent, Op, RecursiveMode, Watcher};
use tempfile::TempDir;
use walkdir::WalkDir;

pub struct BenchmarkRepositoryInner
{
	repository: Repository,
	ssh_user: String,
	user_name: String,
	user_email: String,
}

pub struct BenchmarkRepository
{
	inner: Arc<Mutex<BenchmarkRepositoryInner>>,
	autocommit_channel: (Sender<RawEvent>, Receiver<RawEvent>),
	autocommit_directory: Option<TempDir>,
	autocommit_thread: Option<JoinHandle<()>>,
	autocommit_watcher: Option<RecommendedWatcher>,
	autocommit_locks: Arc<Mutex<usize>>,
}

pub struct TargetPath
{
	pub source: PathBuf,
	pub destination: PathBuf,
}

impl BenchmarkRepository
{
	fn progress_bar_style() -> ProgressStyle
	{
		ProgressStyle::default_bar()
			.template("{bar:74.on_black} {percent:>3} %")
			.progress_chars("█▉▊▋▌▍▎▏  ")
	}

	fn transfer_progress_callback(progress: Progress, progress_bar: &mut ProgressBar) -> bool
	{
		progress_bar.set_length(progress.total_objects() as u64);
		progress_bar.set_position(progress.received_objects() as u64);

		// continue the transfer
		true
	}

	fn push_update_reference_callback(reference: &str, status: Option<&str>) -> Result<(), Error>
	{
		match status
		{
			None => trace!("Reference “{}” pushed successfully to remote “origin”", reference),
			Some(error) => panic!("Couldn’t push reference “{}” to remote “origin”: {}", reference, error),
		};

		Ok(())
	}

	fn checkout_progress_callback(path: Option<&Path>, current: usize, total: usize, progress_bar: &mut ProgressBar)
	{
		progress_bar.set_length(total as u64);
		progress_bar.set_position(current as u64);
	}

	fn reset_origin(&self, remote_url: &str) -> &Self
	{
		let inner = self.inner.lock().unwrap();

		let remote = match inner.repository.find_remote("origin")
		{
			Ok(remote) => remote,
			Err(_) =>
				match inner.repository.remote("origin", remote_url)
				{
					Ok(remote) => remote,
					Err(error) => panic!("Could not reset remote “origin”: {}", error),
				},
		};

		info!("Reset origin to “{}”", remote_url);

		self
	}

	fn fetch_branch(&self, branch_name: &str) -> &Self
	{
		let inner = self.inner.lock().unwrap();

		let mut progress_bar = ProgressBar::new(0);
		progress_bar.set_style(BenchmarkRepository::progress_bar_style());

		{
			let mut remote_callbacks = RemoteCallbacks::new();
			remote_callbacks.credentials(
				|_, _, _|
				{
					match Cred::ssh_key_from_agent(&inner.ssh_user)
					{
						Ok(credentials) => Ok(credentials),
						Err(error) => panic!("could not retrieve key pair for SSH authentication as user “{}”: {}", inner.ssh_user, error),
					}
				});

			remote_callbacks.transfer_progress(
				|progress| BenchmarkRepository::transfer_progress_callback(progress, &mut progress_bar));

			let mut fetch_options = FetchOptions::new();
			fetch_options.remote_callbacks(remote_callbacks);

			let mut origin = inner.repository.find_remote("origin").expect("could not find remote “origin”");

			info!("Updating branch “{}”", branch_name);

			if let Err(error) = origin.fetch(&[branch_name], Some(&mut fetch_options), None)
			{
				panic!("failed to fetch branch “{}” from remote “origin”: {}", branch_name, error);
			}
		}

		progress_bar.finish_and_clear();

		trace!("Branch “{}” is up-to-date", branch_name);

		self
	}

	fn init(base_path: &Path) -> Repository
	{
		let repository = match Repository::init_bare(base_path)
		{
			Ok(repository) => repository,
			Err(error) => panic!("failed to initialize Git repository in “{}”: {}", base_path.display(), error),
		};

		info!("Initialized Git repository in “{}”", base_path.display());

		repository
	}

	pub fn new(remote_url: &str, base_path: &Path, ssh_user: &str, user_name: &str, user_email: &str) -> BenchmarkRepository
	{
		let repository = match Repository::open(base_path)
		{
			Ok(repository) =>
			{
				info!("Using existing Git repository");

				repository
			},
			Err(_) => BenchmarkRepository::init(base_path),
		};

		let benchmark_repository_inner =
			BenchmarkRepositoryInner
			{
				repository: repository,
				ssh_user: ssh_user.to_string(),
				user_name: user_name.to_string(),
				user_email: user_email.to_string(),
			};

		let benchmark_repository =
			BenchmarkRepository
			{
				inner: Arc::new(Mutex::new(benchmark_repository_inner)),
				autocommit_channel: channel(),
				autocommit_directory: None,
				autocommit_thread: None,
				autocommit_watcher: None,
				autocommit_locks: Arc::new(Mutex::new(0)),
			};

		benchmark_repository
			.reset_origin(remote_url)
			.fetch_branch("config")
			.fetch_branch("results")
			.fetch_branch("status");

		benchmark_repository
	}

	pub fn read_file_as_index_entry(inner: &BenchmarkRepositoryInner, file_path: &Path, result_file_path: &Path) -> IndexEntry
	{
		// create a new blob with the file contents
		let object_id = match inner.repository.blob_path(file_path)
		{
			Ok(object_id) => object_id,
			Err(error) => panic!("Could not write blob for “{}”: {}", file_path.display(), error),
		};

		info!("Created object “{}” from “{}”", object_id, file_path.display());

		IndexEntry
		{
			ctime: IndexTime::new(0, 0),
			mtime: IndexTime::new(0, 0),
			dev: 0,
			ino: 0,
			mode: 0o100644,
			uid: 0,
			gid: 0,
			file_size: 0,
			id: object_id,
			flags: 0,
			flags_extended: 0,
			path: result_file_path.to_string_lossy().as_bytes().to_owned(),
		}
	}

	pub fn commit_directory(inner: &BenchmarkRepositoryInner, directory_path: &TargetPath, branch_name: &str)
	{
		let mut file_paths = vec![];

		for entry in WalkDir::new(&directory_path.source)
		{
			let entry = entry.unwrap();

			if entry.path().file_name().unwrap() == ".lock" || entry.file_type().is_dir()
			{
				continue;
			}

			let relative_path = entry.path().strip_prefix(&directory_path.source).unwrap();

			trace!("Adding “{}” (from “{}”) to autocommit list", relative_path.display(), entry.path().display());

			file_paths.push(TargetPath{source: entry.path().to_owned(), destination: directory_path.destination.join(&relative_path)});
		}

		BenchmarkRepository::commit_files(&inner, &file_paths, branch_name);
	}

	pub fn commit_files(inner: &BenchmarkRepositoryInner, file_paths: &[TargetPath], branch_name: &str)
	{
		let tip_reference_name = format!("refs/remotes/origin/{}", branch_name);
		let tip_reference = match inner.repository.find_reference(&tip_reference_name)
		{
			Ok(value) => value,
			Err(error) => panic!("Could not find reference “{}”: {}", tip_reference_name, error),
		};

		let parent_tree = match tip_reference.peel_to_tree()
		{
			Ok(value) => value,
			Err(error) => panic!("Could not peel reference to tree: {}", error),
		};

		let mut index = match Index::new()
		{
			Ok(value) => value,
			Err(error) => panic!("Could not create index: {}", error),
		};

		match index.read_tree(&parent_tree)
		{
			Ok(value) => value,
			Err(error) => panic!("Could not read parent tree into index: {}", error),
		};

		for target_path in file_paths
		{
			let index_entry = BenchmarkRepository::read_file_as_index_entry(inner, &target_path.source, &target_path.destination);

			if let Err(error) = index.add(&index_entry)
			{
				panic!("Could not add index entry for “{}”: {}", target_path.destination.display(), error);
			}
		}

		let tree_object_id = match index.write_tree_to(&inner.repository)
		{
			Ok(value) => value,
			Err(error) => panic!("Could not write index to tree: {}", error),
		};

		let tree = match inner.repository.find_tree(tree_object_id)
		{
			Ok(value) => value,
			Err(error) => panic!("Could obtain tree: {}", error),
		};

		info!("Created tree object “{}”", tree_object_id);

		let signature = Signature::now(&inner.user_name, &inner.user_email).expect("Could not create signature");
		let message = format!("Add files");

		let parent = match tip_reference.peel_to_commit()
		{
			Ok(value) => value,
			Err(error) => panic!("Could not peel reference: {}", error),
		};

		let commit_id = match inner.repository.commit(Some(&tip_reference_name), &signature, &signature, &message, &tree, &[&parent])
		{
			Ok(value) => value,
			Err(error) => panic!("Could not write commit: {}", error),
		};

		let push_refspec = format!("refs/remotes/origin/{}:refs/heads/{}", branch_name, branch_name);

		trace!("Created commit “{}”, using refspec “{}”", commit_id, push_refspec);

		let mut remote_callbacks = RemoteCallbacks::new();
		remote_callbacks.credentials(
			|_, _, _|
			{
				match Cred::ssh_key_from_agent(&inner.ssh_user)
				{
					Ok(value) => Ok(value),
					Err(error) => panic!("could not retrieve key pair for SSH authentication as user “{}”: {}", inner.ssh_user, error),
				}
			});

		remote_callbacks.push_update_reference(
			|reference, status| BenchmarkRepository::push_update_reference_callback(reference, status));

		let mut push_options = PushOptions::new();
		push_options.remote_callbacks(remote_callbacks);

		let mut remote = inner.repository.find_remote("origin").expect("");
		remote.push(&[&push_refspec], Some(&mut push_options)).expect("couldn’t push");
	}

	pub fn file_exists(&self, file_path: &Path, branch_name: &str) -> bool
	{
		let inner = self.inner.lock().unwrap();

		let tip_reference_name = format!("refs/remotes/origin/{}", branch_name);
		let tip_reference = match inner.repository.find_reference(&tip_reference_name)
		{
			Ok(value) => value,
			Err(error) => panic!("Could not find reference “{}”: {}", tip_reference_name, error),
		};

		let tree = match tip_reference.peel_to_tree()
		{
			Ok(value) => value,
			Err(error) => panic!("Could not peel reference to tree: {}", error),
		};

		let object_id = match tree.get_path(file_path)
		{
			Ok(tree_entry) => tree_entry.id(),
			Err(error) => return false,
		};

		let blob = match inner.repository.find_blob(object_id)
		{
			Ok(blob) => blob,
			Err(error) => return false,
		};

		true
	}

	pub fn read_file(&self, file_path: &Path, branch_name: &str) -> Option<String>
	{
		let inner = self.inner.lock().unwrap();

		let tip_reference_name = format!("refs/remotes/origin/{}", branch_name);
		let tip_reference = match inner.repository.find_reference(&tip_reference_name)
		{
			Ok(value) => value,
			Err(error) => panic!("Could not find reference “{}”: {}", tip_reference_name, error),
		};

		let tree = match tip_reference.peel_to_tree()
		{
			Ok(value) => value,
			Err(error) => panic!("Could not peel reference to tree: {}", error),
		};

		let object_id = match tree.get_path(file_path)
		{
			Ok(tree_entry) => tree_entry.id(),
			Err(error) => return None,
		};

		let blob = match inner.repository.find_blob(object_id)
		{
			Ok(blob) => blob,
			Err(error) => return None,
		};

		let content = match std::str::from_utf8(blob.content())
		{
			Ok(content) => content,
			Err(error) => panic!("Could not interpret file “{}” as UTF-8: {}", file_path.display(), error),
		};

		Some(content.to_owned())
	}

	fn prepare_autocommit_directory(&mut self)
	{
		match self.autocommit_thread
		{
			Some(_) => (),
			None =>
			{
				let tmp_dir = match TempDir::new()
				{
					Ok(value) => value,
					Err(error) => panic!("Could not create autocommit directory: {}", error),
				};

				trace!("Created temporary autocommit directory in {}", tmp_dir.path().display());

				let lock_file_path = tmp_dir.path().join(".lock");

				if let Err(error) = File::create(&lock_file_path)
				{
					panic!("Could not create lock file “{}”: {}", lock_file_path.display(), error);
				}

				let (sender, receiver) = channel();

				let mut watcher = match raw_watcher(sender)
				{
					Ok(value) => value,
					Err(error) => panic!("Could not create filesystem watcher: {}", error),
				};

				watcher.watch(&tmp_dir, RecursiveMode::Recursive);

				self.autocommit_watcher = Some(watcher);
				self.autocommit_directory = Some(tmp_dir);

				let autocommit_base_path = self.autocommit_directory.as_ref().unwrap().path().to_owned();
				let inner = self.inner.clone();
				let autocommit_locks = self.autocommit_locks.clone();

				let thread = spawn(
					move ||
					{
						trace!("Autocommit thread running");

						let mut active = true;

						loop
						{
							match receiver.recv()
							{
								Ok(RawEvent{path: Some(path), op: Ok(Op::REMOVE), cookie: _}) =>
								{
									if path == lock_file_path
									{
										trace!("Waiting for remaining autocommit locks to be released");
										active = false;
									}
									else
									{
										trace!("Received “remove” event for path “{}”", path.display());

										let stripped_path = match path.strip_prefix(&autocommit_base_path)
										{
											Ok(value) => value,
											Err(error) => panic!("Cannot handle “remove” event for path “{}”", path.display()),
										};

										let mut components = stripped_path.components();
										let branch_name = components.next().unwrap();
										let result_path = stripped_path.strip_prefix(&branch_name).unwrap().parent().unwrap();

										info!("Autocommiting “{}”", result_path.display());

										let inner = inner.lock().unwrap();

										BenchmarkRepository::commit_directory(&inner, &TargetPath{source: path.parent().unwrap().to_owned(), destination: result_path.to_owned()}, &branch_name.as_os_str().to_str().unwrap());

										let mut autocommit_locks = autocommit_locks.lock().unwrap();
										*autocommit_locks -= 1;
									}
								},
								Err(error) => panic!("Error handling notify event: {}", error),
								_ => (),
							}

							let mut autocommit_locks = autocommit_locks.lock().unwrap();

							if *autocommit_locks == 0
							{
								break;
							}
						}

						trace!("Autocommit thread finished");
					});

				self.autocommit_thread = Some(thread);
			}
		}
	}

	pub fn create_autocommit_directory(&mut self, directory_path: &Path, branch_name: &str) -> Result<PathBuf, String>
	{
		self.prepare_autocommit_directory();

		let result_directory_path = self.autocommit_directory.as_ref().unwrap().path().join(branch_name).join(directory_path);

		match create_dir_all(&result_directory_path)
		{
			Ok(_) => (),
			Err(error) => panic!("Could not create result directory: {}", error),
		}

		{
			let lock_file_path = result_directory_path.join(".lock");

			if lock_file_path.exists()
			{
				panic!("Autocommit directory “{}” already locked", lock_file_path.display());
			}

			if let Err(error) = File::create(&lock_file_path)
			{
				panic!("Could not create temporary file “{}”: {}", lock_file_path.display(), error);
			}
		}

		let mut autocommit_locks = self.autocommit_locks.lock().unwrap();
		*autocommit_locks += 1;

		Ok(result_directory_path)
	}

	pub fn wait_for_autocommit_thread(mut self)
	{
		if let None = self.autocommit_thread
		{
			panic!("No autocommit thread started");
		}

		let lock_file_path = self.autocommit_directory.as_ref().unwrap().path().join(".lock");

		if let Err(error) = remove_file(&lock_file_path)
		{
			panic!("Could not remove lock file “{}”: {}", lock_file_path.display(), error);
		}

		info!("Removed lock file");

		let autocommit_thread = self.autocommit_thread;

		if let Err(RecvError) = autocommit_thread.unwrap().join()
		{
			panic!("Could not join autocommit thread");
		}

		self.autocommit_thread = None;
	}
}

#[cfg(test)]
mod tests
{
}