9301161
use std::fs;
use std::io;
#[cfg(unix)]
use std::os::unix::fs::PermissionsExt;
use std::path::{Path, PathBuf};
use uuid::Uuid;
use crate::sync::event::{PostReceiveRefUpdate, PostReceiveTask, enqueue_post_receive};
use crate::sync::queue::SyncQueue;
use config::git::SyncSettings;
/// 推送侧 AI 触发所需的依赖；不存在时该推送完全不触发 AI。
#[derive(Clone)]
pub struct AiPushCtx {
pub db: std::sync::Arc<export::AppDatabase>,
pub redis: export::AppRedis,
pub config: config::ai::AiConfig,
}
/// Who and where a push belongs to. Grouped into a struct because every
/// post-receive entry point threads the same four values, and three of them are
/// `Uuid`-shaped — positional arguments silently swap space and repo.
pub struct PushTarget<'a> {
pub space_id: Uuid,
pub repo_id: Uuid,
pub actor_id: Option<Uuid>,
pub bare: &'a Path,
}
pub const POST_RECEIVE_HOOK: &str = "post-receive";
pub const CAPTURE_FILE: &str = "_gitdata_post_receive";
const HEX_OID_LEN: usize = 40;
const HEX_OID_SHA256_LEN: usize = 64;
pub fn hook_path(bare: &Path) -> PathBuf {
bare.join("hooks").join(POST_RECEIVE_HOOK)
}
pub fn capture_path(bare: &Path) -> PathBuf {
bare.join(CAPTURE_FILE)
}
pub fn ensure_post_receive_hook(bare: &Path) -> io::Result<()> {
let path = hook_path(bare);
let script = capture_script();
if path.exists() {
let existing = fs::read_to_string(&path)?;
if !existing.contains(CAPTURE_FILE) {
tracing::warn!(
path = %path.display(),
"existing post-receive hook will not be overwritten; post-receive sync may not capture pushes"
);
return Ok(());
}
if existing == script {
return Ok(());
}
fs::write(&path, script)?;
#[cfg(unix)]
fs::set_permissions(&path, fs::Permissions::from_mode(0o755))?;
tracing::debug!(path = %path.display(), "refreshed post-receive capture hook");
return Ok(());
}
let Some(hooks_dir) = path.parent() else {
return Err(io::Error::other("hook path has no parent"));
};
fs::create_dir_all(hooks_dir)?;
fs::write(&path, script)?;
#[cfg(unix)]
fs::set_permissions(&path, fs::Permissions::from_mode(0o755))?;
tracing::debug!(path = %path.display(), "installed post-receive capture hook");
Ok(())
}
fn capture_script() -> String {
format!(
"#!/bin/sh\n# gitdata post-receive capture hook\nwhile read old_oid new_oid refname; do\n printf '%s %s %s\\n' \"$old_oid\" \"$new_oid\" \"$refname\" >> \"$GIT_DIR/{CAPTURE_FILE}\"\ndone\n"
)
}
pub fn clear_post_receive_capture(bare: &Path) -> io::Result<()> {
match fs::remove_file(capture_path(bare)) {
Ok(()) => Ok(()),
Err(e) if e.kind() == io::ErrorKind::NotFound => Ok(()),
Err(e) => Err(e),
}
}
pub fn prepare_post_receive(bare: &Path) -> io::Result<()> {
ensure_post_receive_hook(bare)?;
clear_post_receive_capture(bare)
}
pub fn take_post_receive_capture(bare: &Path) -> Vec<PostReceiveRefUpdate> {
let path = capture_path(bare);
let content = match fs::read_to_string(&path) {
Ok(c) => c,
Err(e) if e.kind() == std::io::ErrorKind::NotFound => {
tracing::warn!(
bare = %bare.display(),
"post-receive capture file missing after push; capture hook may not have run"
);
return Vec::new();
}
Err(e) => {
tracing::warn!(error = %e, "failed to read post-receive capture file");
return Vec::new();
}
};
if let Err(e) = fs::remove_file(&path) {
tracing::warn!(error = %e, "failed to remove post-receive capture file");
}
let mut dropped = 0usize;
let updates: Vec<_> = content
.lines()
.filter_map(|line| {
let parsed = parse_capture_line(line);
if parsed.is_none() {
dropped += 1;
}
parsed
})
.collect();
if dropped > 0 {
tracing::warn!(
bare = %bare.display(),
dropped,
"discarded unparseable post-receive capture lines"
);
}
updates
}
fn parse_capture_line(line: &str) -> Option<PostReceiveRefUpdate> {
let mut parts = line.split_whitespace();
let old_oid = parts.next()?;
let new_oid = parts.next()?;
let name = parts.next()?;
if !is_valid_oid(old_oid) || !is_valid_oid(new_oid) || name.is_empty() {
return None;
}
Some(PostReceiveRefUpdate::new(name, old_oid, new_oid))
}
fn is_valid_oid(oid: &str) -> bool {
let len = oid.len();
if len != HEX_OID_LEN && len != HEX_OID_SHA256_LEN {
return false;
}
oid.bytes().all(|b| b.is_ascii_hexdigit())
}
pub async fn enqueue_post_receive_push(
queue: &SyncQueue,
settings: &SyncSettings,
target: &PushTarget<'_>,
) -> Result<usize, EnqueueError> {
enqueue_push_impl(queue, settings, None, None, target).await
}
/// Enqueues the sync task and (optionally) publishes a CI push event from the
/// same capture file read.
pub async fn enqueue_post_receive_push_with_ci(
queue: &SyncQueue,
settings: &SyncSettings,
ci_topic: Option<&str>,
target: &PushTarget<'_>,
) -> Result<usize, EnqueueError> {
enqueue_push_impl(queue, settings, ci_topic, None, target).await
}
/// 推送入口：同时镜像 CI 事件、投递 AI 任务并排入同步队列。
pub async fn enqueue_post_receive_push_full(
queue: &SyncQueue,
settings: &SyncSettings,
ci_topic: Option<&str>,
ai_ctx: Option<AiPushCtx>,
target: &PushTarget<'_>,
) -> Result<usize, EnqueueError> {
enqueue_push_impl(queue, settings, ci_topic, ai_ctx.as_ref(), target).await
}
async fn enqueue_push_impl(
queue: &SyncQueue,
settings: &SyncSettings,
ci_topic: Option<&str>,
ai_ctx: Option<&AiPushCtx>,
target: &PushTarget<'_>,
) -> Result<usize, EnqueueError> {
// Copied out so the borrow of `target` ends here; the fields are `Copy`.
let (channel_id, repo_id, actor_id) = (target.space_id, target.repo_id, target.actor_id);
let bare = target.bare;
let refs = take_post_receive_capture(bare);
if refs.is_empty() {
return Ok(0);
}
// AI 触发只依赖本次引用更新，且必须容忍失败，因此先于 CI 事件执行。
if let Some(ai) = ai_ctx {
let flat: Vec<(String, String, String)> = refs
.iter()
.map(|r| (r.name.clone(), r.old_oid.clone(), r.new_oid.clone()))
.collect();
crate::sync::ai_task::trigger_push_ai_tasks(
&ai.db, &ai.redis, &ai.config, repo_id, actor_id, &flat,
)
.await;
}
if let Some(topic) = ci_topic {
let ci_refs: Vec<ci_protocol::events::RefUpdate> = refs
.iter()
.map(|r| ci_protocol::events::RefUpdate {
name: r.name.clone(),
old_oid: r.old_oid.clone(),
new_oid: r.new_oid.clone(),
})
.collect();
if ci_refs.iter().any(|r| !r.is_delete()) {
let event = ci_protocol::events::CiEvent::Push(ci_protocol::events::PushEvent {
id: Uuid::now_v7(),
repo_id,
space_id: channel_id,
actor_id,
refs: ci_refs,
trace: export::telemetry::traceparent(),
});
match serde_json::to_vec(&event) {
Ok(payload) => {
if let Err(error) = queue.publish_raw(topic, None, &payload).await {
tracing::warn!(%error, "ci push event publish failed");
}
}
Err(error) => tracing::warn!(%error, "ci push event serialize failed"),
}
}
}
let task = PostReceiveTask::new(repo_id, channel_id, actor_id, refs);
enqueue_post_receive(queue, settings, &task).await?;
Ok(task.refs.len())
}
pub async fn enqueue_post_receive_push_logged(
queue: &SyncQueue,
settings: &SyncSettings,
target: &PushTarget<'_>,
) {
let (space_id, repo_id) = (target.space_id, target.repo_id);
match enqueue_post_receive_push(queue, settings, target).await {
Ok(n) if n > 0 => {
tracing::debug!(
count = n,
space_id = %space_id,
repo_id = %repo_id,
"enqueued post-receive ref updates"
)
}
Ok(_) => tracing::debug!(space_id = %space_id, repo_id = %repo_id, "no refs to enqueue"),
Err(e) => tracing::warn!(
error = %e,
space_id = %space_id,
repo_id = %repo_id,
"post-receive enqueue failed"
),
}
}
pub async fn enqueue_post_receive_push_with_ci_logged(
queue: &SyncQueue,
settings: &SyncSettings,
ci_topic: Option<&str>,
ai_ctx: Option<AiPushCtx>,
target: &PushTarget<'_>,
) {
let (space_id, repo_id) = (target.space_id, target.repo_id);
match enqueue_post_receive_push_full(queue, settings, ci_topic, ai_ctx, target).await {
Ok(n) if n > 0 => tracing::debug!(
count = n,
space_id = %space_id,
repo_id = %repo_id,
"enqueued post-receive ref updates and ci event"
),
Ok(_) => tracing::debug!(space_id = %space_id, repo_id = %repo_id, "no refs to enqueue"),
Err(e) => tracing::warn!(
error = %e,
space_id = %space_id,
repo_id = %repo_id,
"post-receive enqueue failed"
),
}
}
pub use crate::sync::event::EnqueueError;