194c44c
use std::collections::{HashMap, HashSet};
use export::AppDatabase;
use gitbaby::GitBaby;
use gitbaby::commit::types::ListCommitsOptions;
use models::git::git_commits;
use models::users::{user_emails, user_external_identity};
use sea_orm::{ColumnTrait, DatabaseTransaction, EntityTrait, QueryFilter, QuerySelect, Set};
use time::OffsetDateTime;
use uuid::Uuid;
use crate::sync::error::{SyncError, SyncResult};
const BATCH_SIZE: usize = 100;
#[derive(Debug, Clone)]
pub(crate) struct CommitData {
pub sha: String,
pub author_name: String,
pub author_email: String,
pub committer_name: String,
pub committer_email: String,
pub message: String,
pub authored_at: OffsetDateTime,
pub committed_at: OffsetDateTime,
}
pub(crate) async fn collect_new_commits(
baby: &GitBaby,
existing_shas: &HashSet<String>,
max_total: usize,
) -> SyncResult<Vec<CommitData>> {
let branches = baby
.list_branches(Default::default())
.await
.map_err(SyncError::Git)?;
let mut seen: HashSet<String> = existing_shas.clone();
let mut out: Vec<CommitData> = Vec::new();
for branch in branches {
if out.len() >= max_total {
tracing::debug!(
new_commits = out.len(),
max_total,
"commit collection hit max_commits_per_sync cap"
);
break;
}
let remaining = max_total - out.len();
let opts = ListCommitsOptions {
revision: branch.name.clone(),
max_count: Some(remaining as u32),
..Default::default()
};
let commits = match baby.list_commits(opts).await {
Ok(v) => v,
Err(e) => {
tracing::warn!(branch = %branch.name, error = %e, "list_commits failed; skip branch");
continue;
}
};
for info in commits {
let sha = info.id.to_string();
if !seen.insert(sha.clone()) {
continue;
}
out.push(CommitData {
sha,
author_name: info.author.name,
author_email: info.author.email,
committer_name: info.committer.name,
committer_email: info.committer.email,
message: info.message,
authored_at: info.author.time,
committed_at: info.committer.time,
});
if out.len() >= max_total {
break;
}
}
}
Ok(out)
}
pub(crate) async fn existing_commit_shas(
db: &AppDatabase,
repo_id: Uuid,
) -> SyncResult<HashSet<String>> {
let rows = git_commits::Entity::find()
.filter(git_commits::Column::RepoId.eq(repo_id))
.select_only()
.column(git_commits::Column::Sha)
.into_tuple::<String>()
.all(db.read())
.await
.map_err(SyncError::Db)?;
Ok(rows.into_iter().collect())
}
pub(crate) async fn collect_new_commits_for_sync(
baby: &GitBaby,
db: &AppDatabase,
repo_id: Uuid,
max_total: usize,
) -> SyncResult<Vec<CommitData>> {
let existing = existing_commit_shas(db, repo_id).await?;
collect_new_commits(baby, &existing, max_total).await
}
pub(crate) async fn sync_commits(
txn: &DatabaseTransaction,
repo_id: Uuid,
new_commits: &[CommitData],
) -> SyncResult<usize> {
if new_commits.is_empty() {
return Ok(0);
}
let author_map = resolve_user_ids(txn, new_commits).await?;
let now = OffsetDateTime::now_utc();
for chunk in new_commits.chunks(BATCH_SIZE) {
let models = chunk
.iter()
.map(|c| {
let lower = c.author_email.to_lowercase();
git_commits::ActiveModel {
repo_id: Set(repo_id),
sha: Set(c.sha.clone()),
message: Set(c.message.clone()),
author_id: Set(author_map.get(&lower).copied()),
author_name: Set(c.author_name.clone()),
author_email: Set(c.author_email.clone()),
authored_at: Set(c.authored_at),
committer_name: Set(c.committer_name.clone()),
committer_email: Set(c.committer_email.clone()),
committed_at: Set(c.committed_at),
created_at: Set(now),
}
})
.collect::<Vec<_>>();
let mut oc = sea_orm::sea_query::OnConflict::new();
oc.do_nothing();
git_commits::Entity::insert_many(models)
.on_conflict(oc)
.exec(txn)
.await
.map_err(SyncError::Db)?;
}
Ok(new_commits.len())
}
async fn resolve_user_ids(
txn: &DatabaseTransaction,
commits: &[CommitData],
) -> SyncResult<HashMap<String, i64>> {
let mut lower_emails: HashSet<String> = HashSet::new();
for c in commits {
lower_emails.insert(c.author_email.to_lowercase());
lower_emails.insert(c.committer_email.to_lowercase());
}
if lower_emails.is_empty() {
return Ok(HashMap::new());
}
let rows = user_emails::Entity::find()
.filter(
user_emails::Column::LowerEmail.is_in(lower_emails.iter().cloned().collect::<Vec<_>>()),
)
.all(txn)
.await
.map_err(SyncError::Db)?;
let mut map: HashMap<String, i64> = rows
.into_iter()
.map(|r| (r.lower_email, r.user_id))
.collect();
let unmatched: Vec<String> = lower_emails
.iter()
.filter(|email| !map.contains_key(*email))
.cloned()
.collect();
if !unmatched.is_empty() {
let identities = user_external_identity::Entity::find()
.filter(user_external_identity::Column::UserId.is_not_null())
.filter(user_external_identity::Column::Email.is_in(unmatched))
.all(txn)
.await
.map_err(SyncError::Db)?;
for row in identities {
if let (Some(email), Some(uid)) = (row.email, row.user_id) {
map.entry(email.to_lowercase()).or_insert(uid);
}
}
}
tracing::debug!(
emails = lower_emails.len(),
matched = map.len(),
"resolved commit author emails to users"
);
Ok(map)
}