diff --git a/crates/utopia-server/src/api/sources_cleanup_tests.rs b/crates/utopia-server/src/api/sources_cleanup_tests.rs new file mode 100644 index 000000000..5a2e7f7bf --- /dev/null +++ b/crates/utopia-server/src/api/sources_cleanup_tests.rs @@ -0,0 +1,300 @@ +//! 墓碑清理之后,派生要跟上(#875)。`DELETE /documents/{id}` 删完一篇就按库的开关重推一遍 +//! (`settle_derivations`:删除、撤销、同步复活共用);`POST .../missing/cleanup` 一次删一批 +//! 墓碑文档,从前删完却不重推——凭这些文档成立的结论照旧挂着,直到下一轮定时推导。 +//! 同一份夹具两条路各走一遍,单篇删除是对照。 +//! +//! 连库的测试,没有 `UTOPIA_DATABASE_URL` 就跳过(同 documents_routes_tests)。 + +use axum::body::{to_bytes, Body}; +use axum::http::{Request, StatusCode}; +use serde_json::{json, Value}; +use std::sync::Arc; +use tower::ServiceExt; +use utopia_core::models::RelationAxioms; +use utopia_store::business_rules::ConditionInput; +use utopia_store::graph::Validity; +use uuid::Uuid; + +struct Fixture { + pool: sqlx::PgPool, + app: axum::Router, + org: Uuid, + kb: Uuid, + source: Uuid, + token: String, + cup: Uuid, + location: Uuid, + _dir: tempfile::TempDir, +} + +impl Fixture { + async fn new() -> anyhow::Result> { + let Some(url) = utopia_store::test_db::url() else { + return Ok(None); + }; + let pool = sqlx::PgPool::connect(&url).await?; + utopia_store::db::migrate(&pool).await?; + let (org, ws, kb, user, source) = ( + Uuid::now_v7(), + Uuid::now_v7(), + Uuid::now_v7(), + Uuid::now_v7(), + Uuid::now_v7(), + ); + // Only locally generated UUIDs are interpolated into fixture SQL. + // The base keeps the column default: materialized inference is on (0050). + sqlx::raw_sql(&format!( + "INSERT INTO organizations(id,name) VALUES ('{org}','cleanup-settle-test'); + INSERT INTO workspaces(id,org_id,name) VALUES ('{ws}','{org}','cleanup-settle-test'); + INSERT INTO users(id,org_id,email,display_name,password_hash) + VALUES ('{user}','{org}','{user}@cleanup.test','cleanup-test','unused'); + INSERT INTO knowledge_bases(id,workspace_id,name) VALUES ('{kb}','{ws}','observations'); + INSERT INTO kb_members(kb_id,user_id,role) VALUES ('{kb}','{user}','editor'); + INSERT INTO sources(id,kb_id,kind,name) VALUES ('{source}','{kb}','api','robot-1');" + )) + .execute(&pool) + .await?; + let cup = utopia_store::ontology::create_entity_type( + &pool, + kb, + "cup", + "Cup", + "#7fd0ff", + "circle", + &[], + "", + ) + .await?; + let ready = utopia_store::ontology::create_entity_type( + &pool, + kb, + "step_ready", + "Step precondition holds", + "#7fd0ff", + "circle", + &[], + "", + ) + .await?; + let location = utopia_store::ontology::create_relation_type( + &pool, + kb, + "location", + "location", + "state", + RelationAxioms { + functional: true, + ..Default::default() + }, + "", + "attribute", + &[cup], + &[], + Some("text"), + None, + ) + .await?; + utopia_store::business_rules::create( + &pool, + kb, + "pick cup from desk", + "", + cup, + "typing", + Some(ready), + None, + None, + None, + &[ConditionInput { + group: 0, + predicate_id: location, + op: "in".into(), + operand: Some(json!(["desk"])), + }], + ) + .await?; + let dir = tempfile::tempdir()?; + let cfg = utopia_core::config::AppConfig { + data_dir: dir.path().to_string_lossy().into_owned(), + ..Default::default() + }; + let search = Arc::new(utopia_search::SearchIndex::open( + &dir.path().join("search"), + )?); + let state = crate::state::AppState::new(pool.clone(), &cfg, search, "test-only".into()); + let token = crate::auth::issue_token(&state, user)?; + let app = super::super::router(state.clone(), &cfg); + Ok(Some(Self { + pool, + app, + org, + kb, + source, + token, + cup, + location, + _dir: dir, + })) + } + + /// 一样东西、一份挂在来源下的文档、一条以那份文档为唯一证据的读数「在桌上」 + async fn reading(&self, name: &str) -> anyhow::Result<(Uuid, Uuid, Uuid)> { + let (doc, chunk, thing) = (Uuid::now_v7(), Uuid::now_v7(), Uuid::now_v7()); + sqlx::query( + "INSERT INTO documents (id, kb_id, source_id, filename, sha256, external_key, status) + VALUES ($1, $2, $3, $4, $5, $6, 'ready')", + ) + .bind(doc) + .bind(self.kb) + .bind(self.source) + .bind(format!("{name}.json")) + .bind(format!("sha-{doc}")) + .bind(format!("api:{name}")) + .execute(&self.pool) + .await?; + sqlx::query( + "INSERT INTO chunks (id, kb_id, document_id, seq, text) VALUES ($1, $2, $3, 0, $4)", + ) + .bind(chunk) + .bind(self.kb) + .bind(doc) + .bind(format!("{name} is on desk")) + .execute(&self.pool) + .await?; + sqlx::query( + "INSERT INTO entities (id, kb_id, type_id, canonical_name) VALUES ($1, $2, $3, $4)", + ) + .bind(thing) + .bind(self.kb) + .bind(self.cup) + .bind(name) + .execute(&self.pool) + .await?; + let (fact, _) = utopia_store::graph::insert_value_fact( + &self.pool, + self.kb, + thing, + Some(self.location), + &json!({ "value": "desk" }), + Validity::default().attested(Some("2026-09-23T08:00:00Z".parse()?)), + 1.0, + ) + .await?; + utopia_store::graph::add_evidence(&self.pool, fact, chunk, None, Some("is on")).await?; + Ok((doc, thing, fact)) + } + + async fn call(&self, method: &str, uri: &str) -> anyhow::Result<(StatusCode, Value)> { + let response = self + .app + .clone() + .oneshot( + Request::builder() + .method(method) + .uri(uri) + .header("Authorization", format!("Bearer {}", self.token)) + .body(Body::empty())?, + ) + .await?; + let status = response.status(); + let bytes = to_bytes(response.into_body(), 1 << 20).await?; + Ok(( + status, + serde_json::from_slice(&bytes).unwrap_or(Value::Null), + )) + } + + /// 这样东西此刻还挂着规则的结论吗 + async fn concluded(&self, thing: Uuid) -> anyhow::Result { + let (n,): (i64,) = sqlx::query_as( + "SELECT count(*) FROM derived_facts WHERE subject_id = $1 AND invalidated_at IS NULL", + ) + .bind(thing) + .fetch_one(&self.pool) + .await?; + Ok(n > 0) + } + + async fn conclusion_ids(&self, thing: Uuid) -> anyhow::Result> { + Ok(sqlx::query_scalar( + "SELECT id FROM derived_facts WHERE subject_id = $1 AND invalidated_at IS NULL ORDER BY id", + ) + .bind(thing) + .fetch_all(&self.pool) + .await?) + } + + async fn cleanup(self) -> anyhow::Result<()> { + // 库先走:删除记录(document_deletions)随库级联,它记着删的人,人要留到最后 + sqlx::query("DELETE FROM knowledge_bases WHERE id = $1") + .bind(self.kb) + .execute(&self.pool) + .await?; + sqlx::query("DELETE FROM organizations WHERE id = $1") + .bind(self.org) + .execute(&self.pool) + .await?; + self.pool.close().await; + Ok(()) + } +} + +#[tokio::test] +async fn a_cleanup_retires_what_its_documents_concluded() -> anyhow::Result<()> { + let Some(f) = Fixture::new().await? else { + return Ok(()); + }; + let run = async { + let (deleted, by_delete, _) = f.reading("cup-1").await?; + let (_, by_cleanup, reading) = f.reading("cup-2").await?; + let (_, unrelated, _) = f.reading("cup-3").await?; + utopia_store::reasoning::materialize(&f.pool, f.kb).await?; + assert!(f.concluded(by_delete).await? && f.concluded(by_cleanup).await?); + let cleanup_before = f.conclusion_ids(by_cleanup).await?; + let unrelated_before = f.conclusion_ids(unrelated).await?; + assert_eq!(unrelated_before.len(), 1); + + // 对照:删一篇,删完即按开关重推 + let (status, body) = f + .call("DELETE", &format!("/api/v1/documents/{deleted}")) + .await?; + assert_eq!(status, StatusCode::OK, "{body}"); + assert!( + !f.concluded(by_delete).await?, + "a single delete settles the derivations" + ); + assert_eq!(f.conclusion_ids(by_cleanup).await?, cleanup_before); + assert_eq!(f.conclusion_ids(unrelated).await?, unrelated_before); + + // 墓碑 + 清理:同一件事按批做 + utopia_store::documents::mark_missing_keys(&f.pool, f.source, &["api:cup-2".into()]) + .await?; + let (status, body) = f + .call( + "POST", + &format!("/api/v1/kbs/{}/sources/{}/missing/cleanup", f.kb, f.source), + ) + .await?; + assert_eq!(status, StatusCode::OK, "{body}"); + assert_eq!(body["deleted"], json!(1)); + let (gone,): (bool,) = + sqlx::query_as("SELECT invalidated_at IS NOT NULL FROM facts WHERE id = $1") + .bind(reading) + .fetch_one(&f.pool) + .await?; + assert!(gone, "the reading's only evidence was deleted"); + assert!( + !f.concluded(by_cleanup).await?, + "the conclusion that reading supported must be retired as after a single delete" + ); + assert_eq!( + f.conclusion_ids(unrelated).await?, + unrelated_before, + "cleanup must preserve the pre-existing unrelated conclusion row" + ); + anyhow::Ok(()) + } + .await; + f.cleanup().await?; + run +} diff --git a/crates/utopia-server/src/api/sources_routes.rs b/crates/utopia-server/src/api/sources_routes.rs index 105600cc0..fd2725c58 100644 --- a/crates/utopia-server/src/api/sources_routes.rs +++ b/crates/utopia-server/src/api/sources_routes.rs @@ -280,10 +280,19 @@ pub async fn cleanup_missing( .map_err(|e| utopia_core::AppError::Other(e.into()))? .map_err(utopia_core::AppError::Other)?; } + // 与单篇删除同一个收尾:前提没了,派生当场跟上,而不是挂到下一轮定时推导(#875)。 + // 一批只推一遍 + if !ids.is_empty() { + super::documents_routes::settle_derivations(&state, kb_id).await?; + } state.emit_source(kb_id); Ok(Json(json!({ "deleted": ids.len() }))) } +#[cfg(test)] +#[path = "sources_cleanup_tests.rs"] +mod cleanup_tests; + pub async fn delete( State(state): State, AuthUser(user): AuthUser,