f0a1552
ZhenYi
feat(chat): add chat module with room management, message handling, and session revocation
f0a1552
use std::collections::HashMap;
use models::git::git_commits;
use models::users::{user, user_emails, user_external_identity};
use sea_orm::{
ActiveModelTrait, ColumnTrait, DatabaseConnection, EntityTrait, QueryFilter, QuerySelect, Set,
};
use time::OffsetDateTime;
use uuid::Uuid;
use crate::errors::ImportError;
use crate::providers::ImportProvider;
use crate::service::ImportOptions;
struct Candidate {
login: String,
external_id: Option<String>,
name: Option<String>,
email: Option<String>,
avatar_url: Option<String>,
}
pub fn login_from_noreply_email(email: &str) -> Option<String> {
let (local, domain) = email.split_once('@')?;
let domain = domain.to_ascii_lowercase();
let known = domain.ends_with("users.noreply.github.com")
|| domain.ends_with("users.noreply.gitlab.com")
|| domain.starts_with("noreply.");
if !known {
return None;
}
let login = match local.split_once('+') {
Some((_, rest)) if !rest.is_empty() => rest,
_ => local,
};
let login = login.trim();
if login.is_empty() {
None
} else {
Some(login.to_string())
}
}
pub async fn harvest(
db: &DatabaseConnection,
provider: ImportProvider,
repo_id: Uuid,
options: &ImportOptions,
) -> Result<(), ImportError> {
let mut candidates: Vec<Candidate> = Vec::new();
if !options.owner.trim().is_empty() {
candidates.push(Candidate {
login: options.owner.clone(),
external_id: options.owner_external_id.clone(),
name: None,
email: None,
avatar_url: options.owner_avatar_url.clone(),
});
}
let authors: Vec<(String, String)> = git_commits::Entity::find()
.select_only()
.column(git_commits::Column::AuthorName)
.column(git_commits::Column::AuthorEmail)
.filter(git_commits::Column::RepoId.eq(repo_id))
.filter(git_commits::Column::AuthorId.is_null())
.distinct()
.into_tuple()
.all(db)
.await?;
let mut seen: Vec<String> = Vec::new();
for (name, email) in authors {
let Some(login) = login_from_noreply_email(&email) else {
continue;
};
let lower = login.to_lowercase();
if seen.contains(&lower) {
continue;
}
seen.push(lower);
candidates.push(Candidate {
login,
external_id: None,
name: Some(name),
email: Some(email),
avatar_url: None,
});
}
let mut resolved: HashMap<String, Uuid> = HashMap::new();
for candidate in candidates {
if let Some((email, uid)) = upsert(db, provider, &candidate).await? {
resolved.insert(email, uid);
}
}
backfill_commit_authors(db, repo_id, &resolved).await?;
Ok(())
}
async fn resolve_user(
db: &DatabaseConnection,
candidate: &Candidate,
lower_login: &str,
) -> Result<Option<Uuid>, ImportError> {
if let Some(email) = candidate
.email
.as_deref()
.map(str::trim)
.filter(|v| !v.is_empty())
{
let row = user_emails::Entity::find()
.filter(user_emails::Column::LowerEmail.eq(email.to_lowercase()))
.one(db)
.await?;
if let Some(row) = row {
return Ok(Some(row.user_id));
}
}
let row = user::Entity::find()
.filter(user::Column::LowerName.eq(lower_login))
.one(db)
.await?;
Ok(row.map(|row| row.uid))
}
async fn upsert(
db: &DatabaseConnection,
provider: ImportProvider,
candidate: &Candidate,
) -> Result<Option<(String, Uuid)>, ImportError> {
let login = candidate.login.trim();
if login.is_empty() {
return Ok(None);
}
let lower_login = login.to_lowercase();
let lower_email = candidate
.email
.as_deref()
.map(str::trim)
.filter(|v| !v.is_empty())
.map(str::to_lowercase);
let uid = resolve_user(db, candidate, &lower_login).await?;
let now = OffsetDateTime::now_utc();
let existing = user_external_identity::Entity::find()
.filter(user_external_identity::Column::Provider.eq(provider.as_str()))
.filter(user_external_identity::Column::LowerLogin.eq(&lower_login))
.one(db)
.await?;
let resolved_uid = match existing {
Some(row) => {
let mut model = user_external_identity::ActiveModel {
id: Set(row.id),
updated_at: Set(now),
..Default::default()
};
if let Some(value) = candidate.external_id.clone().filter(|v| !v.is_empty()) {
model.external_id = Set(Some(value));
}
if let Some(value) = candidate.name.clone().filter(|v| !v.is_empty()) {
model.name = Set(Some(value));
}
if let Some(value) = candidate.avatar_url.clone().filter(|v| !v.is_empty()) {
model.avatar_url = Set(Some(value));
}
if let Some(value) = lower_email.clone() {
model.email = Set(Some(value));
}
let resolved = row.user_id.or(uid);
if row.user_id.is_none() && resolved.is_some() {
model.user_id = Set(resolved);
}
model.update(db).await?;
resolved
}
None => {
let row = user_external_identity::ActiveModel {
id: Set(Uuid::now_v7()),
provider: Set(provider.as_str().to_string()),
external_id: Set(candidate.external_id.clone()),
lower_login: Set(lower_login),
login: Set(login.to_string()),
name: Set(candidate.name.clone()),
email: Set(lower_email.clone()),
avatar_url: Set(candidate.avatar_url.clone()),
user_id: Set(uid),
created_at: Set(now),
updated_at: Set(now),
}
.insert(db)
.await?;
row.user_id
}
};
Ok(match (candidate.email.clone(), resolved_uid, lower_email) {
(Some(email), Some(uid), Some(_)) if !email.trim().is_empty() => Some((email, uid)),
_ => None,
})
}
async fn backfill_commit_authors(
db: &DatabaseConnection,
repo_id: Uuid,
resolved: &HashMap<String, Uuid>,
) -> Result<(), ImportError> {
for (email, uid) in resolved {
git_commits::Entity::update_many()
.col_expr(
git_commits::Column::AuthorId,
sea_orm::sea_query::Expr::value(*uid),
)
.filter(git_commits::Column::RepoId.eq(repo_id))
.filter(git_commits::Column::AuthorId.is_null())
.filter(git_commits::Column::AuthorEmail.eq(email.clone()))
.exec(db)
.await?;
}
Ok(())
}