2ae9196
ZhenYi
feat(lib/git): add HTTP handler tree with routes and errors across analysis, archives, blobs, branches, commits, conflicts, content, diffs, hooks, lfs, logs, maintenance, packs, patches, pipelines, refs, remotes, repos, signatures, submodules, tags, trees
2ae9196
use std::path::PathBuf;
use std::sync::Arc;
use cache::CacheClient;
use config::git::SyncSettings;
use export::AppDatabase;
use gitbaby::GitBaby;
use models::repo::repos;
use sea_orm::{DatabaseTransaction, TransactionTrait};
use tokio::sync::watch;
use uuid::Uuid;
use crate::context::facade::GitContext;
use crate::sync::branches;
use crate::sync::commits::{CommitData, collect_new_commits_for_sync, sync_commits};
use crate::sync::error::{SyncError, SyncResult};
use crate::sync::fsck;
use crate::sync::languages;
use crate::sync::lock::{self, SyncLock};
use crate::sync::stats;
use crate::sync::tags;
struct HeartbeatGuard {
handle: tokio::task::JoinHandle<()>,
}
impl Drop for HeartbeatGuard {
fn drop(&mut self) {
if !self.handle.is_finished() {
self.handle.abort();
}
}
}
#[derive(Clone)]
pub struct RepoSync {
db: Arc<AppDatabase>,
repo: repos::Model,
baby: GitBaby,
redis: export::AppRedis,
#[allow(dead_code)]
cache: Arc<CacheClient>,
repo_root: PathBuf,
settings: SyncSettings,
}
impl RepoSync {
pub fn new(
db: Arc<AppDatabase>,
redis: export::AppRedis,
cache: Arc<CacheClient>,
repo: repos::Model,
repo_root: PathBuf,
settings: SyncSettings,
) -> Self {
let ctx = GitContext {
model: repo.clone(),
cache: cache.clone(),
database: db.clone(),
repo_root: repo_root.clone(),
objects_root: repo_root.clone(),
};
Self {
db,
repo,
baby: GitBaby::new(Arc::new(ctx)),
redis,
cache,
repo_root,
settings,
}
}
#[must_use]
pub fn repo_id(&self) -> Uuid {
self.repo.id
}
#[must_use]
pub fn repo_dir(&self) -> PathBuf {
crate::context::facade::repo_dir_for(&self.repo_root, self.repo.id)
}
pub async fn sync(&self) -> SyncResult<()> {
let ttl = self.settings.lock_ttl_secs;
let lock = match lock::acquire_lock(&self.redis, self.repo.id, ttl).await? {
Some(l) => Arc::new(l),
None => return Err(SyncError::Locked),
};
let (tx, mut rx) = watch::channel(false);
let hb_lock = lock.clone();
let heartbeat = tokio::spawn(async move {
let interval = (ttl / 3).max(1);
loop {
tokio::time::sleep(std::time::Duration::from_secs(interval)).await;
match hb_lock.refresh(ttl).await {
Ok(true) => {}
Ok(false) | Err(_) => {
let _ = tx.send(true);
break;
}
}
}
});
let heartbeat = HeartbeatGuard { handle: heartbeat };
let result = tokio::select! {
r = self.sync_work() => r,
_ = rx.changed() => Err(SyncError::Locked),
};
let is_locked_failure = matches!(result.as_ref(), Err(SyncError::Locked));
if !is_locked_failure {
let _ = lock.release().await;
}
drop(heartbeat);
result
}
pub async fn sync_work(&self) -> SyncResult<()> {
let baby = self.baby.clone();
let new_commits: Vec<CommitData> = collect_new_commits_for_sync(
&baby,
&self.db,
self.repo.id,
self.settings.max_commits_per_sync,
)
.await?;
let txn: DatabaseTransaction = self.db.write().begin().await.map_err(SyncError::Db)?;
let (branch_count, _synced_commits, tag_count) =
self.sync_refs_and_data(&txn, &new_commits).await?;
languages::sync_repo_languages(&baby, &txn, &self.repo, &self.settings).await?;
stats::sync_repo_stats(
&txn,
&self.repo,
&self.repo_dir(),
new_commits.len(),
branch_count,
tag_count,
)
.await?;
txn.commit().await.map_err(SyncError::Db)?;
fsck::run_gc(&self.repo_dir()).await;
fsck::run_fsck(&self.repo_dir()).await;
Ok(())
}
async fn sync_refs_and_data(
&self,
txn: &DatabaseTransaction,
new_commits: &[CommitData],
) -> SyncResult<(usize, usize, usize)> {
let baby = self.baby.clone();
let branch_count = branches::sync_branches(&baby, txn, &self.repo).await?;
let synced = sync_commits(txn, self.repo.id, new_commits).await?;
let tag_count = tags::sync_tags(&baby, txn, &self.repo).await?;
Ok((branch_count, synced, tag_count))
}
pub async fn set_default_branch(&self, name: &str) -> SyncResult<()> {
self.baby
.set_default_branch(name)
.await
.map_err(SyncError::from)
}
pub async fn force_release_lock(lock: &SyncLock) -> SyncResult<()> {
lock.release().await
}
}