ec939f7
use config::ai::AiConfig;
use export::{AppDatabase, AppRedis};
use models::ai::ai_job::{ActiveModel, AiJobStatus};
use models::git::git_tags;
use models::repo::{repo_release, repos};
use sea_orm::PaginatorTrait;
use sea_orm::{
ActiveModelTrait, ColumnTrait, DatabaseConnection, EntityTrait, QueryFilter, QueryOrder, Set,
};
use time::OffsetDateTime;
use uuid::Uuid;
use crate::sync::error::{SyncError, SyncResult};
use crate::sync::queue::SyncQueue;
/// 生产侧的 AI 任务投递。
///
/// 负责「建作业记录 → 入队」，并在入队失败时把作业标为失败。所有调用点都应
/// 容忍这里的失败：AI 不可用不得阻断推送与合并请求主流程。
pub async fn submit_ai_task(
db: &DatabaseConnection,
redis: &AppRedis,
cfg: &AiConfig,
task: &ci_protocol::ai::AgentTask,
) -> SyncResult<Uuid> {
let job_id = task.job_id();
let now = OffsetDateTime::now_utc();
ActiveModel {
id: Set(job_id),
repo_id: Set(task.repo_id()),
kind: Set(kind_of(task)),
target_id: Set(task.target_id()),
status: Set(AiJobStatus::Pending),
attempts: Set(0),
model: Set(cfg.model.trim().to_string()),
target_ref: Set(task.target_ref()),
language: Set(task.language().unwrap_or_default().to_string()),
request_json: Set(None),
response_json: Set(None),
prompt_tokens: Set(None),
completion_tokens: Set(None),
error: Set(None),
created_at: Set(now),
started_at: Set(None),
finished_at: Set(None),
updated_at: Set(None),
}
.insert(db)
.await?;
let payload = task
.encode()
.map_err(|e| SyncError::Internal(format!("encode ai task: {e}")))?;
let queue = SyncQueue::new(redis.clone());
if let Err(error) = queue.publish_raw(&cfg.queue_topic, None, &payload).await {
mark_submit_failure(db, job_id, &error.to_string()).await;
return Err(error);
}
tracing::debug!(job_id = %job_id, kind = task.kind_str(), "ai task submitted");
Ok(job_id)
}
/// 静默投递：失败只记日志。用于推送与 PR 事件等不允许被 AI 影响的路径。
pub async fn submit_ai_task_logged(
db: &DatabaseConnection,
redis: &AppRedis,
cfg: &AiConfig,
task: &ci_protocol::ai::AgentTask,
) {
match submit_ai_task(db, redis, cfg, task).await {
Ok(job_id) => tracing::debug!(%job_id, kind = task.kind_str(), "ai task submitted"),
Err(error) => {
tracing::warn!(%error, kind = task.kind_str(), "ai task submission failed")
}
}
}
/// 一次推送后投递 AI 任务：分支更新做提交评审，标签推送建 release 草稿并写文案。
///
/// `refs` 直接复用 post-receive 捕获的引用更新（`name`/`old_oid`/`new_oid`）。
pub async fn trigger_push_ai_tasks(
db: &AppDatabase,
redis: &AppRedis,
cfg: &AiConfig,
repo_id: Uuid,
actor_id: Option<Uuid>,
refs: &[(String, String, String)],
) {
if !cfg.enabled {
return;
}
let conn = db.write();
// 一次推送产生的所有任务共用触发者语言。
let language = crate::sync::ai_language::actor_language(conn, actor_id).await;
if cfg.review_on() {
for (name, _old, new) in refs {
if !name.starts_with("refs/heads/") || is_zero_oid(new) {
continue;
}
let task = ci_protocol::ai::AgentTask::CommitReview {
job_id: Uuid::now_v7(),
repo_id,
commit_sha: new.clone(),
language: language.clone(),
};
submit_ai_task_logged(conn, redis, cfg, &task).await;
}
}
if cfg.release_notes_on() {
for (name, _old, new) in refs {
let Some(tag) = name.strip_prefix("refs/tags/") else {
continue;
};
if is_zero_oid(new) {
continue;
}
match ensure_release_draft(
conn,
redis,
cfg,
repo_id,
actor_id,
language.clone(),
tag,
new,
)
.await
{
Ok(Some(job_id)) => {
tracing::debug!(%job_id, tag, "release draft queued");
}
Ok(None) => {}
Err(error) => tracing::warn!(%error, tag, "release draft creation failed"),
}
}
}
}
/// 若该 tag 尚无 release，则建草稿并入队文案任务；已存在则返回 `None`。
#[allow(clippy::too_many_arguments)]
pub async fn ensure_release_draft(
db: &DatabaseConnection,
redis: &AppRedis,
cfg: &AiConfig,
repo_id: Uuid,
actor_id: Option<Uuid>,
language: Option<String>,
tag_name: &str,
target_sha: &str,
) -> SyncResult<Option<Uuid>> {
let lower_tag = tag_name.to_lowercase();
if repo_release::Entity::find()
.filter(repo_release::Column::RepoId.eq(repo_id))
.filter(repo_release::Column::LowerTagName.eq(lower_tag.clone()))
.one(db)
.await?
.is_some()
{
return Ok(None);
}
let publisher = match actor_id {
Some(uid) => uid,
None => repo_publisher_fallback(db, repo_id).await?,
};
let now = OffsetDateTime::now_utc();
let release_id = Uuid::now_v7();
let previous = previous_tag_target(db, repo_id, tag_name).await;
let commit_count = count_commits_between(db, repo_id, previous.as_deref(), target_sha).await;
repo_release::ActiveModel {
id: Set(release_id),
repo_id: Set(repo_id),
// 草稿由系统创建，发布者取仓库所有者；用户发布时再改写为实际操作者。
publisher_id: Set(publisher),
publisher_name: Set(String::new()),
tag_name: Set(tag_name.to_string()),
lower_tag_name: Set(lower_tag),
target: Set(target_sha.to_string()),
title: Set(tag_name.to_string()),
content: Set(None),
sha: Set(target_sha.to_string()),
num_commits: Set(commit_count),
is_draft: Set(true),
is_prerelease: Set(tag_name.contains('-')),
is_tag: Set(true),
created_at: Set(now),
updated_at: Set(now),
}
.insert(db)
.await?;
let task = ci_protocol::ai::AgentTask::ReleaseNotes {
job_id: Uuid::now_v7(),
repo_id,
release_id,
tag_name: tag_name.to_string(),
from_sha: previous,
to_sha: target_sha.to_string(),
language: language.or(resolve_language_for(db, actor_id).await),
};
let job_id = submit_ai_task(db, redis, cfg, &task).await?;
Ok(Some(job_id))
}
/// 草稿发布者的兜底：无推送者（如系统同步）时取空间所有者。
async fn repo_publisher_fallback(db: &DatabaseConnection, repo_id: Uuid) -> SyncResult<Uuid> {
let repo = repos::Entity::find_by_id(repo_id)
.one(db)
.await?
.ok_or_else(|| SyncError::Internal("repo not found for release".to_string()))?;
let space = crate::share::access::space_for_dir(db, repo.dir_id)
.await
.map_err(|e| SyncError::Internal(format!("space lookup: {e}")))?
.ok_or_else(|| SyncError::Internal("space not found for release".to_string()))?;
Ok(space.owner_id)
}
/// 入队时的语言解析：调用方已解析过就直接用，避免重复查用户表。
async fn resolve_language_for(db: &DatabaseConnection, actor_id: Option<Uuid>) -> Option<String> {
crate::sync::ai_language::actor_language(db, actor_id).await
}
/// 该 tag 之前最近一个标签的提交，用作 changelog 区间起点。
async fn previous_tag_target(
db: &DatabaseConnection,
repo_id: Uuid,
tag_name: &str,
) -> Option<String> {
let rows = git_tags::Entity::find()
.filter(git_tags::Column::RepoId.eq(repo_id))
.order_by_desc(git_tags::Column::CreatedAt)
.all(db)
.await
.ok()?;
let mut seen_current = false;
for row in rows {
if row.name == tag_name {
seen_current = true;
continue;
}
if seen_current {
return Some(row.commit_sha);
}
}
// 排序未命中当前 tag（例如首次同步）时，退回时间最早的其他标签。
None
}
async fn count_commits_between(
db: &DatabaseConnection,
repo_id: Uuid,
from: Option<&str>,
to: &str,
) -> i32 {
use models::git::git_commits;
let mut select = git_commits::Entity::find().filter(git_commits::Column::RepoId.eq(repo_id));
if let Some(from) = from {
select = select.filter(git_commits::Column::Sha.ne(from.to_string()));
}
let _ = to;
select
.count(db)
.await
.map(|count| count as i32)
.unwrap_or(0)
}
fn is_zero_oid(oid: &str) -> bool {
oid.is_empty() || oid.chars().all(|c| c == '0')
}
async fn mark_submit_failure(db: &DatabaseConnection, job_id: Uuid, error: &str) {
use models::ai::ai_job;
use sea_orm::EntityTrait;
let now = OffsetDateTime::now_utc();
if let Ok(Some(row)) = ai_job::Entity::find_by_id(job_id).one(db).await {
let mut model: ai_job::ActiveModel = row.into();
model.status = Set(AiJobStatus::Failed);
model.error = Set(Some(error.to_string()));
model.finished_at = Set(Some(now));
model.updated_at = Set(Some(now));
let _ = model.update(db).await;
}
}
fn kind_of(task: &ci_protocol::ai::AgentTask) -> models::ai::ai_job::AiJobKind {
use ci_protocol::ai::AgentTask;
use models::ai::ai_job::AiJobKind;
match task {
AgentTask::PullReview { .. } => AiJobKind::PullReview,
AgentTask::CommitReview { .. } => AiJobKind::CommitReview,
AgentTask::RepoSummary { .. } => AiJobKind::RepoSummary,
AgentTask::ReleaseNotes { .. } => AiJobKind::ReleaseNotes,
}
}