use super::auto_push::push_once;
use super::auto_push_single_flight::PushGit;
use super::config_snapshot::read_config_snapshot;
use super::lfs_pre_push::uses_lfs;
use super::maintenance_run::{AutoMaintenanceOutcome, run_auto_maintenance_with};
use super::service_messages::push_failed_notice;
use super::private_storage::{create_private_directory_all, write_private_file_unsynced};
use super::service_review::{BUDGET, options_for, review_item};
use super::transaction_git::GitContext;
use super::work_item::{DONE, FAILED, WORK_DIRECTORY, WorkItem, WorkKind, claim_next_of, requeue_expired, settle};
use std::path::{Path, PathBuf};
use std::time::{Duration, SystemTime};
pub const SERVICE_KINDS: [WorkKind; 3] = [WorkKind::Push, WorkKind::Maintenance, WorkKind::Review];
pub const LEASE: Duration = Duration::from_secs(30 * 60);
pub const MAX_TRIES: u32 = 3;
pub const NOTICE_DIRECTORY: &str = "cli-git/notices";
#[derive(Clone, Debug, Eq, PartialEq)]
pub enum Verdict {
Done,
Failed(String),
}
pub trait WorkHandlers {
fn push(&self, items: &[WorkItem]) -> Verdict;
fn maintenance(&self, item: &WorkItem) -> Verdict;
fn review(&self, item: &WorkItem) -> Verdict;
}
pub fn work_directory(common_dir: &Path) -> PathBuf {
return common_dir.join(WORK_DIRECTORY);
}
pub fn notice_directory(common_dir: &Path) -> PathBuf {
return common_dir.join(NOTICE_DIRECTORY);
}
pub fn is_acceptable(item: &WorkItem, common_dir: &Path) -> bool {
return Path::new(item.repository.as_str()).is_absolute()
&& Path::new(item.repository.as_str()).is_dir()
&& Path::new(item.git_dir.as_str()) == common_dir
&& item.branch_ref.starts_with("refs/")
&& !item.branch_ref.contains(['\0', '\n', ' '])
&& item.oid.len() >= 40
&& item.oid.bytes().all(|byte| { return byte.is_ascii_hexdigit(); });
}
pub fn write_notice(common_dir: &Path, id: &str, text: &str) -> Result<(), String> {
let directory: PathBuf = notice_directory(common_dir);
create_private_directory_all(directory.as_path()).map_err(|error| { return format!("creating the notice directory failed: {error}"); })?;
let path: PathBuf = directory.join(format!("{id}.txt"));
let _removed = std::fs::remove_file(path.as_path());
return write_private_file_unsynced(path.as_path(), text.as_bytes()).map_err(|error| { return format!("writing the notice failed: {error}"); });
}
pub fn take_notices(common_dir: &Path) -> Vec<String> {
let Ok(listing) = std::fs::read_dir(notice_directory(common_dir)) else { return Vec::new() };
let mut paths: Vec<PathBuf> = listing.flatten().map(|entry| { return entry.path(); }).collect();
paths.sort();
let mut texts: Vec<String> = Vec::new();
for path in paths {
if let Ok(text) = std::fs::read_to_string(path.as_path()) {
texts.push(text);
}
let _removed = std::fs::remove_file(path.as_path());
}
return texts;
}
fn settle_all(work_dir: &Path, claimed: &[super::work_item::Claimed], state: &str) {
for item in claimed {
if let Err(reason) = settle(work_dir, item, state) {
eprintln!("cli-git-service: {reason}");
}
}
}
pub fn drain(common_dir: &Path, handlers: &dyn WorkHandlers, now: SystemTime) {
let work_dir: PathBuf = work_directory(common_dir);
let _requeued = requeue_expired(work_dir.as_path(), now, LEASE, MAX_TRIES);
loop {
let mut claimed: Vec<super::work_item::Claimed> = Vec::new();
loop {
match claim_next_of(work_dir.as_path(), now, &SERVICE_KINDS) {
Ok(Some(item)) => claimed.push(item),
Ok(None) => break,
Err(reason) => {
eprintln!("cli-git-service: {reason}");
break;
}
}
}
if claimed.is_empty() {
return;
}
let (acceptable, refused): (Vec<_>, Vec<_>) = claimed.into_iter().partition(|item| { return is_acceptable(&item.item, common_dir); });
settle_all(work_dir.as_path(), &refused, FAILED);
let (pushes, rest): (Vec<_>, Vec<_>) = acceptable.into_iter().partition(|item| { return item.item.kind == WorkKind::Push; });
let (reviews, upkeep): (Vec<_>, Vec<_>) = rest.into_iter().partition(|item| { return item.item.kind == WorkKind::Review; });
let mut groups: Vec<((String, String), Vec<super::work_item::Claimed>)> = Vec::new();
for item in pushes {
let key = (item.item.repository.clone(), item.item.branch_ref.clone());
match groups.iter_mut().find(|(existing, _)| { return *existing == key; }) {
Some((_, members)) => members.push(item),
None => groups.push((key, vec![item])),
}
}
for (_, members) in groups {
let items: Vec<WorkItem> = members.iter().map(|member| { return member.item.clone(); }).collect();
match handlers.push(items.as_slice()) {
Verdict::Done => settle_all(work_dir.as_path(), &members, DONE),
Verdict::Failed(text) => {
settle_all(work_dir.as_path(), &members, FAILED);
if let Some(newest) = members.last() {
let notice: String = push_failed_notice(newest.item.oid.as_str(), text.as_str());
let _noticed = write_notice(common_dir, newest.id.as_str(), notice.as_str());
}
}
}
}
let mut upkept: Vec<String> = Vec::new();
for item in upkeep {
if !upkept.contains(&item.item.repository) {
upkept.push(item.item.repository.clone());
if let Verdict::Failed(reason) = handlers.maintenance(&item.item) {
eprintln!("cli-git-service: automatic maintenance did not run: {reason}");
}
}
settle_all(work_dir.as_path(), std::slice::from_ref(&item), DONE);
}
for item in reviews {
match handlers.review(&item.item) {
Verdict::Done => settle_all(work_dir.as_path(), std::slice::from_ref(&item), DONE),
Verdict::Failed(reason) => {
eprintln!("cli-git-service: the review of {} failed: {reason}", item.item.oid);
settle_all(work_dir.as_path(), std::slice::from_ref(&item), FAILED);
}
}
}
}
}
pub struct GitHandlers {
pub context: GitContext,
}
impl GitHandlers {
fn push_git(&self, item: &WorkItem) -> PushGit {
let cwd: PathBuf = PathBuf::from(item.repository.as_str());
return PushGit {
real_git: self.context.real_git.clone(),
lfs: read_config_snapshot(&self.context, cwd.as_path()).as_ref().is_none_or(uses_lfs),
cwd,
environment: std::env::vars_os().collect(),
};
}
}
impl WorkHandlers for GitHandlers {
fn push(&self, items: &[WorkItem]) -> Verdict {
let Some(newest) = items.last() else { return Verdict::Done };
let git: PushGit = self.push_git(newest);
let mut output: Vec<u8> = Vec::new();
return match push_once(&git, newest.oid.as_str(), &mut output) {
Ok(result) if result.exit_code == 0 => Verdict::Done,
Ok(result) => Verdict::Failed(result.shown),
Err(error) => Verdict::Failed(format!("{error}")),
};
}
fn maintenance(&self, item: &WorkItem) -> Verdict {
return match run_auto_maintenance_with(&self.context, Path::new(item.repository.as_str()), None) {
Ok(AutoMaintenanceOutcome::Exited(code)) if code != 0 => Verdict::Failed(format!("it exited with status {code}")),
Ok(_) => Verdict::Done,
Err(error) => Verdict::Failed(format!("{error:?}")),
};
}
fn review(&self, item: &WorkItem) -> Verdict {
return review_item(&self.context, item, options_for(item, &|name| { return std::env::var(name).ok(); }), BUDGET);
}
}
#[cfg(test)]
#[path = "service_work_tests.rs"]
mod tests;