dev-manager-tauri/src-tauri/src/commands/gitea.rs
lanrtop 47ff656083 fix(resource/credentials): 保留删除失败引用并纠正迁移前状态
删除受保护凭证失败时回滚数据库删除,并恢复已删引用。旧凭证明确标为 pending_migration,避免误报 configured。
2026-09-23 21:16:11 +09:00

3152 lines
117 KiB
Rust
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

//! Gitea 集成——控制面:实例管理 + 仓库操作 + webhook 注册。
//!
//! 观测面(webhook 接收端)在 mcp_server.rs 实现。
use rusqlite::{params, OptionalExtension};
use serde::{Deserialize, Serialize};
// ── 数据结构 ──────────────────────────────────────────────────────────────────
#[derive(Debug, Serialize, Deserialize, Clone)]
pub struct GiteaInstance {
pub id: String,
pub name: String,
pub url: String,
pub is_default: bool,
pub created_at: Option<String>,
}
#[derive(Debug, Deserialize)]
pub struct SaveGiteaInstanceInput {
pub name: String,
pub url: String,
pub access_token: String,
pub is_default: Option<bool>,
}
#[derive(Debug, Serialize, Deserialize, Clone)]
pub struct GiteaRepo {
pub id: String,
pub instance_id: String,
pub project_id: String,
pub gitea_owner: String,
pub gitea_repo_name: String,
pub clone_url: Option<String>,
pub webhook_id: Option<i64>,
pub linked_at: Option<String>,
}
#[derive(Debug, Serialize)]
pub struct GiteaUser {
pub login: String,
pub full_name: String,
pub email: String,
}
// ── 辅助函数 ──────────────────────────────────────────────────────────────────
fn row_to_instance(row: &rusqlite::Row<'_>) -> rusqlite::Result<GiteaInstance> {
Ok(GiteaInstance {
id: row.get(0)?,
name: row.get(1)?,
url: row.get(2)?,
is_default: row.get::<_, i64>(3)? != 0,
created_at: row.get(4)?,
})
}
fn row_to_repo(row: &rusqlite::Row<'_>) -> rusqlite::Result<GiteaRepo> {
Ok(GiteaRepo {
id: row.get(0)?,
instance_id: row.get(1)?,
project_id: row.get(2)?,
gitea_owner: row.get(3)?,
gitea_repo_name: row.get(4)?,
clone_url: row.get(5)?,
webhook_id: row.get(6)?,
linked_at: row.get(7)?,
})
}
/// 规范化 Gitea URL:去掉末尾 /
fn normalize_url(url: &str) -> String {
url.trim_end_matches('/').to_string()
}
#[derive(Debug, Serialize)]
pub struct ResourceReadiness {
pub state: &'static str,
pub source: &'static str,
}
#[derive(Debug, Serialize)]
pub struct ProjectResourceStatus {
pub project_id: String,
pub gitea_pr: ResourceReadiness,
pub gitea_webhook: ResourceReadiness,
}
fn credential_readiness(
store: &impl crate::credential_store::CredentialStore,
reference: Option<String>,
has_legacy: bool,
) -> ResourceReadiness {
if let Some(reference) = reference.filter(|reference| !reference.is_empty()) {
return match store.get(&reference) {
Ok(Some(value)) if !value.is_empty() => ResourceReadiness {
state: "configured",
source: "protected_store",
},
Ok(_) => ResourceReadiness {
state: "needs_repair",
source: "protected_store",
},
Err(_) => ResourceReadiness {
state: "unavailable",
source: "protected_store",
},
};
}
if has_legacy {
match store.get("resource-readiness/probe") {
Ok(_) => ResourceReadiness {
state: "pending_migration",
source: "legacy_pending_migration",
},
Err(_) => ResourceReadiness {
state: "unavailable",
source: "protected_store",
},
}
} else {
ResourceReadiness {
state: "not_configured",
source: "gitea_binding",
}
}
}
pub fn project_resource_status_for_conn(
conn: &rusqlite::Connection,
store: &impl crate::credential_store::CredentialStore,
project_id: &str,
) -> Result<ProjectResourceStatus, String> {
let exists: bool = conn
.query_row(
"SELECT EXISTS(SELECT 1 FROM project_workspaces WHERE id = ?1)",
params![project_id],
|row| row.get(0),
)
.map_err(|_| "项目状态不可用".to_string())?;
if !exists {
return Err("项目不存在".into());
}
let binding: Option<(Option<String>, bool, Option<String>, bool)> = conn.query_row(
"SELECT gi.access_token_ref, length(gi.access_token) > 0, gr.webhook_secret_ref, coalesce(length(gr.webhook_secret), 0) > 0
FROM gitea_repos gr JOIN gitea_instances gi ON gi.id = gr.instance_id WHERE gr.project_id = ?1",
params![project_id],
|row| Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?)),
).optional().map_err(|_| "项目资源状态不可用".to_string())?;
let (gitea_pr, gitea_webhook) = match binding {
Some((token_ref, token_legacy, webhook_ref, webhook_legacy)) => (
credential_readiness(store, token_ref, token_legacy),
credential_readiness(store, webhook_ref, webhook_legacy),
),
None => (
ResourceReadiness {
state: "not_configured",
source: "gitea_binding",
},
ResourceReadiness {
state: "not_configured",
source: "gitea_binding",
},
),
};
Ok(ProjectResourceStatus {
project_id: project_id.into(),
gitea_pr,
gitea_webhook,
})
}
#[tauri::command]
pub fn get_project_resource_status(project_id: String) -> Result<ProjectResourceStatus, String> {
let conn = crate::db::pool()
.get()
.map_err(|_| "项目资源状态不可用".to_string())?;
project_resource_status_for_conn(
&conn,
&crate::credential_store::WindowsCredentialStore,
&project_id,
)
}
fn save_gitea_instance_with_store(
conn: &rusqlite::Connection,
store: &impl crate::credential_store::CredentialStore,
input: SaveGiteaInstanceInput,
) -> Result<GiteaInstance, String> {
let id = uuid::Uuid::new_v4().to_string();
let url = normalize_url(&input.url);
let is_default = input.is_default.unwrap_or(false);
let reference = crate::credential_store::versioned_reference(
crate::credential_store::LegacyCredentialKind::GiteaAccessToken,
&id,
);
crate::credential_store::put_verified(store, &reference, input.access_token.as_bytes())?;
let result = (|| {
let tx = conn
.unchecked_transaction()
.map_err(|_| "保存 Gitea 实例失败".to_string())?;
if is_default {
tx.execute("UPDATE gitea_instances SET is_default = 0", [])
.map_err(|_| "保存 Gitea 实例失败".to_string())?;
}
tx.execute(
"INSERT INTO gitea_instances (id, name, url, access_token, is_default, access_token_ref)
VALUES (?1, ?2, ?3, '', ?4, ?5)",
params![id, input.name, url, is_default as i64, reference],
)
.map_err(|_| "保存 Gitea 实例失败".to_string())?;
let instance = tx
.query_row(
"SELECT id, name, url, is_default, created_at FROM gitea_instances WHERE id = ?1",
params![id],
row_to_instance,
)
.map_err(|_| "保存 Gitea 实例后读取失败".to_string())?;
tx.commit().map_err(|_| "保存 Gitea 实例失败".to_string())?;
Ok(instance)
})();
if result.is_err() {
let _ = store.delete(&reference);
}
result
}
// ── 实例管理命令 ──────────────────────────────────────────────────────────────
/// 保存 Gitea 实例(新增或覆盖同名实例)
#[tauri::command]
pub fn gitea_save_instance(input: SaveGiteaInstanceInput) -> Result<GiteaInstance, String> {
let conn = crate::db::pool().get().map_err(|e| e.to_string())?;
let store = crate::credential_store::WindowsCredentialStore;
save_gitea_instance_with_store(&conn, &store, input)
}
/// 列出所有 Gitea 实例
#[tauri::command]
pub fn gitea_list_instances() -> Result<Vec<GiteaInstance>, String> {
let conn = crate::db::pool().get().map_err(|e| e.to_string())?;
let mut stmt = conn.prepare(
"SELECT id, name, url, is_default, created_at FROM gitea_instances ORDER BY is_default DESC, created_at ASC"
).map_err(|e| e.to_string())?;
let rows = stmt
.query_map([], row_to_instance)
.map_err(|e| e.to_string())?
.collect::<rusqlite::Result<Vec<_>>>()
.map_err(|e| e.to_string())?;
Ok(rows)
}
/// 删除 Gitea 实例(级联删除 gitea_repos)
#[tauri::command]
pub fn gitea_delete_instance(id: String) -> Result<(), String> {
let conn = crate::db::pool().get().map_err(|e| e.to_string())?;
delete_gitea_instance_with_store(&conn, &crate::credential_store::WindowsCredentialStore, &id)
}
fn delete_gitea_instance_with_store(
conn: &rusqlite::Connection,
store: &impl crate::credential_store::CredentialStore,
id: &str,
) -> Result<(), String> {
let instance_reference: Option<String> = conn
.query_row(
"SELECT access_token_ref FROM gitea_instances WHERE id = ?1",
params![id],
|row| row.get(0),
)
.map_err(|_| "Gitea 实例不存在".to_string())?;
let repo_references: Vec<String> = conn
.prepare("SELECT webhook_secret_ref FROM gitea_repos WHERE instance_id = ?1 AND webhook_secret_ref IS NOT NULL")
.map_err(|_| "删除 Gitea 实例失败".to_string())?
.query_map(params![id], |row| row.get(0))
.map_err(|_| "删除 Gitea 实例失败".to_string())?
.collect::<rusqlite::Result<_>>()
.map_err(|_| "删除 Gitea 实例失败".to_string())?;
let tx = conn
.unchecked_transaction()
.map_err(|_| "删除 Gitea 实例失败".to_string())?;
tx.execute("DELETE FROM gitea_instances WHERE id = ?1", params![id])
.map_err(|_| "删除 Gitea 实例失败".to_string())?;
let references: Vec<String> = instance_reference
.into_iter()
.chain(repo_references)
.collect();
let snapshots = crate::credential_store::delete_references(store, &references)?;
if tx.commit().is_err() {
crate::credential_store::restore_references(store, &snapshots)?;
return Err("删除 Gitea 实例失败:数据库提交失败".into());
}
Ok(())
}
/// 更新 Gitea 实例(仅更新非空字段)
#[tauri::command]
pub fn gitea_update_instance(
id: String,
name: Option<String>,
url: Option<String>,
access_token: Option<String>,
is_default: Option<bool>,
) -> Result<GiteaInstance, String> {
let conn = crate::db::pool().get().map_err(|e| e.to_string())?;
update_gitea_instance_with_store(
&conn,
&crate::credential_store::WindowsCredentialStore,
&id,
name,
url,
access_token,
is_default,
)
}
fn update_gitea_instance_with_store(
conn: &rusqlite::Connection,
store: &impl crate::credential_store::CredentialStore,
id: &str,
name: Option<String>,
url: Option<String>,
access_token: Option<String>,
is_default: Option<bool>,
) -> Result<GiteaInstance, String> {
let old_reference: Option<String> = conn
.query_row(
"SELECT access_token_ref FROM gitea_instances WHERE id = ?1",
params![id],
|row| row.get(0),
)
.map_err(|_| "Gitea 实例不存在".to_string())?;
let new_reference = access_token.as_ref().map(|_| {
crate::credential_store::versioned_reference(
crate::credential_store::LegacyCredentialKind::GiteaAccessToken,
id,
)
});
if let (Some(token), Some(reference)) = (&access_token, &new_reference) {
crate::credential_store::put_verified(store, reference, token.as_bytes())?;
}
let result = (|| {
let tx = conn
.unchecked_transaction()
.map_err(|_| "更新 Gitea 实例失败".to_string())?;
if let Some(ref n) = name {
tx.execute(
"UPDATE gitea_instances SET name = ?1 WHERE id = ?2",
params![n, id],
)
.map_err(|_| "更新 Gitea 实例失败".to_string())?;
}
if let Some(ref u) = url {
tx.execute(
"UPDATE gitea_instances SET url = ?1 WHERE id = ?2",
params![normalize_url(u), id],
)
.map_err(|_| "更新 Gitea 实例失败".to_string())?;
}
if let Some(ref reference) = new_reference {
let changed = tx.execute(
"UPDATE gitea_instances SET access_token = '', access_token_ref = ?1 WHERE id = ?2 AND access_token_ref IS ?3",
params![reference, id, old_reference],
)
.map_err(|_| "更新 Gitea 实例失败".to_string())?;
if changed != 1 {
return Err("更新 Gitea 实例失败:凭证引用已变化".to_string());
}
}
if let Some(def) = is_default {
if def {
tx.execute("UPDATE gitea_instances SET is_default = 0", [])
.map_err(|_| "更新 Gitea 实例失败".to_string())?;
}
tx.execute(
"UPDATE gitea_instances SET is_default = ?1 WHERE id = ?2",
params![def as i64, id],
)
.map_err(|_| "更新 Gitea 实例失败".to_string())?;
}
let instance = tx
.query_row(
"SELECT id, name, url, is_default, created_at FROM gitea_instances WHERE id = ?1",
params![id],
row_to_instance,
)
.map_err(|_| "Gitea 实例不存在".to_string())?;
tx.commit().map_err(|_| "更新 Gitea 实例失败".to_string())?;
Ok(instance)
})();
if result.is_err() {
if let Some(reference) = &new_reference {
let _ = store.delete(reference);
}
} else if let Some(old) = old_reference.filter(|_| new_reference.is_some()) {
let _ = store.delete(&old);
}
result
}
/// 测试 Gitea 连接:调 GET /api/v1/user,返回用户信息验证 token 有效性
#[tauri::command]
pub fn gitea_test_connection(instance_id: String) -> Result<GiteaUser, String> {
let conn = crate::db::pool().get().map_err(|e| e.to_string())?;
let (url, token) = instance_auth(&conn, &instance_id)?;
gitea_api_get_user(&url, &token)
}
// ── Gitea API 辅助 ────────────────────────────────────────────────────────────
/// 读取实例的 (url, token)
pub(crate) fn instance_auth(
conn: &rusqlite::Connection,
instance_id: &str,
) -> Result<(String, String), String> {
instance_auth_with_store(
conn,
&crate::credential_store::WindowsCredentialStore,
instance_id,
)
}
fn instance_auth_with_store(
conn: &rusqlite::Connection,
store: &impl crate::credential_store::CredentialStore,
instance_id: &str,
) -> Result<(String, String), String> {
let url: String = conn
.query_row(
"SELECT url FROM gitea_instances WHERE id = ?1",
params![instance_id],
|r| r.get(0),
)
.map_err(|_| "Gitea 实例不存在".to_string())?;
let token = crate::credential_store::get_or_migrate_legacy(
conn,
store,
crate::credential_store::LegacyCredentialKind::GiteaAccessToken,
instance_id,
)?;
let token = String::from_utf8(token)
.map_err(|_| "凭证状态 needs_repair:Gitea token 编码无效".to_string())?;
Ok((url, token))
}
/// 把 ureq 错误转成可读信息(尽量提取 Gitea 返回的 message 字段)
fn api_err(prefix: &str, e: ureq::Error, token: &str) -> String {
let detail = match e {
ureq::Error::Status(code, resp) => {
let msg = resp
.into_json::<serde_json::Value>()
.ok()
.and_then(|v| v["message"].as_str().map(|s| s.to_string()))
.unwrap_or_default();
if msg.is_empty() {
format!("{prefix}: HTTP {code}")
} else {
format!("{prefix}: {msg}(HTTP {code})")
}
}
other => format!("{prefix}: {other}"),
};
if token.is_empty() {
detail
} else {
detail.replace(token, "[redacted]")
}
}
/// `get_delivery_status` 使用的 Gitea 投影。它只反映远端信号,不给出发布决策。
#[derive(Debug, Clone)]
pub struct GiteaDeliverySignals {
pub pull_request: crate::commands::delivery_status::DeliverySignal,
pub ci: crate::commands::delivery_status::DeliverySignal,
pub review: crate::commands::delivery_status::DeliverySignal,
}
type DeliverySignal = crate::commands::delivery_status::DeliverySignal;
fn unavailable_delivery_signals(detail: impl Into<String>) -> GiteaDeliverySignals {
let detail = detail.into();
GiteaDeliverySignals {
pull_request: DeliverySignal::new(
"unavailable",
"gitea",
detail.clone(),
serde_json::json!({}),
),
ci: DeliverySignal::new(
"unavailable",
"gitea",
detail.clone(),
serde_json::json!({}),
),
review: DeliverySignal::new("unavailable", "gitea", detail, serde_json::json!({})),
}
}
fn gitea_get_json(
base_url: &str,
token: &str,
path: &str,
label: &str,
) -> Result<serde_json::Value, String> {
ureq::get(&format!("{base_url}{path}"))
.set("Authorization", &format!("token {token}"))
.call()
.map_err(|e| api_err(label, e, token))?
.into_json()
.map_err(|e| format!("{label}: 解析响应失败: {e}"))
}
/// 从 Gitea 拉取当前分支的 PR、当前 SHA 的 CI 以及该 PR 的 review 摘要。
///
/// 远端不可用必须作为数据返回,以便上层继续展示本地 Git 与部署未知状态。
pub fn get_delivery_signals(
conn: &rusqlite::Connection,
project_id: &str,
branch: Option<&str>,
commit_sha: Option<&str>,
) -> GiteaDeliverySignals {
let binding: rusqlite::Result<(String, String, String)> = conn.query_row(
"SELECT instance_id, gitea_owner, gitea_repo_name FROM gitea_repos WHERE project_id = ?1",
params![project_id],
|r| Ok((r.get(0)?, r.get(1)?, r.get(2)?)),
);
let (instance_id, owner, repo) = match binding {
Ok(binding) => binding,
Err(rusqlite::Error::QueryReturnedNoRows) => {
let detail = "项目未绑定 Gitea 仓库";
return GiteaDeliverySignals {
pull_request: DeliverySignal::new(
"not_configured",
"gitea",
detail,
serde_json::json!({}),
),
ci: DeliverySignal::new("not_configured", "gitea", detail, serde_json::json!({})),
review: DeliverySignal::new(
"not_configured",
"gitea",
detail,
serde_json::json!({}),
),
};
}
Err(error) => return unavailable_delivery_signals(format!("读取 Gitea 绑定失败:{error}")),
};
let (base_url, token) = match instance_auth(conn, &instance_id) {
Ok(auth) => auth,
Err(error) => return unavailable_delivery_signals(error),
};
let pull_path =
format!("/api/v1/repos/{owner}/{repo}/pulls?state=open&limit=50&type=pullrequest");
let pull_requests = match gitea_get_json(&base_url, &token, &pull_path, "读取 Gitea PR 失败")
{
Ok(value) => value,
Err(error) => return unavailable_delivery_signals(error),
};
if !pull_requests.is_array() {
return unavailable_delivery_signals("Gitea PR 响应不是数组");
}
let matched_pr = branch.and_then(|name| select_open_pull_request(&pull_requests, name));
let pull_request = match (branch, matched_pr) {
(None, _) => DeliverySignal::new(
"unknown",
"gitea",
"本地 Git 未提供当前分支,无法匹配开放 PR",
serde_json::json!({}),
),
(Some(_), Some(pr)) => DeliverySignal::new(
"open",
"gitea",
"当前分支存在开放 PR",
pull_request_summary(pr),
),
(Some(_), None) => DeliverySignal::new(
"none",
"gitea",
"当前分支没有开放 PR",
serde_json::json!({}),
),
};
let ci = match commit_sha {
Some(sha) => {
let path =
format!("/api/v1/repos/{owner}/{repo}/statuses/{sha}?sort=recentupdate&limit=50");
match gitea_get_json(&base_url, &token, &path, "读取 Gitea CI 状态失败") {
Ok(statuses) => summarize_ci_statuses(&statuses, Some(sha)),
Err(error) => {
DeliverySignal::new("unavailable", "gitea", error, serde_json::json!({}))
}
}
}
None => DeliverySignal::new(
"unknown",
"gitea",
"本地 Git 未提供最近 commit SHA,无法查询 CI",
serde_json::json!({}),
),
};
let review = match matched_pr.and_then(|pr| pr["number"].as_i64()) {
Some(number) => {
let path = format!("/api/v1/repos/{owner}/{repo}/pulls/{number}/reviews");
match gitea_get_json(&base_url, &token, &path, "读取 Gitea review 状态失败") {
Ok(reviews) => summarize_reviews(&reviews),
Err(error) => {
DeliverySignal::new("unavailable", "gitea", error, serde_json::json!({}))
}
}
}
None if branch.is_some() => DeliverySignal::new(
"not_applicable",
"gitea",
"当前分支没有开放 PR,因此无 review 状态",
serde_json::json!({}),
),
None => DeliverySignal::new(
"unknown",
"gitea",
"无法确定当前分支,未查询 review",
serde_json::json!({}),
),
};
GiteaDeliverySignals {
pull_request,
ci,
review,
}
}
pub fn select_open_pull_request<'a>(
pull_requests: &'a serde_json::Value,
branch: &str,
) -> Option<&'a serde_json::Value> {
pull_requests.as_array()?.iter().find(|pr| {
pr["head"]["ref"].as_str() == Some(branch)
|| pr["head"]["label"]
.as_str()
.map(|label| label.ends_with(&format!(":{branch}")) || label == branch)
.unwrap_or(false)
})
}
fn pull_request_summary(pr: &serde_json::Value) -> serde_json::Value {
serde_json::json!({
"number": pr["number"],
"title": pr["title"],
"url": pr["html_url"],
"head": pr["head"]["ref"],
"base": pr["base"]["ref"],
"updated_at": pr["updated_at"],
})
}
fn status_timestamp(status: &serde_json::Value) -> Option<chrono::DateTime<chrono::FixedOffset>> {
status["updated_at"]
.as_str()
.or_else(|| status["created_at"].as_str())
.and_then(|value| chrono::DateTime::parse_from_rfc3339(value).ok())
}
fn normalized_status_state(status: &serde_json::Value) -> String {
let raw_state = status["status"]
.as_str()
.or_else(|| status["state"].as_str())
.unwrap_or("unknown")
.to_ascii_lowercase();
let description = status["description"]
.as_str()
.unwrap_or("")
.trim()
.to_ascii_lowercase();
// Gitea 1.22 Actions 会把条件未命中的 job 记录为 status=success,
// 仅在 description 中标明 "Has been skipped"。部署信号必须还原成 skipped,
// 否则会把“未部署”误报为“部署成功”。
if raw_state == "skipped" || matches!(description.as_str(), "has been skipped" | "skipped") {
"skipped".to_string()
} else {
raw_state
}
}
fn latest_statuses_by_context<'a>(
items: &'a [serde_json::Value],
) -> std::collections::BTreeMap<String, &'a serde_json::Value> {
let mut latest: std::collections::BTreeMap<String, &'a serde_json::Value> =
std::collections::BTreeMap::new();
for item in items {
let context = item["context"].as_str().unwrap_or("unknown").to_string();
let should_replace = latest
.get(&context)
.map(
|existing| match (status_timestamp(item), status_timestamp(existing)) {
(Some(candidate), Some(current)) => candidate > current,
(Some(_), None) => true,
_ => false,
},
)
.unwrap_or(true);
if should_replace {
latest.insert(context, item);
}
}
latest
}
fn status_context_summary(context: &str, item: &serde_json::Value) -> serde_json::Value {
let state = normalized_status_state(item);
serde_json::json!({
"context": context,
"state": state,
"description": item["description"],
"target_url": item["target_url"],
"created_at": item["created_at"],
"updated_at": item["updated_at"],
})
}
/// 把 Gitea commit status 归并成稳定枚举;保留每个 context 的可追溯字段,供 Agent 自行判断门禁。
pub fn summarize_ci_statuses(
statuses: &serde_json::Value,
target_sha: Option<&str>,
) -> DeliverySignal {
let Some(items) = statuses.as_array() else {
return DeliverySignal::new(
"unknown",
"gitea",
"Gitea CI 响应不是数组",
serde_json::json!({}),
);
};
if items.is_empty() {
return DeliverySignal::new(
"unknown",
"gitea",
"当前 commit 暂无 CI 记录",
serde_json::json!({ "target_sha": target_sha, "contexts": [] }),
);
}
let latest = latest_statuses_by_context(items);
let contexts: Vec<serde_json::Value> = latest
.iter()
.map(|(context, item)| status_context_summary(context, item))
.collect();
let states: Vec<String> = contexts
.iter()
.filter_map(|item| item["state"].as_str())
.map(|state| state.to_ascii_lowercase())
.collect();
let state = if states
.iter()
.any(|state| state == "failure" || state == "error")
{
"failed"
} else if states.iter().all(|state| state == "skipped") {
"not_applicable"
} else if states
.iter()
.all(|state| state == "success" || state == "skipped")
{
"passed"
} else {
"pending"
};
let detail = match state {
"passed" => "所有最新 CI context 均成功",
"failed" => "至少一个最新 CI context 失败",
"not_applicable" => "所有最新 CI context 均未执行",
_ => "存在尚未成功的 CI context",
};
DeliverySignal::new(
state,
"gitea",
detail,
serde_json::json!({ "target_sha": target_sha, "contexts": contexts }),
)
}
/// 从同一批 commit statuses 中按项目声明的精确 context 提取部署过程信号。
///
/// 本函数不猜测 job 名,也不把普通 CI 成功当作部署成功;context 的所有权属于项目 profile。
/// P3-D 接入双证据聚合前先由 P3-A 固化并验证纯函数契约。
#[allow(dead_code)]
pub fn summarize_status_context(
statuses: &serde_json::Value,
target_sha: &str,
expected_context: &str,
) -> DeliverySignal {
let Some(items) = statuses.as_array() else {
return DeliverySignal::new(
"unknown",
"gitea",
"Gitea status 响应不是数组",
serde_json::json!({ "target_sha": target_sha, "expected_context": expected_context }),
);
};
let latest = latest_statuses_by_context(items);
let available_contexts: Vec<&str> = latest.keys().map(String::as_str).collect();
let Some(status) = latest.get(expected_context) else {
return DeliverySignal::new(
"unknown",
"gitea",
"目标 deployment context 尚无状态",
serde_json::json!({
"target_sha": target_sha,
"expected_context": expected_context,
"available_contexts": available_contexts,
}),
);
};
let raw_state = normalized_status_state(status);
let state = match raw_state.as_str() {
"success" => "success",
"failure" | "error" => "failure",
"pending" => "pending",
"warning" => "warning",
"skipped" => "skipped",
_ => "unknown",
};
let detail = match state {
"success" => "目标 deployment context 已成功",
"failure" => "目标 deployment context 已失败",
"pending" => "目标 deployment context 尚未结束",
"warning" => "目标 deployment context 带警告结束",
"skipped" => "目标 deployment context 未执行",
_ => "目标 deployment context 返回未知状态",
};
DeliverySignal::new(
state,
"gitea",
detail,
serde_json::json!({
"target_sha": target_sha,
"status": status_context_summary(expected_context, status),
"available_contexts": available_contexts,
}),
)
}
/// production 目标始终来自 Gitea 默认分支,不取本地工作分支的 HEAD。
/// 这样 Agent 在 feature branch 上开发时也不会把未合并 SHA 当成线上目标。
pub fn get_production_context_signals(
conn: &rusqlite::Connection,
project_id: &str,
contexts: &[String],
) -> Result<(String, std::collections::BTreeMap<String, DeliverySignal>), DeliverySignal> {
get_production_context_signals_with_store(
conn,
&crate::credential_store::WindowsCredentialStore,
project_id,
contexts,
)
}
fn get_production_context_signals_with_store(
conn: &rusqlite::Connection,
store: &impl crate::credential_store::CredentialStore,
project_id: &str,
contexts: &[String],
) -> Result<(String, std::collections::BTreeMap<String, DeliverySignal>), DeliverySignal> {
let binding: rusqlite::Result<(String, String, String)> = conn.query_row(
"SELECT instance_id, gitea_owner, gitea_repo_name FROM gitea_repos WHERE project_id = ?1",
params![project_id],
|row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)),
);
let (instance_id, owner, repo) = binding.map_err(|error| {
DeliverySignal::new(
if matches!(error, rusqlite::Error::QueryReturnedNoRows) {
"not_configured"
} else {
"unavailable"
},
"gitea",
format!("读取项目 Gitea 绑定失败:{error}"),
serde_json::json!({}),
)
})?;
let (base_url, token) =
instance_auth_with_store(conn, store, &instance_id).map_err(|error| {
DeliverySignal::new("unavailable", "gitea", error, serde_json::json!({}))
})?;
let agent = ureq::builder()
.timeout(std::time::Duration::from_secs(5))
.build();
let read = |path: &str, label: &str| {
agent
.get(&format!("{base_url}{path}"))
.set("Authorization", &format!("token {token}"))
.call()
.map_err(|error| api_err(label, error, &token))
.and_then(|response| {
response
.into_json::<serde_json::Value>()
.map_err(|error| format!("{label}: 解析响应失败: {error}"))
})
.map_err(|error| {
DeliverySignal::new("unavailable", "gitea", error, serde_json::json!({}))
})
};
let repository = read(
&format!("/api/v1/repos/{owner}/{repo}"),
"读取 Gitea 仓库失败",
)?;
let branch = repository["default_branch"]
.as_str()
.filter(|value| !value.is_empty())
.ok_or_else(|| {
DeliverySignal::new(
"unknown",
"gitea",
"仓库未返回默认分支",
serde_json::json!({}),
)
})?;
let branch_path = format!(
"/api/v1/repos/{owner}/{repo}/branches/{}",
urlencoding::encode(branch)
);
let branch_response = read(&branch_path, "读取 Gitea 默认分支失败")?;
let target_sha = branch_response["commit"]["id"]
.as_str()
.or_else(|| branch_response["commit"]["sha"].as_str())
.filter(|value| value.len() == 40 && value.bytes().all(|ch| ch.is_ascii_hexdigit()))
.ok_or_else(|| {
DeliverySignal::new(
"unknown",
"gitea",
"默认分支未返回完整 commit SHA",
serde_json::json!({}),
)
})?
.to_string();
let statuses = read(
&format!("/api/v1/repos/{owner}/{repo}/statuses/{target_sha}?sort=recentupdate&limit=50"),
"读取 Gitea 部署 context 失败",
)?;
let signals = contexts
.iter()
.map(|context| {
(
context.clone(),
summarize_status_context(&statuses, &target_sha, context),
)
})
.collect();
Ok((target_sha, signals))
}
/// 对每位 reviewer 仅保留提交时间最新的状态,避免旧的 approve 掩盖后续 request changes。
pub fn summarize_reviews(reviews: &serde_json::Value) -> DeliverySignal {
let Some(items) = reviews.as_array() else {
return DeliverySignal::new(
"unknown",
"gitea",
"Gitea review 响应不是数组",
serde_json::json!({}),
);
};
let mut latest: std::collections::BTreeMap<String, &serde_json::Value> =
std::collections::BTreeMap::new();
for (index, review) in items.iter().enumerate() {
let reviewer = review["user"]["login"]
.as_str()
.map(str::to_string)
.unwrap_or_else(|| format!("unknown-{index}"));
let should_replace = latest
.get(&reviewer)
.map(|existing| review_timestamp(review) >= review_timestamp(existing))
.unwrap_or(true);
if should_replace {
latest.insert(reviewer, review);
}
}
let reviewers: Vec<serde_json::Value> = latest
.iter()
.map(|(reviewer, review)| {
serde_json::json!({
"reviewer": reviewer,
"state": review["state"],
"submitted_at": review["submitted_at"],
})
})
.collect();
let states: Vec<String> = reviewers
.iter()
.filter_map(|review| review["state"].as_str())
.map(|state| state.to_ascii_uppercase())
.collect();
let state = if states.iter().any(|state| state == "REQUEST_CHANGES") {
"changes_requested"
} else if states.iter().any(|state| state == "APPROVED") {
"approved"
} else {
"pending"
};
let detail = match state {
"changes_requested" => "存在 reviewer 请求修改",
"approved" => "至少一位 reviewer 已批准,未发现后续请求修改",
_ => "尚无有效 approve/reject review",
};
DeliverySignal::new(
state,
"gitea",
detail,
serde_json::json!({ "reviewers": reviewers }),
)
}
fn review_timestamp(review: &serde_json::Value) -> &str {
review["submitted_at"].as_str().unwrap_or("")
}
/// 写入或更新项目 ↔ 仓库绑定(换绑时重置 webhook 字段,旧 hook 已随旧仓库失效)
fn upsert_repo_link(
conn: &rusqlite::Connection,
instance_id: &str,
project_id: &str,
owner: &str,
repo_name: &str,
clone_url: Option<&str>,
) -> Result<GiteaRepo, String> {
let old_reference: Option<String> = conn
.query_row(
"SELECT webhook_secret_ref FROM gitea_repos WHERE project_id = ?1",
params![project_id],
|row| row.get(0),
)
.optional()
.map_err(|_| "读取旧 webhook 凭证引用失败".to_string())?
.flatten();
let id = uuid::Uuid::new_v4().to_string();
conn.execute(
"INSERT INTO gitea_repos (id, instance_id, project_id, gitea_owner, gitea_repo_name, clone_url)
VALUES (?1, ?2, ?3, ?4, ?5, ?6)
ON CONFLICT(project_id) DO UPDATE SET
instance_id = ?2, gitea_owner = ?4, gitea_repo_name = ?5, clone_url = ?6,
webhook_secret = NULL, webhook_secret_ref = NULL, webhook_id = NULL",
params![id, instance_id, project_id, owner, repo_name, clone_url],
).map_err(|e| e.to_string())?;
let repo = conn
.query_row(
"SELECT id, instance_id, project_id, gitea_owner, gitea_repo_name,
clone_url, webhook_id, linked_at
FROM gitea_repos WHERE project_id = ?1",
params![project_id],
row_to_repo,
)
.map_err(|e| e.to_string())?;
if let Some(reference) = old_reference {
let _ = crate::credential_store::CredentialStore::delete(
&crate::credential_store::WindowsCredentialStore,
&reference,
);
}
Ok(repo)
}
/// GET /api/v1/user — 验证 token 并返回用户信息
pub fn gitea_api_get_user(base_url: &str, token: &str) -> Result<GiteaUser, String> {
let url = format!("{}/api/v1/user", base_url);
let resp = ureq::get(&url)
.set("Authorization", &format!("token {token}"))
.set("Content-Type", "application/json")
.call()
.map_err(|e| format!("Gitea 连接失败: {e}"))?;
let json: serde_json::Value = resp.into_json().map_err(|e| format!("解析响应失败: {e}"))?;
let login = json["login"]
.as_str()
.ok_or_else(|| "token 无效或权限不足".to_string())?
.to_string();
Ok(GiteaUser {
login,
full_name: json["full_name"].as_str().unwrap_or("").to_string(),
email: json["email"].as_str().unwrap_or("").to_string(),
})
}
/// GET /api/v1/repos/:owner/:repo — 获取仓库信息(G2 仓库绑定时使用)
pub fn gitea_api_get_repo(
base_url: &str,
token: &str,
owner: &str,
repo: &str,
) -> Result<serde_json::Value, String> {
let url = format!("{}/api/v1/repos/{owner}/{repo}", base_url);
let resp = ureq::get(&url)
.set("Authorization", &format!("token {token}"))
.call()
.map_err(|e| api_err("获取仓库信息失败", e, token))?;
resp.into_json().map_err(|e| format!("解析失败: {e}"))
}
// ── 仓库创建 / 迁移 / 绑定(G2)──────────────────────────────────────────────
/// 在 Gitea 新建仓库并绑定到项目
#[tauri::command]
pub fn gitea_create_repo(
instance_id: String,
project_id: String,
repo_name: String,
description: Option<String>,
private: Option<bool>,
) -> Result<GiteaRepo, String> {
let conn = crate::db::pool().get().map_err(|e| e.to_string())?;
let (base_url, token) = instance_auth(&conn, &instance_id)?;
let resp = ureq::post(&format!("{base_url}/api/v1/user/repos"))
.set("Authorization", &format!("token {token}"))
.send_json(serde_json::json!({
"name": repo_name,
"description": description.unwrap_or_default(),
"private": private.unwrap_or(true),
"auto_init": false,
}))
.map_err(|e| api_err("创建仓库失败", e, &token))?;
let json: serde_json::Value = resp.into_json().map_err(|e| format!("解析响应失败: {e}"))?;
let owner = json["owner"]["login"]
.as_str()
.ok_or("响应缺少 owner 字段")?
.to_string();
let clone_url = json["clone_url"].as_str().map(|s| s.to_string());
upsert_repo_link(
&conn,
&instance_id,
&project_id,
&owner,
&repo_name,
clone_url.as_deref(),
)
}
/// 从 GitHub/GitLab 等迁移仓库到 Gitea 并绑定到项目
#[tauri::command]
pub fn gitea_migrate_repo(
instance_id: String,
project_id: String,
clone_addr: String,
repo_name: String,
auth_token: Option<String>,
private: Option<bool>,
) -> Result<GiteaRepo, String> {
let conn = crate::db::pool().get().map_err(|e| e.to_string())?;
let (base_url, token) = instance_auth(&conn, &instance_id)?;
let mut body = serde_json::json!({
"clone_addr": clone_addr,
"repo_name": repo_name,
"private": private.unwrap_or(true),
"mirror": false,
"service": "git",
});
if let Some(t) = auth_token.filter(|t| !t.trim().is_empty()) {
body["auth_token"] = serde_json::Value::String(t);
}
let resp = ureq::post(&format!("{base_url}/api/v1/repos/migrate"))
.set("Authorization", &format!("token {token}"))
.send_json(body)
.map_err(|e| api_err("迁移仓库失败", e, &token))?;
let json: serde_json::Value = resp.into_json().map_err(|e| format!("解析响应失败: {e}"))?;
let owner = json["owner"]["login"]
.as_str()
.ok_or("响应缺少 owner 字段")?
.to_string();
let clone_url = json["clone_url"].as_str().map(|s| s.to_string());
upsert_repo_link(
&conn,
&instance_id,
&project_id,
&owner,
&repo_name,
clone_url.as_deref(),
)
}
/// 绑定 Gitea 上已存在的仓库到项目(先验证仓库存在)
#[tauri::command]
pub fn gitea_link_repo(
instance_id: String,
project_id: String,
owner: String,
repo_name: String,
) -> Result<GiteaRepo, String> {
let conn = crate::db::pool().get().map_err(|e| e.to_string())?;
let (base_url, token) = instance_auth(&conn, &instance_id)?;
let json = gitea_api_get_repo(&base_url, &token, &owner, &repo_name)?;
let clone_url = json["clone_url"].as_str().map(|s| s.to_string());
upsert_repo_link(
&conn,
&instance_id,
&project_id,
&owner,
&repo_name,
clone_url.as_deref(),
)
}
// ── Webhook 注册 / 解绑(G4)─────────────────────────────────────────────────
/// 为项目绑定的仓库注册 push webhook。
/// callback_base 形如 http://192.168.1.5:27190(Gitea 服务器视角可达的宿主机地址)。
/// 先建新 hook 并持久化凭证,再清理旧 hook;失败时旧 hook 保持可用。
#[tauri::command]
pub fn gitea_setup_webhook(project_id: String, callback_base: String) -> Result<GiteaRepo, String> {
let conn = crate::db::pool().get().map_err(|e| e.to_string())?;
setup_webhook_with_store(
&conn,
&crate::credential_store::WindowsCredentialStore,
&project_id,
&callback_base,
)
}
fn setup_webhook_with_store(
conn: &rusqlite::Connection,
store: &impl crate::credential_store::CredentialStore,
project_id: &str,
callback_base: &str,
) -> Result<GiteaRepo, String> {
let link = conn
.query_row(
"SELECT id, instance_id, project_id, gitea_owner, gitea_repo_name,
clone_url, webhook_id, linked_at
FROM gitea_repos WHERE project_id = ?1",
params![project_id],
row_to_repo,
)
.map_err(|_| "项目尚未绑定 Gitea 仓库".to_string())?;
let (base_url, token) = instance_auth_with_store(conn, store, &link.instance_id)?;
let secret = format!(
"{}{}",
uuid::Uuid::new_v4().simple(),
uuid::Uuid::new_v4().simple()
);
let callback_url = format!(
"{}/webhook/gitea/{}",
callback_base.trim_end_matches('/'),
project_id
);
let reference = crate::credential_store::versioned_reference(
crate::credential_store::LegacyCredentialKind::GiteaWebhookSecret,
&project_id,
);
crate::credential_store::put_verified(store, &reference, secret.as_bytes())?;
let create_result = ureq::post(&format!(
"{base_url}/api/v1/repos/{}/{}/hooks",
link.gitea_owner, link.gitea_repo_name
))
.set("Authorization", &format!("token {token}"))
.send_json(serde_json::json!({
"type": "gitea",
"active": true,
"branch_filter": "*",
"events": ["push", "pull_request"],
"config": {
"url": callback_url,
"content_type": "json",
"secret": secret,
"http_method": "post",
},
}))
.map_err(|e| api_err("注册 webhook 失败", e, &token).replace(&secret, "[redacted]"));
let resp = match create_result {
Ok(resp) => resp,
Err(error) => {
let _ = store.delete(&reference);
return Err(error);
}
};
let json: serde_json::Value = match resp.into_json() {
Ok(json) => json,
Err(_) => {
let _ = store.delete(&reference);
return Err("webhook 状态 needs_repair:远端已接收注册,但响应无法解析;旧 webhook 保持不变,请检查远端重复 hook".into());
}
};
let hook_id = match json["id"].as_i64() {
Some(id) => id,
None => {
let _ = store.delete(&reference);
return Err("webhook 状态 needs_repair:远端未返回 hook id;旧 webhook 保持不变,请检查远端重复 hook".into());
}
};
if let Err(error) = persist_webhook_secret(conn, store, project_id, &reference, hook_id) {
let _ = ureq::delete(&format!(
"{base_url}/api/v1/repos/{}/{}/hooks/{hook_id}",
link.gitea_owner, link.gitea_repo_name
))
.set("Authorization", &format!("token {token}"))
.call();
let _ = store.delete(&reference);
return Err(error);
}
if let Some(old_id) = link.webhook_id {
let _ = ureq::delete(&format!(
"{base_url}/api/v1/repos/{}/{}/hooks/{old_id}",
link.gitea_owner, link.gitea_repo_name
))
.set("Authorization", &format!("token {token}"))
.call();
}
conn.query_row(
"SELECT id, instance_id, project_id, gitea_owner, gitea_repo_name,
clone_url, webhook_id, linked_at
FROM gitea_repos WHERE project_id = ?1",
params![project_id],
row_to_repo,
)
.map_err(|e| e.to_string())
}
fn persist_webhook_secret(
conn: &rusqlite::Connection,
store: &impl crate::credential_store::CredentialStore,
project_id: &str,
reference: &str,
hook_id: i64,
) -> Result<(), String> {
let old_reference: Option<String> = conn
.query_row(
"SELECT webhook_secret_ref FROM gitea_repos WHERE project_id = ?1",
params![project_id],
|row| row.get(0),
)
.map_err(|_| "webhook 绑定不存在".to_string())?;
store
.get(reference)
.map_err(|_| "凭证状态 needs_repair:受保护存储不可用".to_string())?
.ok_or_else(|| "凭证状态 needs_repair:新 webhook 凭证不存在".to_string())?;
let tx = conn
.unchecked_transaction()
.map_err(|_| "webhook 状态 needs_repair:数据库切换失败".to_string())?;
let changed = tx.execute(
"UPDATE gitea_repos SET webhook_secret = NULL, webhook_secret_ref = ?1, webhook_id = ?2 WHERE project_id = ?3 AND webhook_secret_ref IS ?4",
params![reference, hook_id, project_id, old_reference],
).map_err(|_| "webhook 状态 needs_repair:数据库切换失败".to_string())?;
if changed != 1 {
return Err("webhook 状态 needs_repair:绑定已变化".into());
}
tx.commit()
.map_err(|_| "webhook 状态 needs_repair:数据库切换失败".to_string())?;
if let Some(old) = old_reference {
let _ = store.delete(&old);
}
Ok(())
}
/// 解绑项目与 Gitea 仓库(顺带删除 Gitea 侧 webhook,失败不阻断)
#[tauri::command]
pub fn gitea_unlink_repo(project_id: String) -> Result<(), String> {
let conn = crate::db::pool().get().map_err(|e| e.to_string())?;
let link = match conn.query_row(
"SELECT id, instance_id, project_id, gitea_owner, gitea_repo_name,
clone_url, webhook_id, linked_at
FROM gitea_repos WHERE project_id = ?1",
params![project_id],
row_to_repo,
) {
Ok(l) => l,
Err(rusqlite::Error::QueryReturnedNoRows) => return Ok(()),
Err(e) => return Err(e.to_string()),
};
let old_reference: Option<String> = conn
.query_row(
"SELECT webhook_secret_ref FROM gitea_repos WHERE project_id = ?1",
params![project_id],
|row| row.get(0),
)
.map_err(|_| "读取 webhook 凭证引用失败".to_string())?;
if let Some(hook_id) = link.webhook_id {
if let Ok((base_url, token)) = instance_auth(&conn, &link.instance_id) {
let _ = ureq::delete(&format!(
"{base_url}/api/v1/repos/{}/{}/hooks/{hook_id}",
link.gitea_owner, link.gitea_repo_name
))
.set("Authorization", &format!("token {token}"))
.call();
}
}
conn.execute(
"DELETE FROM gitea_repos WHERE project_id = ?1",
params![project_id],
)
.map_err(|e| e.to_string())?;
if let Some(reference) = old_reference {
let _ = crate::credential_store::CredentialStore::delete(
&crate::credential_store::WindowsCredentialStore,
&reference,
);
}
Ok(())
}
// ── 远端轮询采集(G6:webhook 不可达时的降级路径)────────────────────────────
/// 从 Gitea commits API 响应解析出 (sha, message, date) 列表(纯函数,可测)
pub fn parse_gitea_commits(json: &serde_json::Value) -> Vec<(String, String, Option<String>)> {
json.as_array()
.map(|arr| {
arr.iter()
.filter_map(|c| {
let sha = c["sha"].as_str()?.to_string();
let message = c["commit"]["message"].as_str().unwrap_or("").to_string();
let date = c["commit"]["committer"]["date"]
.as_str()
.or_else(|| c["created"].as_str())
.map(|s| s.to_string());
Some((sha, message, date))
})
.collect()
})
.unwrap_or_default()
}
/// 轮询所有绑定仓库的最新 commit,幂等写入 commit_metrics。
/// 返回 (轮询仓库数, 新写入 commit 数)。纯出站 HTTPS,适配公网 Gitea + 内网炼境拓扑。
pub fn poll_all_repos() -> Result<(usize, usize), String> {
let conn = crate::db::pool().get().map_err(|e| e.to_string())?;
let repos: Vec<(String, String, String, String)> = {
let mut stmt = conn
.prepare(
"SELECT gr.project_id, gr.gitea_owner, gr.gitea_repo_name, gi.id
FROM gitea_repos gr JOIN gitea_instances gi ON gi.id = gr.instance_id",
)
.map_err(|e| e.to_string())?;
let rows = stmt
.query_map([], |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?, r.get(3)?)))
.map_err(|e| e.to_string())?;
rows.collect::<rusqlite::Result<Vec<_>>>()
.map_err(|e| e.to_string())?
};
let mut ingested = 0usize;
let mut polled = 0usize;
for (project_id, owner, repo, instance_id) in &repos {
let (base_url, token) = instance_auth(&conn, instance_id)?;
let url = format!(
"{base_url}/api/v1/repos/{owner}/{repo}/commits?limit=50&stat=false&verification=false"
);
let resp = match ureq::get(&url)
.set("Authorization", &format!("token {token}"))
.call()
{
Ok(r) => r,
Err(e) => {
crate::db::log_event(
"gitea_poll",
"warn",
&format!(
"轮询 {owner}/{repo} 失败: {}",
api_err("拉取 commits", e, &token)
),
None,
);
continue;
}
};
let json: serde_json::Value = match resp.into_json() {
Ok(j) => j,
Err(_) => continue,
};
polled += 1;
// U3:按 project_id 加载规则 ID 集合(每仓库一次),剔除 trailer 里的未知 ID
let valid_rules =
crate::commands::commit_metrics::load_rule_ids_by_project_id(&conn, project_id);
for (sha, message, date) in parse_gitea_commits(&json) {
let before: i64 = conn
.query_row(
"SELECT COUNT(*) FROM commit_metrics WHERE sha = ?1",
params![sha],
|r| r.get(0),
)
.unwrap_or(0);
if before == 0
&& crate::commands::commit_metrics::record_commit_validated(
&conn,
project_id,
&sha,
&message,
date.as_deref(),
valid_rules.as_ref(),
)
.is_ok()
{
ingested += 1;
}
}
}
if ingested > 0 {
crate::db::log_event(
"gitea_poll",
"info",
&format!("轮询 {polled} 个仓库,新采集 {ingested} 条 commit"),
None,
);
}
// PR 状态轮询:弥补公网 Gitea + 内网炼境时 webhook 不可达的场景
let mut pr_ingested = 0usize;
for (project_id, owner, repo, instance_id) in &repos {
let (base_url, token) = instance_auth(&conn, instance_id)?;
pr_ingested += poll_pr_events(&conn, project_id, owner, repo, &base_url, &token);
}
if pr_ingested > 0 {
crate::db::log_event(
"gitea_poll",
"info",
&format!("PR 轮询新采集 {pr_ingested} 条事件"),
None,
);
}
// staleness 自知:至少成功拉到一个仓库才算一次有效同步
if polled > 0 {
crate::db::touch_sync_state(&conn, "flywheel");
}
Ok((polled, ingested))
}
/// 轮询单个仓库的 PR 状态,将未记录的事件写入 pr_events,返回新增条数
fn poll_pr_events(
conn: &rusqlite::Connection,
project_id: &str,
owner: &str,
repo: &str,
base_url: &str,
token: &str,
) -> usize {
let mut count = 0;
for state in ["open", "closed"] {
let url = format!(
"{base_url}/api/v1/repos/{owner}/{repo}/pulls?state={state}&limit=50&type=pullrequest"
);
let resp = match ureq::get(&url)
.set("Authorization", &format!("token {token}"))
.call()
{
Ok(r) => r,
Err(_) => continue,
};
let prs: serde_json::Value = match resp.into_json() {
Ok(j) => j,
Err(_) => continue,
};
let prs = match prs.as_array() {
Some(a) => a,
None => continue,
};
for pr in prs {
let pr_number = match pr["number"].as_i64() {
Some(n) => n,
None => continue,
};
let title = pr["title"].as_str().unwrap_or("").to_string();
let head_branch = pr["head"]["ref"].as_str().unwrap_or("").to_string();
let base_branch = pr["base"]["ref"].as_str().unwrap_or("").to_string();
let author = pr["user"]["login"].as_str().unwrap_or("").to_string();
let merged = pr["merged"].as_bool().unwrap_or(false);
let pr_state = pr["state"].as_str().unwrap_or("open");
// 确定动作:已合并 > 关闭 > 开启
let action = if merged {
"merged"
} else if pr_state == "closed" {
"closed"
} else {
"opened"
};
let existing: i64 = conn.query_row(
"SELECT COUNT(*) FROM pr_events WHERE project_id = ?1 AND pr_number = ?2 AND action = ?3",
params![project_id, pr_number, action],
|r| r.get(0),
).unwrap_or(0);
if existing == 0 {
let id = uuid::Uuid::new_v4().to_string();
if conn.execute(
"INSERT INTO pr_events (id, project_id, pr_number, title, action, head_branch, base_branch, author)
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)",
params![id, project_id, pr_number, title, action, head_branch, base_branch, author],
).is_ok() {
count += 1;
}
}
}
}
count
}
/// 手动触发一次轮询(UI「立即拉取」按钮)
#[tauri::command]
pub fn gitea_poll_now() -> Result<serde_json::Value, String> {
let (polled, ingested) = poll_all_repos()?;
Ok(serde_json::json!({ "polled": polled, "ingested": ingested }))
}
// ── 仓库绑定查询命令 ──────────────────────────────────────────────────────────
/// 查询所有已绑定的仓库(含实例名)
#[tauri::command]
pub fn gitea_list_linked_repos() -> Result<Vec<serde_json::Value>, String> {
let conn = crate::db::pool().get().map_err(|e| e.to_string())?;
let mut stmt = conn
.prepare(
"SELECT gr.id, gr.instance_id, gi.name as instance_name, gi.url as instance_url,
gr.project_id, gr.gitea_owner, gr.gitea_repo_name,
gr.clone_url, (gr.webhook_secret IS NOT NULL OR gr.webhook_secret_ref IS NOT NULL), gr.webhook_id, gr.linked_at
FROM gitea_repos gr
JOIN gitea_instances gi ON gi.id = gr.instance_id
ORDER BY gr.linked_at DESC",
)
.map_err(|e| e.to_string())?;
let rows = stmt
.query_map([], |row| {
Ok(serde_json::json!({
"id": row.get::<_, String>(0)?,
"instance_id": row.get::<_, String>(1)?,
"instance_name": row.get::<_, String>(2)?,
"instance_url": row.get::<_, String>(3)?,
"project_id": row.get::<_, String>(4)?,
"gitea_owner": row.get::<_, String>(5)?,
"gitea_repo_name": row.get::<_, String>(6)?,
"clone_url": row.get::<_, Option<String>>(7)?,
"has_webhook": row.get::<_, bool>(8)?,
"webhook_id": row.get::<_, Option<i64>>(9)?,
"linked_at": row.get::<_, Option<String>>(10)?,
}))
})
.map_err(|e| e.to_string())?
.collect::<rusqlite::Result<Vec<_>>>()
.map_err(|e| e.to_string())?;
Ok(rows)
}
/// 通过炼境存储的凭证为项目创建 PR,返回 JSON 字符串(含 pr_url / pr_number / already_exists)。
/// 409 冲突(PR 已存在)时自动查找并返回现有 PR,already_exists=true。
pub fn gitea_api_create_pull_request(
project_id: &str,
head: &str,
base: &str,
title: &str,
body: &str,
) -> Result<String, String> {
let conn = crate::db::pool().get().map_err(|e| e.to_string())?;
create_pull_request_with_store(
&conn,
&crate::credential_store::WindowsCredentialStore,
project_id,
head,
base,
title,
body,
)
}
fn create_pull_request_with_store(
conn: &rusqlite::Connection,
store: &impl crate::credential_store::CredentialStore,
project_id: &str,
head: &str,
base: &str,
title: &str,
body: &str,
) -> Result<String, String> {
let (instance_id, owner, repo_name): (String, String, String) = conn.query_row(
"SELECT instance_id, gitea_owner, gitea_repo_name FROM gitea_repos WHERE project_id = ?1",
params![project_id],
|r| Ok((r.get(0)?, r.get(1)?, r.get(2)?)),
).map_err(|_| format!("项目 {} 未绑定 Gitea 仓库", project_id))?;
let (base_url, token) = instance_auth_with_store(conn, store, &instance_id)?;
let create_url = format!("{base_url}/api/v1/repos/{owner}/{repo_name}/pulls");
let (json, already_exists) = match ureq::post(&create_url)
.set("Authorization", &format!("token {token}"))
.set("Content-Type", "application/json")
.send_json(serde_json::json!({ "head": head, "base": base, "title": title, "body": body }))
{
Ok(resp) => (
resp.into_json::<serde_json::Value>()
.map_err(|e| format!("解析响应失败: {e}"))?,
false,
),
Err(ureq::Error::Status(409, _)) => {
// PR 已存在,列出开放 PR 找到匹配的分支
let list_url =
format!("{base_url}/api/v1/repos/{owner}/{repo_name}/pulls?state=open&limit=30");
let prs: serde_json::Value = ureq::get(&list_url)
.set("Authorization", &format!("token {token}"))
.call()
.map_err(|e| api_err("获取已有 PR 列表失败", e, &token))?
.into_json()
.map_err(|e| format!("解析 PR 列表失败: {e}"))?;
let arr = prs.as_array().ok_or("PR 列表格式错误")?;
let existing = arr
.iter()
.find(|pr| {
pr["head"]["ref"].as_str() == Some(head)
|| pr["head"]["label"]
.as_str()
.map(|l| l.ends_with(head))
.unwrap_or(false)
})
.cloned()
.ok_or_else(|| format!("PR 已存在但找不到(head={head})"))?;
(existing, true)
}
Err(e) => return Err(api_err("创建 PR 失败", e, &token)),
};
let pr_number = json["number"].as_u64().unwrap_or(0);
let pr_url = json["html_url"].as_str().unwrap_or("").to_string();
let pr_title = json["title"].as_str().unwrap_or(title).to_string();
serde_json::to_string(&serde_json::json!({
"pr_number": pr_number,
"pr_url": pr_url,
"head": head,
"base": base,
"title": pr_title,
"already_exists": already_exists,
}))
.map_err(|e| e.to_string())
}
/// 查询单个项目的 Gitea 绑定
#[tauri::command]
pub fn gitea_get_repo_link(project_id: String) -> Result<Option<GiteaRepo>, String> {
let conn = crate::db::pool().get().map_err(|e| e.to_string())?;
match conn.query_row(
"SELECT id, instance_id, project_id, gitea_owner, gitea_repo_name,
clone_url, webhook_id, linked_at
FROM gitea_repos WHERE project_id = ?1",
params![project_id],
row_to_repo,
) {
Ok(r) => Ok(Some(r)),
Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None),
Err(e) => Err(e.to_string()),
}
}
// ── MCP 工具层:CI 状态查询 + PR 合并 ────────────────────────────────────────
/// 查询某 commit 的所有 CI 检查状态,返回 Markdown 格式摘要。
/// 调用 GET /api/v1/repos/{owner}/{repo}/statuses/{sha}
pub fn gitea_api_get_ci_status(project_id: &str, commit_sha: &str) -> Result<String, String> {
let conn = crate::db::pool().get().map_err(|e| e.to_string())?;
let (instance_id, owner, repo_name): (String, String, String) = conn.query_row(
"SELECT instance_id, gitea_owner, gitea_repo_name FROM gitea_repos WHERE project_id = ?1",
params![project_id],
|r| Ok((r.get(0)?, r.get(1)?, r.get(2)?)),
).map_err(|_| format!("项目 {} 未绑定 Gitea 仓库", project_id))?;
let (base_url, token) = instance_auth(&conn, &instance_id)?;
let url = format!("{base_url}/api/v1/repos/{owner}/{repo_name}/statuses/{commit_sha}?limit=50");
let resp = ureq::get(&url)
.set("Authorization", &format!("token {token}"))
.call()
.map_err(|e| api_err("查询 CI 状态失败", e, &token))?;
let statuses: serde_json::Value = resp.into_json().map_err(|e| format!("解析响应失败: {e}"))?;
let arr = statuses
.as_array()
.ok_or_else(|| "响应格式错误".to_string())?;
let short_sha = &commit_sha[..8.min(commit_sha.len())];
if arr.is_empty() {
return Ok(format!(
"commit `{}` 暂无 CI 记录(CI 可能尚未触发,或 commit 不存在)",
short_sha
));
}
// 同名 context 只取最新一条
let mut latest: std::collections::HashMap<String, &serde_json::Value> =
std::collections::HashMap::new();
for s in arr {
let ctx = s["context"].as_str().unwrap_or("unknown").to_string();
latest.entry(ctx).or_insert(s);
}
let mut lines = vec![format!("## CI 状态:commit `{}`\n", short_sha)];
let mut all_pass = true;
let mut has_failure = false;
for (ctx, s) in &latest {
// Gitea 1.22 Actions 使用 "status" 字段;传统 commit status API 用 "state"
let state = s["status"]
.as_str()
.or_else(|| s["state"].as_str())
.unwrap_or("unknown");
let desc = s["description"].as_str().unwrap_or("");
let icon = match state {
"success" => "✅",
"failure" | "error" => {
all_pass = false;
has_failure = true;
"❌"
}
_ => {
all_pass = false;
"⏳"
}
};
lines.push(format!("- {} `{}` — {} {}", icon, ctx, state, desc));
}
let conclusion = if all_pass {
"✅ 全部通过,可以合并"
} else if has_failure {
"❌ 存在失败检查,暂不建议合并"
} else {
"⏳ 检查进行中,请稍后再查"
};
lines.push(format!("\n**结论:{}**", conclusion));
Ok(lines.join("\n"))
}
fn is_full_git_sha(value: &str) -> bool {
value.len() == 40 && value.chars().all(|ch| ch.is_ascii_hexdigit())
}
/// 合并门禁的纯函数部分。网络层先抓取同一时刻的 PR / CI / review,再一次性校验。
/// 返回已确认的实际 head SHA,供执行结果与审计信息复用。
fn validate_merge_preconditions(
confirmed: bool,
expected_head_sha: &str,
pull_request: &serde_json::Value,
ci: &DeliverySignal,
review: &DeliverySignal,
) -> Result<String, String> {
if !confirmed {
return Err("合并需要当前用户明确确认(confirmed=true)".to_string());
}
if !is_full_git_sha(expected_head_sha) {
return Err("expected_head_sha 必须是完整 40 位 Git SHA".to_string());
}
let state = pull_request["state"].as_str().unwrap_or("unknown");
if state != "open" || pull_request["merged"].as_bool() == Some(true) {
return Err(format!("PR 当前不可合并:state={state}"));
}
if pull_request["mergeable"].as_bool() == Some(false) {
return Err("PR 存在冲突或被 Gitea 判定为不可合并".to_string());
}
let actual_head_sha = pull_request["head"]["sha"]
.as_str()
.ok_or_else(|| "Gitea PR 响应缺少 head.sha,拒绝合并".to_string())?;
if actual_head_sha != expected_head_sha {
return Err(format!(
"PR head SHA 已变化:期望 {expected_head_sha},实际 {actual_head_sha};请重新检查后确认"
));
}
if ci.state != "passed" {
return Err(format!("CI 门禁未通过:{}({})", ci.state, ci.detail));
}
if ci.data["target_sha"].as_str() != Some(actual_head_sha) {
return Err("CI 证据未绑定当前 PR head SHA,拒绝合并".to_string());
}
match review.state.as_str() {
"approved" | "pending" => {}
"changes_requested" => return Err("review 门禁阻塞:存在 reviewer 请求修改".to_string()),
other => {
return Err(format!(
"review 状态不可用或未知:{other};为避免误合并已拒绝"
))
}
}
Ok(actual_head_sha.to_string())
}
fn merge_request_body(pr_number: i64, head_sha: &str) -> serde_json::Value {
serde_json::json!({
"Do": "merge",
"merge_title_field": format!("Merge PR #{pr_number}"),
"merge_message_field": "",
"head_commit_id": head_sha,
})
}
/// 合并指定 PR。所有高影响边界在能力本体内强制校验,MCP/UI 不得绕过。
/// 调用 POST /api/v1/repos/{owner}/{repo}/pulls/{index}/merge。
pub fn gitea_api_merge_pull_request(
conn: &rusqlite::Connection,
project_id: &str,
pr_number: i64,
expected_head_sha: &str,
confirmed: bool,
) -> Result<String, String> {
if !confirmed {
return Err("合并需要当前用户明确确认(confirmed=true)".to_string());
}
if !is_full_git_sha(expected_head_sha) {
return Err("expected_head_sha 必须是完整 40 位 Git SHA".to_string());
}
let (instance_id, owner, repo_name): (String, String, String) = conn.query_row(
"SELECT instance_id, gitea_owner, gitea_repo_name FROM gitea_repos WHERE project_id = ?1",
params![project_id],
|r| Ok((r.get(0)?, r.get(1)?, r.get(2)?)),
).map_err(|_| format!("项目 {} 未绑定 Gitea 仓库", project_id))?;
let (base_url, token) = instance_auth(conn, &instance_id)?;
let pull_path = format!("/api/v1/repos/{owner}/{repo_name}/pulls/{pr_number}");
let pull_request = gitea_get_json(&base_url, &token, &pull_path, "读取待合并 PR 失败")?;
let actual_head_sha = pull_request["head"]["sha"]
.as_str()
.ok_or_else(|| "Gitea PR 响应缺少 head.sha,拒绝合并".to_string())?;
let status_path = format!(
"/api/v1/repos/{owner}/{repo_name}/statuses/{actual_head_sha}?sort=recentupdate&limit=50"
);
let statuses = gitea_get_json(
&base_url,
&token,
&status_path,
"读取待合并 PR 的 CI 状态失败",
)?;
let ci = summarize_ci_statuses(&statuses, Some(actual_head_sha));
let reviews_path = format!("/api/v1/repos/{owner}/{repo_name}/pulls/{pr_number}/reviews");
let reviews = gitea_get_json(
&base_url,
&token,
&reviews_path,
"读取待合并 PR 的 review 状态失败",
)?;
let review = summarize_reviews(&reviews);
let validated_sha =
validate_merge_preconditions(confirmed, expected_head_sha, &pull_request, &ci, &review)?;
let url = format!("{base_url}/api/v1/repos/{owner}/{repo_name}/pulls/{pr_number}/merge");
ureq::post(&url)
.set("Authorization", &format!("token {token}"))
.set("Content-Type", "application/json")
.send_json(merge_request_body(pr_number, &validated_sha))
.map_err(|e| api_err("合并 PR 失败", e, &token))?;
serde_json::to_string_pretty(&serde_json::json!({
"merged": true,
"project_id": project_id,
"repository": format!("{owner}/{repo_name}"),
"pr_number": pr_number,
"head_sha": validated_sha,
"ci_state": ci.state,
"review_state": review.state,
}))
.map_err(|error| error.to_string())
}
// ── MCP 工具:一键建仓库 + 配置 remote ───────────────────────────────────────
/// 为项目在默认 Gitea 实例上创建远程仓库,并配置本地 git remote origin。
/// 幂等:仓库已存在(409)则直接绑定;remote 已配置则 set-url 更新。
pub fn gitea_api_setup_repo(
project_id: &str,
repo_name: Option<&str>,
private: bool,
) -> Result<String, String> {
let conn = crate::db::pool().get().map_err(|e| e.to_string())?;
// 项目基本信息
let (project_name, win_path): (String, Option<String>) = conn
.query_row(
"SELECT pp.name, pw.win_path
FROM project_workspaces pw
JOIN project_profiles pp ON pp.id = pw.profile_id
WHERE pw.id = ?1",
params![project_id],
|r| Ok((r.get(0)?, r.get(1)?)),
)
.map_err(|_| format!("项目 {} 不存在", project_id))?;
// 默认 Gitea 实例
let instance_id: String = conn
.query_row(
"SELECT id FROM gitea_instances WHERE is_default = 1 LIMIT 1",
[],
|r| r.get(0),
)
.map_err(|_| "未配置默认 Gitea 实例,请先在炼境 UI 中添加并设为默认".to_string())?;
let (base_url, token) = instance_auth(&conn, &instance_id)?;
// 仓库名 slug 化(保留 ASCII 字母数字 + - _ .);空字符串等同未传
let final_repo_name = repo_name
.filter(|s| !s.trim().is_empty())
.map(|s| s.to_string())
.unwrap_or_else(|| {
let slug: String = project_name
.to_ascii_lowercase()
.replace(' ', "-")
.chars()
.filter(|c| c.is_ascii_alphanumeric() || *c == '-' || *c == '_' || *c == '.')
.collect();
if slug.is_empty() {
format!("project-{}", &project_id[..8.min(project_id.len())])
} else {
slug
}
});
// 仓库拥有者(从 token 获取登录名)
let owner = gitea_api_get_user(&base_url, &token)?.login;
// 创建仓库,409=已存在时直接绑定
let (clone_url, already_existed) = match ureq::post(&format!("{base_url}/api/v1/user/repos"))
.set("Authorization", &format!("token {token}"))
.send_json(serde_json::json!({
"name": final_repo_name,
"description": format!("{} — 由炼境自动创建", project_name),
"private": private,
"auto_init": false,
})) {
Ok(resp) => {
let json: serde_json::Value =
resp.into_json().map_err(|e| format!("解析响应失败: {e}"))?;
let url = json["clone_url"]
.as_str()
.ok_or("响应缺少 clone_url")?
.to_string();
(url, false)
}
Err(ureq::Error::Status(409, _)) => {
let json = gitea_api_get_repo(&base_url, &token, &owner, &final_repo_name)?;
let url = json["clone_url"]
.as_str()
.ok_or("响应缺少 clone_url")?
.to_string();
(url, true)
}
Err(e) => return Err(api_err("创建仓库失败", e, &token)),
};
// 写入炼境 DB 绑定关系
upsert_repo_link(
&conn,
&instance_id,
project_id,
&owner,
&final_repo_name,
Some(&clone_url),
)?;
// 配置本地 git remote origin
let remote_status = configure_git_remote(win_path.as_deref(), &clone_url);
let existed_tag = if already_existed {
"已存在,直接绑定"
} else {
"新建"
};
let remote_tag = match &remote_status {
Ok(action) => format!("✅ {action}"),
Err(e) => format!("⚠️ {e}"),
};
let mut out = format!(
"Gitea 仓库已就绪:{base_url}/{owner}/{final_repo_name}\n\n\
- 仓库:{owner}/{final_repo_name}({existed_tag})\n\
- 炼境绑定:✅ project_id 已关联\n\
- git remote origin:{remote_tag}\n"
);
if remote_status.is_ok() {
out.push_str(
"\n建议下一步:git push -u origin main,然后调用 setup_webhook 注册 webhook。",
);
}
Ok(out)
}
/// 配置本地仓库的 git remote origin(add 或 set-url,幂等)
fn configure_git_remote(win_path: Option<&str>, clone_url: &str) -> Result<String, String> {
let path = win_path.ok_or_else(|| "项目无本地路径".to_string())?;
if !std::path::Path::new(path).join(".git").exists() {
return Err("本地目录无 .git,跳过 remote 配置".to_string());
}
let has_origin = std::process::Command::new("git")
.args(["-C", path, "remote", "get-url", "origin"])
.output()
.map(|o| o.status.success())
.unwrap_or(false);
let out = if has_origin {
std::process::Command::new("git")
.args(["-C", path, "remote", "set-url", "origin", clone_url])
.output()
} else {
std::process::Command::new("git")
.args(["-C", path, "remote", "add", "origin", clone_url])
.output()
}
.map_err(|e| format!("执行 git remote 失败: {e}"))?;
if out.status.success() {
Ok(if has_origin {
"origin set-url 更新成功".to_string()
} else {
"origin 已添加".to_string()
})
} else {
Err(String::from_utf8_lossy(&out.stderr).trim().to_string())
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::credential_store::CredentialStore;
#[test]
fn normalize_url_strips_trailing_slash() {
assert_eq!(
normalize_url("https://gitea.example.com/"),
"https://gitea.example.com"
);
assert_eq!(
normalize_url("https://gitea.example.com"),
"https://gitea.example.com"
);
assert_eq!(
normalize_url("https://gitea.example.com///"),
"https://gitea.example.com"
);
}
#[test]
fn gitea_tables_exist_after_migrate() {
let conn = crate::db::conn_with_schema();
let count: i64 = conn.query_row(
"SELECT COUNT(*) FROM sqlite_master WHERE type='table' AND name IN ('gitea_instances','gitea_repos')",
[],
|r| r.get(0),
).unwrap();
assert_eq!(count, 2, "gitea_instances 或 gitea_repos 表未建");
}
#[test]
fn save_and_list_instance() {
let conn = crate::db::conn_with_schema();
conn.execute(
"INSERT INTO gitea_instances (id, name, url, access_token, is_default)
VALUES ('inst-1', '测试实例', 'https://gitea.test', 'token123', 0)",
[],
)
.unwrap();
let count: i64 = conn
.query_row(
"SELECT COUNT(*) FROM gitea_instances WHERE id = 'inst-1'",
[],
|r| r.get(0),
)
.unwrap();
assert_eq!(count, 1);
}
#[test]
fn new_gitea_instance_stores_secret_before_database_reference_and_rolls_back_on_db_failure() {
let conn = crate::db::conn_with_schema();
let store = crate::credential_store::MemoryCredentialStore::new();
let instance = save_gitea_instance_with_store(
&conn,
&store,
SaveGiteaInstanceInput {
name: "test".into(),
url: "https://gitea.test/".into(),
access_token: "synthetic-gitea-token".into(),
is_default: Some(true),
},
)
.unwrap();
assert!(serde_json::to_string(&instance)
.unwrap()
.find("synthetic-gitea-token")
.is_none());
let (legacy, reference): (String, String) = conn
.query_row(
"SELECT access_token, access_token_ref FROM gitea_instances WHERE id = ?1",
params![instance.id],
|row| Ok((row.get(0)?, row.get(1)?)),
)
.unwrap();
assert!(legacy.is_empty());
assert_eq!(
store.get(&reference).unwrap().as_deref(),
Some(b"synthetic-gitea-token".as_slice())
);
conn.execute_batch("CREATE TRIGGER fail_gitea_insert BEFORE INSERT ON gitea_instances BEGIN SELECT RAISE(ABORT, 'injected database error'); END;").unwrap();
let failed_store = crate::credential_store::MemoryCredentialStore::new();
let result = save_gitea_instance_with_store(
&conn,
&failed_store,
SaveGiteaInstanceInput {
name: "failure".into(),
url: "https://gitea.test".into(),
access_token: "synthetic-failed-token".into(),
is_default: None,
},
);
assert!(result.is_err());
assert!(!failed_store.contains_value(b"synthetic-failed-token"));
let default_count: i64 = conn
.query_row(
"SELECT COUNT(*) FROM gitea_instances WHERE is_default = 1",
[],
|row| row.get(0),
)
.unwrap();
assert_eq!(default_count, 1, "失败的保存不应改变已有默认实例");
}
#[test]
fn deleting_instance_keeps_instance_and_repo_when_protected_store_delete_fails() {
let conn = crate::db::conn_with_schema();
let store = crate::credential_store::MemoryCredentialStore::failing_deletes();
store
.put("gitea/access-token/i/v1", b"synthetic-token")
.unwrap();
conn.execute(
"INSERT INTO gitea_instances (id, name, url, access_token, access_token_ref) VALUES ('i', 'test', 'https://example.invalid', '', 'gitea/access-token/i/v1')",
[],
).unwrap();
conn.execute(
"INSERT INTO gitea_repos (id, instance_id, project_id, gitea_owner, gitea_repo_name) VALUES ('r', 'i', 'p', 'owner', 'repo')",
[],
).unwrap();
assert!(delete_gitea_instance_with_store(&conn, &store, "i")
.unwrap_err()
.contains("needs_repair"));
let count: i64 = conn
.query_row("SELECT COUNT(*) FROM gitea_repos WHERE id='r'", [], |row| {
row.get(0)
})
.unwrap();
assert_eq!(count, 1);
assert_eq!(
store.get("gitea/access-token/i/v1").unwrap().as_deref(),
Some(b"synthetic-token".as_slice())
);
}
#[test]
fn deleting_instance_restores_token_when_later_webhook_delete_fails() {
let conn = crate::db::conn_with_schema();
let store = crate::credential_store::MemoryCredentialStore::failing_delete_for(
"gitea/webhook-secret/p/v1",
);
store
.put("gitea/access-token/i/v1", b"synthetic-token")
.unwrap();
store
.put("gitea/webhook-secret/p/v1", b"synthetic-hook")
.unwrap();
conn.execute("INSERT INTO gitea_instances (id, name, url, access_token, access_token_ref) VALUES ('i', 'test', 'https://example.invalid', '', 'gitea/access-token/i/v1')", []).unwrap();
conn.execute("INSERT INTO gitea_repos (id, instance_id, project_id, gitea_owner, gitea_repo_name, webhook_secret_ref) VALUES ('r', 'i', 'p', 'owner', 'repo', 'gitea/webhook-secret/p/v1')", []).unwrap();
assert!(delete_gitea_instance_with_store(&conn, &store, "i")
.unwrap_err()
.contains("needs_repair"));
let count: i64 = conn
.query_row(
"SELECT COUNT(*) FROM gitea_instances WHERE id='i'",
[],
|row| row.get(0),
)
.unwrap();
assert_eq!(count, 1);
assert_eq!(
store.get("gitea/access-token/i/v1").unwrap().as_deref(),
Some(b"synthetic-token".as_slice())
);
assert_eq!(
store.get("gitea/webhook-secret/p/v1").unwrap().as_deref(),
Some(b"synthetic-hook".as_slice())
);
}
#[test]
fn gitea_auth_migrates_legacy_token_and_update_preserves_old_on_store_failure() {
let conn = crate::db::conn_with_schema();
conn.execute(
"INSERT INTO gitea_instances (id, name, url, access_token) VALUES ('legacy', 'test', 'https://example.invalid', 'synthetic-old-token')",
[],
).unwrap();
let failing = crate::credential_store::MemoryCredentialStore::failing();
let error = instance_auth_with_store(&conn, &failing, "legacy").unwrap_err();
assert!(error.contains("needs_repair"));
assert!(!error.contains("synthetic-old-token"));
let old: String = conn
.query_row(
"SELECT access_token FROM gitea_instances WHERE id='legacy'",
[],
|r| r.get(0),
)
.unwrap();
assert_eq!(old, "synthetic-old-token");
let store = crate::credential_store::MemoryCredentialStore::new();
assert_eq!(
instance_auth_with_store(&conn, &store, "legacy").unwrap().1,
"synthetic-old-token"
);
let (old, reference): (String, String) = conn
.query_row(
"SELECT access_token, access_token_ref FROM gitea_instances WHERE id='legacy'",
[],
|r| Ok((r.get(0)?, r.get(1)?)),
)
.unwrap();
assert!(old.is_empty());
assert_eq!(
store.get(&reference).unwrap().as_deref(),
Some(b"synthetic-old-token".as_slice())
);
let result = update_gitea_instance_with_store(
&conn,
&failing,
"legacy",
Some("changed".into()),
None,
Some("synthetic-new-token".into()),
None,
);
assert!(result.is_err());
let (name, still_reference): (String, String) = conn
.query_row(
"SELECT name, access_token_ref FROM gitea_instances WHERE id='legacy'",
[],
|r| Ok((r.get(0)?, r.get(1)?)),
)
.unwrap();
assert_eq!(name, "test");
assert_eq!(still_reference, reference);
let result = update_gitea_instance_with_store(
&conn,
&store,
"legacy",
Some("changed".into()),
None,
Some("synthetic-new-token".into()),
None,
)
.unwrap();
assert_eq!(result.name, "changed");
assert_eq!(
instance_auth_with_store(&conn, &store, "legacy").unwrap().1,
"synthetic-new-token"
);
assert!(store.get(&reference).unwrap().is_none());
}
#[test]
fn create_pull_request_uses_stored_token_without_returning_it() {
use std::io::{BufRead, BufReader, Read, Write};
let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
let address = listener.local_addr().unwrap();
let server = std::thread::spawn(move || {
let (mut stream, _) = listener.accept().unwrap();
stream
.set_read_timeout(Some(std::time::Duration::from_secs(3)))
.unwrap();
let mut reader = BufReader::new(stream.try_clone().unwrap());
let mut line = String::new();
reader.read_line(&mut line).unwrap();
assert!(line.starts_with("POST /api/v1/repos/owner/repo/pulls "));
let mut authorized = false;
let mut content_length = 0;
loop {
line.clear();
reader.read_line(&mut line).unwrap();
if line == "\r\n" || line.is_empty() {
break;
}
if line.eq_ignore_ascii_case("authorization: token synthetic-pr-token\r\n") {
authorized = true;
}
if let Some(value) = line.to_ascii_lowercase().strip_prefix("content-length:") {
content_length = value.trim().parse().unwrap();
}
}
assert!(authorized, "请求未使用已存储的 Gitea 凭证");
let mut request_body = vec![0; content_length];
reader.read_exact(&mut request_body).unwrap();
let request: serde_json::Value = serde_json::from_slice(&request_body).unwrap();
assert_eq!(request["head"], "feature");
let body = r#"{"number":42,"html_url":"https://example.invalid/pr/42","title":"test"}"#;
write!(stream, "HTTP/1.1 201 Created\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}", body.len(), body).unwrap();
});
let conn = crate::db::conn_with_schema();
let store = crate::credential_store::MemoryCredentialStore::new();
let instance = save_gitea_instance_with_store(
&conn,
&store,
SaveGiteaInstanceInput {
name: "test".into(),
url: format!("http://{address}"),
access_token: "synthetic-pr-token".into(),
is_default: None,
},
)
.unwrap();
conn.execute("INSERT INTO gitea_repos (id, instance_id, project_id, gitea_owner, gitea_repo_name) VALUES ('r', ?1, 'p', 'owner', 'repo')", params![instance.id]).unwrap();
let result =
create_pull_request_with_store(&conn, &store, "p", "feature", "main", "test", "body")
.unwrap();
server.join().unwrap();
assert!(result.contains("\"pr_number\":42"));
assert!(!result.contains("synthetic-pr-token"));
}
#[test]
fn gitea_api_error_redacts_token_echoed_by_remote() {
use std::io::{BufRead, BufReader, Write};
let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
let address = listener.local_addr().unwrap();
let server = std::thread::spawn(move || {
let (mut stream, _) = listener.accept().unwrap();
let mut reader = BufReader::new(stream.try_clone().unwrap());
let mut line = String::new();
loop {
line.clear();
reader.read_line(&mut line).unwrap();
if line == "\r\n" || line.is_empty() {
break;
}
}
let body = r#"{"message":"bad synthetic-error-token"}"#;
write!(stream, "HTTP/1.1 401 Unauthorized\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}", body.len(), body).unwrap();
});
let error = ureq::get(&format!("http://{address}/test"))
.call()
.unwrap_err();
let result = api_err("test", error, "synthetic-error-token");
server.join().unwrap();
assert!(result.contains("[redacted]"));
assert!(!result.contains("synthetic-error-token"));
}
#[test]
fn webhook_reference_switch_keeps_legacy_value_on_database_failure() {
let conn = crate::db::conn_with_schema();
conn.execute("INSERT INTO gitea_instances (id, name, url, access_token) VALUES ('i', 'test', 'https://example.invalid', 'synthetic-token')", []).unwrap();
conn.execute("INSERT INTO gitea_repos (id, instance_id, project_id, gitea_owner, gitea_repo_name, webhook_secret, webhook_id) VALUES ('r', 'i', 'p', 'owner', 'repo', 'synthetic-old-secret', 1)", []).unwrap();
let store = crate::credential_store::MemoryCredentialStore::new();
let reference = crate::credential_store::versioned_reference(
crate::credential_store::LegacyCredentialKind::GiteaWebhookSecret,
"p",
);
crate::credential_store::put_verified(&store, &reference, b"synthetic-new-secret").unwrap();
conn.execute_batch("CREATE TRIGGER fail_webhook_switch BEFORE UPDATE OF webhook_secret_ref ON gitea_repos BEGIN SELECT RAISE(ABORT, 'injected failure'); END;").unwrap();
assert!(persist_webhook_secret(&conn, &store, "p", &reference, 2)
.unwrap_err()
.contains("needs_repair"));
let (legacy, hook_id): (String, i64) = conn
.query_row(
"SELECT webhook_secret, webhook_id FROM gitea_repos WHERE project_id='p'",
[],
|r| Ok((r.get(0)?, r.get(1)?)),
)
.unwrap();
assert_eq!(legacy, "synthetic-old-secret");
assert_eq!(hook_id, 1);
conn.execute_batch("DROP TRIGGER fail_webhook_switch;")
.unwrap();
persist_webhook_secret(&conn, &store, "p", &reference, 2).unwrap();
let (legacy, saved_reference, hook_id): (Option<String>, String, i64) = conn.query_row("SELECT webhook_secret, webhook_secret_ref, webhook_id FROM gitea_repos WHERE project_id='p'", [], |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?))).unwrap();
assert!(legacy.is_none());
assert_eq!(saved_reference, reference);
assert_eq!(hook_id, 2);
}
#[test]
fn webhook_setup_registers_mock_remote_and_stores_signature_secret() {
use std::io::{BufRead, BufReader, Read, Write};
let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
let address = listener.local_addr().unwrap();
let (sender, receiver) = std::sync::mpsc::channel();
let server = std::thread::spawn(move || {
let (mut stream, _) = listener.accept().unwrap();
stream
.set_read_timeout(Some(std::time::Duration::from_secs(3)))
.unwrap();
let mut reader = BufReader::new(stream.try_clone().unwrap());
let mut line = String::new();
reader.read_line(&mut line).unwrap();
assert!(line.starts_with("POST /api/v1/repos/owner/repo/hooks "));
let mut content_length = 0;
let mut authorized = false;
loop {
line.clear();
reader.read_line(&mut line).unwrap();
if line == "\r\n" || line.is_empty() {
break;
}
if let Some(value) = line.to_ascii_lowercase().strip_prefix("content-length:") {
content_length = value.trim().parse().unwrap();
}
if line.eq_ignore_ascii_case("authorization: token synthetic-hook-token\r\n") {
authorized = true;
}
}
assert!(authorized);
let mut body = vec![0; content_length];
reader.read_exact(&mut body).unwrap();
let request: serde_json::Value = serde_json::from_slice(&body).unwrap();
assert_eq!(
request["config"]["url"],
"http://127.0.0.1:12345/webhook/gitea/p"
);
sender
.send(request["config"]["secret"].as_str().unwrap().to_string())
.unwrap();
let response = r#"{"id":9}"#;
write!(stream, "HTTP/1.1 201 Created\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}", response.len(), response).unwrap();
});
let conn = crate::db::conn_with_schema();
let store = crate::credential_store::MemoryCredentialStore::new();
let instance = save_gitea_instance_with_store(
&conn,
&store,
SaveGiteaInstanceInput {
name: "test".into(),
url: format!("http://{address}"),
access_token: "synthetic-hook-token".into(),
is_default: None,
},
)
.unwrap();
conn.execute("INSERT INTO gitea_repos (id, instance_id, project_id, gitea_owner, gitea_repo_name) VALUES ('r', ?1, 'p', 'owner', 'repo')", params![instance.id]).unwrap();
let repo = setup_webhook_with_store(&conn, &store, "p", "http://127.0.0.1:12345").unwrap();
server.join().unwrap();
assert_eq!(repo.webhook_id, Some(9));
let sent_secret = receiver.recv().unwrap();
let (legacy, reference): (Option<String>, String) = conn
.query_row(
"SELECT webhook_secret, webhook_secret_ref FROM gitea_repos WHERE project_id='p'",
[],
|row| Ok((row.get(0)?, row.get(1)?)),
)
.unwrap();
assert!(legacy.is_none());
assert_eq!(
store.get(&reference).unwrap().as_deref(),
Some(sent_secret.as_bytes())
);
assert!(!serde_json::to_string(&repo).unwrap().contains(&sent_secret));
}
#[test]
fn project_resource_status_reports_only_scoped_non_secret_readiness() {
let conn = crate::db::conn_with_schema();
conn.execute(
"INSERT INTO project_profiles (id, name) VALUES ('profile', 'test')",
[],
)
.unwrap();
conn.execute(
"INSERT INTO project_workspaces (id, profile_id) VALUES ('p1', 'profile')",
[],
)
.unwrap();
let store = crate::credential_store::MemoryCredentialStore::new();
let empty = project_resource_status_for_conn(&conn, &store, "p1").unwrap();
assert_eq!(empty.gitea_pr.state, "not_configured");
assert!(project_resource_status_for_conn(&conn, &store, "p2").is_err());
conn.execute("INSERT INTO gitea_instances (id, name, url, access_token) VALUES ('i', 'test', 'https://secret-host.invalid', 'synthetic-token')", []).unwrap();
conn.execute("INSERT INTO gitea_repos (id, instance_id, project_id, gitea_owner, gitea_repo_name, webhook_secret) VALUES ('r', 'i', 'p1', 'sensitive-owner', 'sensitive-repo', 'synthetic-webhook-secret')", []).unwrap();
let legacy = project_resource_status_for_conn(&conn, &store, "p1").unwrap();
assert_eq!(legacy.gitea_pr.state, "pending_migration");
assert_eq!(legacy.gitea_pr.source, "legacy_pending_migration");
let unavailable = project_resource_status_for_conn(
&conn,
&crate::credential_store::MemoryCredentialStore::failing(),
"p1",
)
.unwrap();
assert_eq!(unavailable.gitea_pr.state, "unavailable");
let reference = crate::credential_store::versioned_reference(
crate::credential_store::LegacyCredentialKind::GiteaAccessToken,
"i",
);
conn.execute(
"UPDATE gitea_instances SET access_token='', access_token_ref=?1 WHERE id='i'",
params![reference],
)
.unwrap();
let broken = project_resource_status_for_conn(&conn, &store, "p1").unwrap();
assert_eq!(broken.gitea_pr.state, "needs_repair");
let json = serde_json::to_string(&broken).unwrap();
for sensitive in [
"synthetic-token",
"synthetic-webhook-secret",
"secret-host",
"sensitive-owner",
"sensitive-repo",
&reference,
] {
assert!(!json.contains(sensitive));
}
}
#[test]
fn upsert_repo_link_resets_webhook_on_rebind() {
let conn = crate::db::conn_with_schema();
conn.execute(
"INSERT INTO gitea_instances (id, name, url, access_token) VALUES ('i1', 'a', 'u', 't')",
[],
).unwrap();
// 首次绑定
let r1 = upsert_repo_link(
&conn,
"i1",
"proj-1",
"alice",
"repo-a",
Some("https://g/a.git"),
)
.unwrap();
assert_eq!(r1.gitea_repo_name, "repo-a");
// 模拟已设置 webhook
conn.execute(
"UPDATE gitea_repos SET webhook_secret = 'sec', webhook_id = 42 WHERE project_id = 'proj-1'",
[],
).unwrap();
// 换绑到另一个仓库 → webhook 字段应被重置
let r2 = upsert_repo_link(&conn, "i1", "proj-1", "alice", "repo-b", None).unwrap();
assert_eq!(r2.gitea_repo_name, "repo-b");
assert!(r2.webhook_id.is_none(), "换绑后 webhook_id 未重置");
let (secret, reference): (Option<String>, Option<String>) = conn.query_row(
"SELECT webhook_secret, webhook_secret_ref FROM gitea_repos WHERE project_id = 'proj-1'",
[],
|row| Ok((row.get(0)?, row.get(1)?)),
).unwrap();
assert!(
secret.is_none() && reference.is_none(),
"换绑后 webhook 凭证引用未重置"
);
// 仍然只有一行
let count: i64 = conn
.query_row(
"SELECT COUNT(*) FROM gitea_repos WHERE project_id = 'proj-1'",
[],
|r| r.get(0),
)
.unwrap();
assert_eq!(count, 1);
}
#[test]
fn unique_index_on_project_id() {
let conn = crate::db::conn_with_schema();
conn.execute(
"INSERT INTO gitea_instances (id, name, url, access_token) VALUES ('i1', 'a', 'u', 't')",
[],
).unwrap();
conn.execute(
"INSERT INTO gitea_repos (id, instance_id, project_id, gitea_owner, gitea_repo_name)
VALUES ('r1', 'i1', 'proj-uuid-1', 'owner', 'repo')",
[],
)
.unwrap();
// 同一 project_id 第二次应失败
let result = conn.execute(
"INSERT INTO gitea_repos (id, instance_id, project_id, gitea_owner, gitea_repo_name)
VALUES ('r2', 'i1', 'proj-uuid-1', 'owner', 'repo2')",
[],
);
assert!(result.is_err(), "project_id 唯一索引未生效");
}
// ── G6:轮询解析 ──────────────────────────────────────────────────────
#[test]
fn parse_gitea_commits_extracts_fields() {
let json = serde_json::json!([
{"sha": "abc", "commit": {"message": "feat(x): a", "committer": {"date": "2026-07-02T10:00:00+08:00"}}},
{"sha": "def", "commit": {"message": "fix: b", "committer": {}}, "created": "2026-07-01T09:00:00+08:00"}
]);
let commits = parse_gitea_commits(&json);
assert_eq!(commits.len(), 2);
assert_eq!(commits[0].0, "abc");
assert_eq!(commits[0].2.as_deref(), Some("2026-07-02T10:00:00+08:00"));
assert_eq!(
commits[1].2.as_deref(),
Some("2026-07-01T09:00:00+08:00"),
"committer.date 缺失时回退 created"
);
}
#[test]
fn parse_gitea_commits_tolerates_garbage() {
assert!(parse_gitea_commits(&serde_json::json!({"not": "array"})).is_empty());
assert!(parse_gitea_commits(&serde_json::json!([{"no_sha": true}])).is_empty());
}
#[test]
fn delivery_ci_summary_uses_latest_context_and_preserves_failure() {
let statuses = serde_json::json!([
{
"context": "test",
"status": "failure",
"description": "old failure",
"target_url": "https://gitea.test/actions/runs/524/jobs/0",
"created_at": "2026-09-22T01:00:00Z",
"updated_at": "2026-09-22T01:01:00Z"
},
{
"context": "test",
"status": "success",
"description": "green",
"target_url": "https://gitea.test/actions/runs/525/jobs/0",
"created_at": "2026-09-22T02:00:00Z",
"updated_at": "2026-09-22T02:01:00Z"
},
{"context": "lint", "state": "failure", "description": "bad"}
]);
let summary = summarize_ci_statuses(&statuses, Some("530888eb"));
assert_eq!(summary.state, "failed");
assert_eq!(summary.data["target_sha"], "530888eb");
assert_eq!(summary.data["contexts"].as_array().unwrap().len(), 2);
let test = summary.data["contexts"]
.as_array()
.unwrap()
.iter()
.find(|item| item["context"] == "test")
.unwrap();
assert_eq!(test["state"], "success");
assert_eq!(
test["target_url"],
"https://gitea.test/actions/runs/525/jobs/0"
);
assert_eq!(test["updated_at"], "2026-09-22T02:01:00Z");
}
#[test]
fn delivery_context_summary_uses_exact_context_and_preserves_evidence() {
// 字段形状来自 Gitea 1.22 CommitStatus schema;context/job 名与 target URL
// 来自 enterprise-system 的真实 run #525 页面,敏感 creator 信息不进入 fixture。
let statuses = serde_json::json!([
{
"context": "CI — DeepSeek Agent / agent (push)",
"status": "success",
"description": "success",
"target_url": "https://git.yljt.info/lanrtop/enterprise-system/actions/runs/525/jobs/0",
"created_at": "2026-09-22T03:35:00Z",
"updated_at": "2026-09-22T03:36:36Z"
},
{
"context": "CI — DeepSeek Agent / deploy-api (push)",
"status": "success",
"description": "success",
"target_url": "https://git.yljt.info/lanrtop/enterprise-system/actions/runs/525/jobs/2",
"created_at": "2026-09-22T03:35:00Z",
"updated_at": "2026-09-22T03:38:55Z"
}
]);
let signal = summarize_status_context(
&statuses,
"530888eb90983f21be4efa8191aae5711e03e2dc",
"CI — DeepSeek Agent / deploy-api (push)",
);
assert_eq!(signal.state, "success");
assert_eq!(
signal.data["target_sha"],
"530888eb90983f21be4efa8191aae5711e03e2dc"
);
assert_eq!(
signal.data["status"]["context"],
"CI — DeepSeek Agent / deploy-api (push)"
);
assert_eq!(
signal.data["status"]["target_url"],
"https://git.yljt.info/lanrtop/enterprise-system/actions/runs/525/jobs/2"
);
assert_eq!(signal.data["status"]["updated_at"], "2026-09-22T03:38:55Z");
}
#[test]
fn delivery_context_summary_normalizes_terminal_and_missing_states() {
for (raw_state, expected) in [
("pending", "pending"),
("failure", "failure"),
("error", "failure"),
("warning", "warning"),
("skipped", "skipped"),
] {
let statuses = serde_json::json!([
{"context": "deploy-api", "state": raw_state, "description": raw_state},
{"context": "agent", "status": "success"}
]);
let signal = summarize_status_context(&statuses, "abc", "deploy-api");
assert_eq!(signal.state, expected, "raw state={raw_state}");
}
let statuses = serde_json::json!([
{"context": "agent", "status": "success"}
]);
let missing = summarize_status_context(&statuses, "abc", "deploy-api");
assert_eq!(missing.state, "unknown");
assert_eq!(
missing.data["available_contexts"],
serde_json::json!(["agent"])
);
}
#[test]
fn production_context_uses_remote_default_branch_and_exact_job_name() {
use std::io::{BufRead, BufReader, Write};
let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
listener.set_nonblocking(true).unwrap();
let address = listener.local_addr().unwrap();
let sha = "a".repeat(40);
let status_body = serde_json::json!([{
"context": "CI — DeepSeek Agent / deploy-api (push)",
"status": "success",
"description": "Success",
"updated_at": "2026-09-22T03:38:55Z"
}])
.to_string();
let responses = [
(
"/api/v1/repos/o/r",
r#"{"default_branch":"main"}"#.to_string(),
),
(
"/api/v1/repos/o/r/branches/main",
format!(r#"{{"commit":{{"id":"{sha}"}}}}"#),
),
("/api/v1/repos/o/r/statuses/", status_body),
];
let server = std::thread::spawn(move || {
let started = std::time::Instant::now();
for (expected_path, body) in responses {
let mut stream = loop {
if let Ok((stream, _)) = listener.accept() {
break stream;
}
assert!(started.elapsed() < std::time::Duration::from_secs(5));
std::thread::sleep(std::time::Duration::from_millis(10));
};
// Windows 上从 nonblocking listener 接受的 socket 也可能保持 nonblocking。
stream.set_nonblocking(false).unwrap();
stream
.set_read_timeout(Some(std::time::Duration::from_secs(3)))
.unwrap();
let mut request = String::new();
BufReader::new(&mut stream).read_line(&mut request).unwrap();
assert!(
request.starts_with(&format!("GET {expected_path}")),
"{request}"
);
let response = format!(
"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
stream.write_all(response.as_bytes()).unwrap();
}
});
let conn = crate::db::conn_with_schema();
conn.execute(
"INSERT INTO gitea_instances (id, name, url, access_token) VALUES ('i1','test',?1,'token')",
params![format!("http://{address}")],
)
.unwrap();
conn.execute(
"INSERT INTO gitea_repos (id, instance_id, project_id, gitea_owner, gitea_repo_name) VALUES ('r1','i1','p1','o','r')",
[],
)
.unwrap();
let context = "CI — DeepSeek Agent / deploy-api (push)".to_string();
let store = crate::credential_store::MemoryCredentialStore::new();
let (target, signals) =
get_production_context_signals_with_store(&conn, &store, "p1", &[context.clone()])
.unwrap();
assert_eq!(target, sha);
assert_eq!(signals[&context].state, "success");
server.join().unwrap();
}
#[test]
fn gitea_1_22_skipped_job_is_not_deployment_success() {
// 脱敏自 enterprise-system 真实 run #529:Gitea 1.22.6 把 skipped job
// 编码为 status=success,仅通过 description 表达未执行,target_url 也是相对路径。
let statuses = serde_json::json!([
{
"context": "CI — DeepSeek Agent / deploy-api (push)",
"created_at": "2026-09-22T15:12:03+08:00",
"description": "Has been skipped",
"status": "success",
"target_url": "/lanrtop/enterprise-system/actions/runs/529/jobs/2",
"updated_at": "2026-09-22T15:12:03+08:00"
}
]);
let signal = summarize_status_context(
&statuses,
"a2e85f50b093cbd6af58a09eaa9a927849723724",
"CI — DeepSeek Agent / deploy-api (push)",
);
assert_eq!(signal.state, "skipped");
assert_eq!(signal.data["status"]["state"], "skipped");
assert_eq!(
signal.data["status"]["target_url"],
"/lanrtop/enterprise-system/actions/runs/529/jobs/2"
);
let ci = summarize_ci_statuses(&statuses, Some("a2e85f50b093cbd6af58a09eaa9a927849723724"));
assert_eq!(ci.state, "not_applicable");
assert_eq!(ci.data["contexts"][0]["state"], "skipped");
}
#[test]
fn delivery_ci_summary_compares_rfc3339_timestamps_across_offsets() {
let statuses = serde_json::json!([
{
"context": "test",
"status": "failure",
"description": "older despite later local calendar date",
"updated_at": "2026-09-22T02:00:00+09:00"
},
{
"context": "test",
"status": "success",
"description": "newer in UTC",
"updated_at": "2026-09-21T18:00:00Z"
}
]);
let summary = summarize_ci_statuses(&statuses, Some("abc"));
assert_eq!(summary.state, "passed");
assert_eq!(summary.data["contexts"][0]["description"], "newer in UTC");
}
#[test]
fn delivery_ci_summary_does_not_infer_skipped_from_unrelated_description() {
let statuses = serde_json::json!([{
"context": "test",
"status": "success",
"description": "success; no checks skipped",
"updated_at": "2026-09-22T10:00:00Z"
}]);
let summary = summarize_ci_statuses(&statuses, Some("abc"));
assert_eq!(summary.state, "passed");
assert_eq!(summary.data["contexts"][0]["state"], "success");
}
#[test]
fn delivery_review_summary_prefers_latest_reviewer_decision() {
let reviews = serde_json::json!([
{"user": {"login": "alice"}, "state": "REQUEST_CHANGES", "submitted_at": "2026-09-21T10:00:00Z"},
{"user": {"login": "alice"}, "state": "APPROVED", "submitted_at": "2026-09-21T09:00:00Z"},
{"user": {"login": "bob"}, "state": "APPROVED", "submitted_at": "2026-09-21T10:01:00Z"}
]);
let summary = summarize_reviews(&reviews);
assert_eq!(summary.state, "changes_requested");
assert_eq!(summary.data["reviewers"].as_array().unwrap().len(), 2);
}
#[test]
fn delivery_pr_selection_matches_current_branch_only() {
let pulls = serde_json::json!([
{"number": 2, "head": {"ref": "feat/other"}},
{"number": 3, "head": {"ref": "feat/delivery"}}
]);
assert_eq!(
select_open_pull_request(&pulls, "feat/delivery").expect("应匹配当前分支")["number"]
.as_i64(),
Some(3)
);
assert!(select_open_pull_request(&pulls, "main").is_none());
}
fn merge_gate_fixture() -> (serde_json::Value, DeliverySignal, DeliverySignal) {
let sha = "0123456789abcdef0123456789abcdef01234567";
let pull_request = serde_json::json!({
"state": "open",
"merged": false,
"mergeable": true,
"head": { "sha": sha }
});
let ci = DeliverySignal::new(
"passed",
"gitea",
"所有最新 CI context 均成功",
serde_json::json!({ "target_sha": sha, "contexts": [] }),
);
let review = DeliverySignal::new(
"pending",
"gitea",
"尚无有效 approve/reject review",
serde_json::json!({ "reviewers": [] }),
);
(pull_request, ci, review)
}
#[test]
fn merge_gate_requires_explicit_confirmation_and_full_sha() {
let (pull_request, ci, review) = merge_gate_fixture();
let sha = pull_request["head"]["sha"].as_str().unwrap();
assert!(
validate_merge_preconditions(false, sha, &pull_request, &ci, &review)
.unwrap_err()
.contains("明确确认")
);
assert!(
validate_merge_preconditions(true, "01234567", &pull_request, &ci, &review)
.unwrap_err()
.contains("完整 40 位")
);
}
#[test]
fn merge_gate_rejects_sha_drift_and_non_open_pr() {
let (mut pull_request, ci, review) = merge_gate_fixture();
let different_sha = "fedcba9876543210fedcba9876543210fedcba98";
assert!(
validate_merge_preconditions(true, different_sha, &pull_request, &ci, &review,)
.unwrap_err()
.contains("SHA 已变化")
);
pull_request["state"] = serde_json::json!("closed");
let original_sha = "0123456789abcdef0123456789abcdef01234567";
assert!(
validate_merge_preconditions(true, original_sha, &pull_request, &ci, &review,)
.unwrap_err()
.contains("不可合并")
);
}
#[test]
fn merge_gate_rejects_unready_ci_and_blocking_review() {
let (pull_request, mut ci, mut review) = merge_gate_fixture();
let sha = pull_request["head"]["sha"].as_str().unwrap();
for state in ["pending", "failed", "unavailable"] {
ci.state = state.to_string();
assert!(
validate_merge_preconditions(true, sha, &pull_request, &ci, &review)
.unwrap_err()
.contains("CI 门禁未通过")
);
}
ci.state = "passed".to_string();
review.state = "changes_requested".to_string();
assert!(
validate_merge_preconditions(true, sha, &pull_request, &ci, &review)
.unwrap_err()
.contains("请求修改")
);
review.state = "unavailable".to_string();
assert!(
validate_merge_preconditions(true, sha, &pull_request, &ci, &review)
.unwrap_err()
.contains("不可用或未知")
);
}
#[test]
fn merge_gate_allows_current_sha_with_passed_ci_and_no_blocking_review() {
let (pull_request, ci, mut review) = merge_gate_fixture();
let sha = pull_request["head"]["sha"].as_str().unwrap();
assert_eq!(
validate_merge_preconditions(true, sha, &pull_request, &ci, &review).unwrap(),
sha
);
review.state = "approved".to_string();
assert_eq!(
validate_merge_preconditions(true, sha, &pull_request, &ci, &review).unwrap(),
sha
);
}
#[test]
fn merge_request_body_binds_the_confirmed_head_sha() {
let sha = "0123456789abcdef0123456789abcdef01234567";
let body = merge_request_body(13, sha);
assert_eq!(body["Do"], "merge");
assert_eq!(body["head_commit_id"], sha);
}
#[test]
fn public_gitea_dtos_never_serialize_credentials() {
let instance = GiteaInstance {
id: "i1".into(),
name: "test".into(),
url: "https://g.example".into(),
is_default: true,
created_at: None,
};
let repo = GiteaRepo {
id: "r1".into(),
instance_id: "i1".into(),
project_id: "p1".into(),
gitea_owner: "owner".into(),
gitea_repo_name: "repo".into(),
clone_url: None,
webhook_id: Some(1),
linked_at: None,
};
let instance_json = serde_json::to_value(instance).unwrap();
let repo_json = serde_json::to_value(repo).unwrap();
assert!(instance_json.get("access_token").is_none());
assert!(repo_json.get("webhook_secret").is_none());
}
}