From 5fd5fbc82d22c859ef31f3877c1ddee6b1b5249f Mon Sep 17 00:00:00 2001 From: Bogdan Szabo Date: Fri, 16 Oct 2026 13:59:46 +0200 Subject: [PATCH] A client reads a document back by the external_id it pushed it under Signed-off-by: Bogdan Szabo Co-Authored-By: Claude Opus 5.5 --- .../utopia-server/src/api/documents_routes.rs | 99 ++++++++++- crates/utopia-server/src/api/mod.rs | 5 + .../utopia-server/src/api/sources_routes.rs | 15 +- .../src/api/sources_statements_tests.rs | 157 ++++++++++++++++++ crates/utopia-server/src/auth.rs | 2 + ...ty-from-the-person-scope-from-the-token.md | 9 +- ...052-document-content-is-a-read-contract.md | 13 ++ docs/design/access-and-audit.md | 7 + web/src/docs/ingest.md | 35 ++++ 9 files changed, 330 insertions(+), 12 deletions(-) diff --git a/crates/utopia-server/src/api/documents_routes.rs b/crates/utopia-server/src/api/documents_routes.rs index 0d184dea7..cff69bcca 100644 --- a/crates/utopia-server/src/api/documents_routes.rs +++ b/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, +} + +/// 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, + serves: impl FnOnce(&Document) -> bool, +) -> ApiResult> { 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, + reader: SessionOrToken, + Path((kb_id, source_id)): Path<(Uuid, Uuid)>, + Query(query): Query, +) -> ApiResult { + 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, diff --git a/crates/utopia-server/src/api/mod.rs b/crates/utopia-server/src/api/mod.rs index ec25e1fe6..db71ba3b0 100644 --- a/crates/utopia-server/src/api/mod.rs +++ b/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), diff --git a/crates/utopia-server/src/api/sources_routes.rs b/crates/utopia-server/src/api/sources_routes.rs index 6e8cec739..7a4e62eb8 100644 --- a/crates/utopia-server/src/api/sources_routes.rs +++ b/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 { 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 diff --git a/crates/utopia-server/src/api/sources_statements_tests.rs b/crates/utopia-server/src/api/sources_statements_tests.rs index 9d97827aa..ce1085d7f 100644 --- a/crates/utopia-server/src/api/sources_statements_tests.rs +++ b/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 { + 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 +} diff --git a/crates/utopia-server/src/auth.rs b/crates/utopia-server/src/auth.rs index f091f4f91..f8a9ff910 100644 --- a/crates/utopia-server/src/auth.rs +++ b/crates/utopia-server/src/auth.rs @@ -179,6 +179,8 @@ impl FromRequestParts for AuthUser { /// 用这个提取器的只有文档原件（0052）、「把一个库搭起来」那几条和读图那几条：搜实体、 /// 实体详情、认知历史、事实证据、图总览、邻域（决定 7）。 /// +/// 按推送时的 `external_id` 读回一篇文档（#1276）也是文档原件，开给令牌的理由写在决定 7。 +/// /// 令牌只收窄：`pat` 在的时候，路由先问 [`Self::require_write`] 或 /// [`Self::reaches`]，角色照旧由 `require_kb` 与工作区角色判。 pub struct SessionOrToken { diff --git a/docs/decisions/0014-identity-from-the-person-scope-from-the-token.md b/docs/decisions/0014-identity-from-the-person-scope-from-the-token.md index 850e9015b..f005c3589 100644 --- a/docs/decisions/0014-identity-from-the-person-scope-from-the-token.md +++ b/docs/decisions/0014-identity-from-the-person-scope-from-the-token.md @@ -100,6 +100,13 @@ push documents in. ceiling and Viewer suffices; a token whose `kb_ids` exclude the base gets 404, as for document content (0052). Changing an entity on the same path (`PATCH`) still takes a session. + Reading a pushed document back by the `external_id` it was pushed under + (`GET /kbs/{id}/sources/{sid}/documents/by-external-id`, added 2026-10-19, #1276) takes a + token on the same terms. It is a document-content read (0052), but a client pushing into an + `api` or `statements` source names its documents by its own ids and keeps none of ours, so + without it the client can check what landed only by searching for its own text. The source's + push key is not a credential here, so a leaked key stays push-only. + ## Dead ends - **Base-level machine tokens**, one per base and unrelated to a person. It would be a third @@ -126,7 +133,7 @@ push documents in. token now creates `api` sources too. - 2026-10-16: decision 7 opens four graph reads to any token; decision 6's "only these routes and document content" now also counts them. 2026-10-19: the graph overview and neighbourhood - join them. + join them, and so does reading a pushed document back by its `external_id` (#1276). ## Open questions diff --git a/docs/decisions/0052-document-content-is-a-read-contract.md b/docs/decisions/0052-document-content-is-a-read-contract.md index a53c45597..650859f1d 100644 --- a/docs/decisions/0052-document-content-is-a-read-contract.md +++ b/docs/decisions/0052-document-content-is-a-read-contract.md @@ -57,3 +57,16 @@ Range requests, multipart previews, transcoding, a hash-keyed public blob route, or a projection of bytes into derived text. Those are media-delivery contracts and should be designed after callers rely on this byte-exact baseline. + +## Revisions + +- 2026-10-19 (#1276): a third Viewer-level read finds a pushed document by the identity its + source pushed it under: + `GET /api/v1/kbs/{kb_id}/sources/{source_id}/documents/by-external-id?external_id=…`. It + answers an `api` or `statements` source's document with its metadata, the current version's + content as text, and the list `/versions` returns; older text stays behind + `/documents/{id}/content?version=N`. It takes the same credentials as the two reads above and + not the source's push key, so a leaked push key stays push-only. The trash, tombstone and + identity are judged on the document row the content read locks, and anything but a live + document under that identity is `404`; purge erases the identity, so a purged one is `404` + as well, not `410`. diff --git a/docs/design/access-and-audit.md b/docs/design/access-and-audit.md index 86e533c38..2ad0ab6c2 100644 --- a/docs/design/access-and-audit.md +++ b/docs/design/access-and-audit.md @@ -23,6 +23,13 @@ creates no base. These routes, document content and six graph reads (entity sear history, fact evidence, overview, neighbourhood; Viewer suffices, `kb_ids` answers 404) take `SessionOrToken`; apart from MCP, every other route is session-only [0014 d6, d7]. `ingest_token` on a source is a different kind: plaintext, push-only [0014]. +**A client reads a pushed document back by its `external_id`.** +`GET /kbs/{id}/sources/{sid}/documents/by-external-id?external_id=…` finds an `api` or `statements` +source's document by the id it was pushed under and returns its metadata, current text and version +list. It is document content: Viewer, a session or any token whose `kb_ids` reach the base, and not +the source's push key, so a leaked key stays push-only; a document in the trash, a tombstoned id +and one never pushed are all 404 [0014 d7, 0052]. + **Another app can trade its user's fresh ID token for a session.** `POST /auth/oidc/exchange` accepts an ID token from the configured issuer whose audience is in `UTOPIA_OIDC_EXCHANGE_AUDIENCES` and that was issued in the last ten minutes, maps it only to an diff --git a/web/src/docs/ingest.md b/web/src/docs/ingest.md index e7bce0500..e2611fade 100644 --- a/web/src/docs/ingest.md +++ b/web/src/docs/ingest.md @@ -162,6 +162,41 @@ The response is the API push's: `{ "action": "created" }` · `updated` · `uncha --- +## Reading a pushed document back + +An API or Statements source's document can be read back by the identity it was pushed under: its `external_id`, or for an API push without one, its `filename`. The identity goes in the query string, URL-encoded, so it may contain `/`. + +``` +GET {your-utopia-base}/api/v1/kbs/{kb_id}/sources/{source_id}/documents/by-external-id?external_id=note-42 +Authorization: Bearer utp_pat_… +``` + +The credential is a personal access token (or a signed-in session) of someone who can view the base; a token limited to other bases gets `404`. The source's push token is **not** accepted here, so a leaked push token can only push. + +```json +{ + "id": "0199…", + "external_id": "note-42", + "filename": "runbook.md", + "mime": "text/plain", + "sha256": "9f2c…", + "size_bytes": 57, + "status": "ready", + "doc_time": "2026-08-20T09:00:00Z", + "updated_at": "2026-08-20T09:00:04Z", + "version": 2, + "content": "# Runbook v2 …", + "versions": [ + { "version": 1, "sha256": "41aa…", "size_bytes": 54, "ingested_at": "2026-08-19T10:00:00Z" }, + { "version": 2, "sha256": "9f2c…", "size_bytes": 57, "ingested_at": "2026-08-20T09:00:04Z" } + ] +} +``` + +`content` is the current version's text; older versions are at `GET /api/v1/documents/{id}/content?version=N`. An identity marked "Not in source", a document in the trash or purged, and an identity this source never pushed are all `404`. + +--- + ## Shared semantics - **Identity, not filenames.** Documents are tracked by `custom:{id}` / `api:{external_id}` / `statements:{external_id}` keys. Renames are recognized as moves; content changes update the same document.