A client reads a document back by the external_id it pushed it under

crates/utopia-server/src/api/documents_routes.rs

@@ -108,19 +108,43 @@ pub async fn content(
     if document.purged_at.is_some() {
         return Ok(gone(PURGED_MESSAGE));
     }
+    let Some(retained) = read_retained(&state, id, query.version, |_| true).await? else {
+        return Ok(gone(PURGED_MESSAGE));
+    };
+    let headers = content_headers(&retained.document, &retained.version, retained.bytes.len())?;
+    Ok((StatusCode::OK, headers, retained.bytes).into_response())
+}
 
-    // Hold the row lock through the blob read. Replacement and purge otherwise
-    // can move or delete the selected blob after the ledger says we may serve it.
+/// One retained original with the ledger row and document row that named it.
+struct Retained {
+    document: Document,
+    version: utopia_store::documents::DocumentVersion,
+    bytes: Vec<u8>,
+}
+
+/// Read a retained original, `None` when the document has been purged or the
+/// locked row is not one `serves` accepts.
+///
+/// Hold the row lock through the blob read. Replacement and purge otherwise
+/// can move or delete the selected blob after the ledger says we may serve it.
+/// A caller's own conditions on the row go in `serves`, so they are judged on
+/// the row the lock holds rather than on one read before it.
+async fn read_retained(
+    state: &AppState,
+    id: Uuid,
+    requested: Option<i32>,
+    serves: impl FnOnce(&Document) -> bool,
+) -> ApiResult<Option<Retained>> {
     let mut tx = state.pool.begin().await?;
     let document: Document =
         sqlx::query_as("SELECT * FROM documents WHERE id = $1 FOR NO KEY UPDATE")
             .bind(id)
             .fetch_one(&mut *tx)
             .await?;
-    if document.purged_at.is_some() {
-        return Ok(gone(PURGED_MESSAGE));
+    if document.purged_at.is_some() || !serves(&document) {
+        return Ok(None);
     }
-    let version: utopia_store::documents::DocumentVersion = match query.version {
+    let version: utopia_store::documents::DocumentVersion = match requested {
         Some(requested) => sqlx::query_as(
             "SELECT version, sha256, size_bytes, ingested_at
                FROM document_versions WHERE document_id = $1 AND version = $2",
@@ -161,8 +185,11 @@ pub async fn content(
         )
         .into());
     }
-    let headers = content_headers(&document, &version, bytes.len())?;
-    Ok((StatusCode::OK, headers, bytes).into_response())
+    Ok(Some(Retained {
+        document,
+        version,
+        bytes,
+    }))
 }
 
 /// Name the exact retained versions the content route can address.
@@ -180,6 +207,64 @@ pub async fn versions(
     Ok((StatusCode::OK, Json(json!({ "versions": versions }))).into_response())
 }
 
+#[derive(Deserialize)]
+pub struct ExternalIdQuery {
+    pub external_id: String,
+}
+
+/// Read a pushed document back by the identity its source pushed it under (#1276).
+///
+/// The reader is the person, as on the content route; a source's push key is no
+/// credential here, so a leaked key stays push-only (0052). A document in the
+/// trash, a tombstoned identity and one never pushed are all 404, and purge
+/// erases the identity, so a purged one is 404 too.
+pub async fn by_external_id(
+    State(state): State<AppState>,
+    reader: SessionOrToken,
+    Path((kb_id, source_id)): Path<(Uuid, Uuid)>,
+    Query(query): Query<ExternalIdQuery>,
+) -> ApiResult<Response> {
+    require_reader_kb(&state, &reader, kb_id).await?;
+    let source = super::sources_routes::source_in_kb(&state, kb_id, source_id).await?;
+    if !super::sources_routes::is_push_kind(&source.kind) {
+        return Err(AppError::NotFound.into());
+    }
+    let external_id = query.external_id.trim();
+    let key = super::sources_routes::pushed_key(&source, external_id);
+    let found = utopia_store::documents::find_by_external_key(&state.pool, source.id, &key)
+        .await?
+        .ok_or(AppError::NotFound)?;
+    let live_under_key = |document: &Document| {
+        document.source_id == Some(source.id)
+            && document.external_key.as_deref() == Some(key.as_str())
+            && document.deleted_at.is_none()
+            && document.missing_since.is_none()
+    };
+    let Retained {
+        document,
+        version,
+        bytes,
+    } = read_retained(&state, found.id, None, live_under_key)
+        .await?
+        .ok_or(AppError::NotFound)?;
+    let versions = utopia_store::documents::versions(&state.pool, document.id).await?;
+    let body = json!({
+        "id": document.id,
+        "external_id": external_id,
+        "filename": document.filename,
+        "mime": document.mime,
+        "sha256": document.sha256,
+        "size_bytes": document.size_bytes,
+        "status": document.status,
+        "doc_time": document.doc_time,
+        "updated_at": document.updated_at,
+        "version": version.version,
+        "content": String::from_utf8_lossy(&bytes),
+        "versions": versions,
+    });
+    Ok((StatusCode::OK, Json(body)).into_response())
+}
+
 /// 批量上传（multipart，可多文件）。重复内容（同 KB 同 sha256）跳过。
 pub async fn upload(
     State(state): State<AppState>,

crates/utopia-server/src/api/mod.rs

@@ -474,6 +474,11 @@ pub fn router(state: AppState, cfg: &AppConfig) -> Router {
             "/kbs/{id}/sources/{source_id}/runs",
             get(sources_routes::runs),
         )
+        // 按推送时的 external_id 读回文档（#1276）：id 里可能有 `/`，所以走查询参数
+        .route(
+            "/kbs/{id}/sources/{source_id}/documents/by-external-id",
+            get(documents_routes::by_external_id),
+        )
         .route(
             "/kbs/{id}/sources/{source_id}/re-extract",
             post(sources_routes::re_extract),

crates/utopia-server/src/api/sources_routes.rs

@@ -47,7 +47,7 @@ fn push_key_matches(stored: Option<&str>, offered: &str) -> bool {
 /// `require_kb` 只查人对库的权限；来源 id 是另一个维度——不比对的话，A 库的
 /// Editor 拿着 B 库来源的 id 就能同步、清理、删除它。不属于就当不存在（404），
 /// 与 `get_token` 一直以来的做法一致
-async fn source_in_kb(
+pub(crate) async fn source_in_kb(
     state: &AppState,
     kb_id: Uuid,
     source_id: Uuid,
@@ -158,10 +158,17 @@ pub async fn create(
 
 /// 有推送密钥的来源：`api` 推文档，`statements` 推陈述（0054）。两种的密钥同一套
 /// 生成、查看、轮换
-fn is_push_kind(kind: &str) -> bool {
+pub(crate) fn is_push_kind(kind: &str) -> bool {
     matches!(kind, "api" | "statements")
 }
 
+/// The `external_key` a push source files a pushed identity under (`api:…`, `statements:…`).
+/// The push handlers write it and the read by `external_id` (#1276) looks it up, so the two
+/// cannot drift apart.
+pub(crate) fn pushed_key(source: &utopia_core::models::Source, identity: &str) -> String {
+    format!("{}:{identity}", source.kind)
+}
+
 /// 查看推送密钥（Editor；列表响应从不携带，查看走这里）。
 ///
 /// 令牌要 write：密钥能往库里推东西，读令牌拿到它就等于绕过了自己的 scope
@@ -496,7 +503,7 @@ async fn handle_push(
             "external_id or filename is required".into(),
         ));
     }
-    let key = format!("api:{identity}");
+    let key = pushed_key(source, &identity);
 
     // 墓碑：标记"不在来源中"（与 custom 的 deleted[] 同一条路径），content 可省略
     if body.deleted {
@@ -810,7 +817,7 @@ async fn handle_statements_push(
 ) -> Result<crate::ingest_sources::IngestAction, PushError> {
     let (body, content) = validate_statements_payload(bytes).map_err(PushError::Rejected)?;
     let identity = body.external_id.trim().to_string();
-    let key = format!("statements:{identity}");
+    let key = pushed_key(source, &identity);
     let Some(content) = content else {
         utopia_store::documents::mark_missing_keys(&state.pool, source.id, &[key])
             .await

crates/utopia-server/src/api/sources_statements_tests.rs

@@ -97,6 +97,7 @@ struct Fixture {
     app: axum::Router,
     org: Uuid,
     kb: Uuid,
+    user: Uuid,
     source: Uuid,
     api_source: Uuid,
     token: String,
@@ -157,6 +158,7 @@ impl Fixture {
             app,
             org,
             kb,
+            user,
             source,
             api_source,
             token,
@@ -234,6 +236,22 @@ impl Fixture {
         Ok((status, value))
     }
 
+    /// 编辑者的一枚读令牌；`kb_ids` 为 `None` 时不限库
+    async fn read_token(&self, kb_ids: Option<&[Uuid]>) -> anyhow::Result<String> {
+        let (_, token) =
+            utopia_store::tokens::issue(&self.pool, self.user, "ci", "read", kb_ids, None).await?;
+        Ok(token)
+    }
+
+    /// 按 external_id 读回 statements 来源的文档
+    async fn read_back(
+        &self,
+        token: &str,
+        external_id: &str,
+    ) -> anyhow::Result<(StatusCode, Value)> {
+        read_back(self.app.clone(), self.kb, self.source, token, external_id).await
+    }
+
     async fn cleanup(self) -> anyhow::Result<()> {
         sqlx::query("DELETE FROM organizations WHERE id = $1")
             .bind(self.org)
@@ -244,6 +262,32 @@ impl Fixture {
     }
 }
 
+/// 不借 `Fixture`，好在测试里另起一个任务去读
+async fn read_back(
+    app: axum::Router,
+    kb: Uuid,
+    source: Uuid,
+    token: &str,
+    external_id: &str,
+) -> anyhow::Result<(StatusCode, Value)> {
+    let path = format!(
+        "/api/v1/kbs/{kb}/sources/{source}/documents/by-external-id?external_id={external_id}"
+    );
+    let response = app
+        .oneshot(
+            Request::get(path)
+                .header("Authorization", format!("Bearer {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),
+    ))
+}
+
 fn observation(when: &str, place: &str) -> Value {
     json!({
         "external_id": "obs-000412",
@@ -736,3 +780,116 @@ async fn a_trashed_id_pushed_with_another_documents_content_stays_in_the_trash(
     assert!(a.deleted_at.is_some(), "a stays in the trash");
     f.cleanup().await
 }
+
+/// 推上去的身份按 external_id 读得回来（#1276）：读令牌拿到当前正文与两个版本；
+/// 限定到别的库的令牌和读不到的文档一样是 404
+#[tokio::test]
+async fn a_read_token_reads_an_updated_identity_back_and_one_for_another_base_gets_404(
+) -> anyhow::Result<()> {
+    let Some(f) = Fixture::new().await? else {
+        return Ok(());
+    };
+    for place in ["kitchen table", "counter"] {
+        let (status, body) = f
+            .push(f.source, &f.token, &observation("08:14:03", place))
+            .await?;
+        assert_eq!(status, StatusCode::OK, "{body}");
+    }
+    let token = f.read_token(Some(&[f.kb])).await?;
+    let (status, body) = f.read_back(&token, "obs-000412").await?;
+    assert_eq!(status, StatusCode::OK, "{body}");
+    let content: Value = serde_json::from_str(body["content"].as_str().unwrap_or_default())?;
+    assert_eq!(content["e"][1][0], "counter", "the current content");
+    let versions: Vec<&Value> = body["versions"]
+        .as_array()
+        .into_iter()
+        .flatten()
+        .map(|v| &v["version"])
+        .collect();
+    assert_eq!(
+        (&body["external_id"], &body["version"], versions),
+        (&json!("obs-000412"), &json!(2), vec![&json!(1), &json!(2)])
+    );
+    let elsewhere = f.read_token(Some(&[Uuid::now_v7()])).await?;
+    let (status, body) = f.read_back(&elsewhere, "obs-000412").await?;
+    assert_eq!(status, StatusCode::NOT_FOUND, "{body}");
+    f.cleanup().await
+}
+
+/// 墓碑标记过的身份和回收站里的文档都读不到，与没推过的身份一样是 404
+#[tokio::test]
+async fn a_tombstoned_identity_and_a_trashed_document_read_back_as_404() -> anyhow::Result<()> {
+    let Some(f) = Fixture::new().await? else {
+        return Ok(());
+    };
+    let mut trashed = observation("08:14:03", "counter");
+    trashed["external_id"] = json!("obs-000413");
+    for body in [
+        observation("08:14:03", "kitchen table"),
+        json!({ "external_id": "obs-000412", "deleted": true }),
+        trashed,
+    ] {
+        let (status, body) = f.push(f.source, &f.token, &body).await?;
+        assert_eq!(status, StatusCode::OK, "{body}");
+    }
+    let doc = documents::find_by_external_key(&f.pool, f.source, "statements:obs-000413")
+        .await?
+        .expect("obs-000413 is stored");
+    documents::delete(&f.pool, f.kb, doc.id, None).await?;
+
+    let token = f.read_token(None).await?;
+    for external_id in ["obs-000412", "obs-000413"] {
+        let (status, body) = f.read_back(&token, external_id).await?;
+        assert_eq!(status, StatusCode::NOT_FOUND, "{external_id}: {body}");
+    }
+    f.cleanup().await
+}
+
+/// 查到文档之后、读原件之前它被墓碑标记了：判断落在锁住的那一行上，所以答 404，
+/// 而不是把锁之前读到的那一行当真，照样把正文交出去
+#[tokio::test]
+async fn an_identity_tombstoned_while_the_read_waits_for_its_row_reads_back_as_404(
+) -> anyhow::Result<()> {
+    let Some(f) = Fixture::new().await? else {
+        return Ok(());
+    };
+    let (status, body) = f
+        .push(f.source, &f.token, &observation("08:14:03", "counter"))
+        .await?;
+    assert_eq!(status, StatusCode::OK, "{body}");
+    let doc = documents::find_by_external_key(&f.pool, f.source, "statements:obs-000412")
+        .await?
+        .expect("obs-000412 is stored");
+    let token = f.read_token(None).await?;
+
+    // 未提交的墓碑握着行锁：读路由的查找看到的还是活的那行，随后卡在锁上
+    let mut tombstone = f.pool.begin().await?;
+    let holder: i32 = sqlx::query_scalar("SELECT pg_backend_pid()")
+        .fetch_one(&mut *tombstone)
+        .await?;
+    sqlx::query("UPDATE documents SET missing_since = now() WHERE id = $1")
+        .bind(doc.id)
+        .execute(&mut *tombstone)
+        .await?;
+    let (app, kb, source) = (f.app.clone(), f.kb, f.source);
+    let read = tokio::spawn(async move { read_back(app, kb, source, &token, "obs-000412").await });
+    let mut waiting = false;
+    for _ in 0..500 {
+        waiting = sqlx::query_scalar(
+            "SELECT EXISTS (SELECT 1 FROM pg_stat_activity WHERE $1 = ANY(pg_blocking_pids(pid)))",
+        )
+        .bind(holder)
+        .fetch_one(&f.pool)
+        .await?;
+        if waiting {
+            break;
+        }
+        tokio::time::sleep(std::time::Duration::from_millis(10)).await;
+    }
+    assert!(waiting, "the read waits on the tombstone's row lock");
+    tombstone.commit().await?;
+
+    let (status, body) = read.await??;
+    assert_eq!(status, StatusCode::NOT_FOUND, "{body}");
+    f.cleanup().await
+}

crates/utopia-server/src/auth.rs

@@ -179,6 +179,8 @@ impl FromRequestParts<AppState> for AuthUser {
 /// 用这个提取器的只有文档原件（0052）、「把一个库搭起来」那几条和读图那几条：搜实体、
 /// 实体详情、认知历史、事实证据、图总览、邻域（决定 7）。
 ///
+/// 按推送时的 `external_id` 读回一篇文档（#1276）也是文档原件，开给令牌的理由写在决定 7。
+///
 /// 令牌只收窄：`pat` 在的时候，路由先问 [`Self::require_write`] 或
 /// [`Self::reaches`]，角色照旧由 `require_kb` 与工作区角色判。
 pub struct SessionOrToken {