From 304b0ef99002362948a204f1d491ee877a85a394 Mon Sep 17 00:00:00 2001 From: wangzifei Date: Tue, 22 Sep 2026 00:41:27 +0800 Subject: [PATCH] feat(api): serve retained document content and versions Signed-off-by: wangzifei --- .../utopia-server/src/api/documents_routes.rs | 221 +++++++++++++++ .../src/api/documents_routes_tests.rs | 253 +++++++++++++++++- crates/utopia-server/src/api/mod.rs | 2 + .../utopia-server/src/api/sources_routes.rs | 2 +- crates/utopia-store/src/documents.rs | 24 ++ crates/utopia-store/src/tokens.rs | 2 +- ...052-document-content-is-a-read-contract.md | 59 ++++ 7 files changed, 560 insertions(+), 3 deletions(-) create mode 100644 docs/decisions/0052-document-content-is-a-read-contract.md diff --git a/crates/utopia-server/src/api/documents_routes.rs b/crates/utopia-server/src/api/documents_routes.rs index bf111f716..4a2fcdc97 100644 --- a/crates/utopia-server/src/api/documents_routes.rs +++ b/crates/utopia-server/src/api/documents_routes.rs @@ -1,13 +1,21 @@ +use axum::extract::FromRequestParts; use axum::extract::{Multipart, Path, Query, State}; +use axum::http::request::Parts; +use axum::http::{header, HeaderMap, HeaderValue, StatusCode}; +use axum::response::{IntoResponse, Response}; use axum::Json; +use axum_extra::extract::cookie::CookieJar; +use percent_encoding::{percent_encode, AsciiSet, NON_ALPHANUMERIC}; use serde::Deserialize; use serde_json::json; use sha2::{Digest, Sha256}; use utopia_core::models::{Document, Role}; use utopia_core::AppError; +use utopia_store::tokens::Authenticated; use uuid::Uuid; use crate::auth::AuthUser; +use crate::error::ApiErr; use crate::error::ApiResult; use crate::state::AppState; @@ -18,6 +26,219 @@ pub struct UploadQuery { pub source: Option, } +const PAT_PREFIX: &str = utopia_store::tokens::PREFIX; + +/// Web session or personal access token, resolved far enough to enforce both +/// identity and token scope. +/// +/// `AuthUser` cannot play this role: it interprets every bearer string as a +/// JWT, so the PAT designed for API clients would become a 401 before the +/// route could apply its Viewer check. +pub struct DocumentReader { + pub user: utopia_core::models::User, + pub pat: Option, +} + +impl FromRequestParts for DocumentReader { + type Rejection = ApiErr; + + async fn from_request_parts( + parts: &mut Parts, + state: &AppState, + ) -> Result { + let raw = CookieJar::from_headers(&parts.headers) + .get(crate::auth::COOKIE_NAME) + .map(|cookie| cookie.value().to_string()) + .or_else(|| { + parts + .headers + .get(header::AUTHORIZATION) + .and_then(|value| value.to_str().ok()) + .and_then(|value| value.strip_prefix("Bearer ")) + .map(str::to_owned) + }) + .ok_or(AppError::Unauthorized)?; + + if raw.starts_with(PAT_PREFIX) { + let auth = utopia_store::tokens::authenticate(&state.pool, raw.trim()).await?; + let user = utopia_store::accounts::find_user_by_id(&state.pool, auth.user_id) + .await? + .ok_or(AppError::Unauthorized)?; + return Ok(Self { + user, + pat: Some(auth), + }); + } + + let user_id = crate::auth::decode_user_id(state, &raw)?; + let user = utopia_store::accounts::find_user_by_id(&state.pool, user_id) + .await? + .ok_or(AppError::Unauthorized)?; + Ok(Self { user, pat: None }) + } +} + +#[derive(Deserialize)] +pub struct ContentQuery { + #[serde(default)] + pub version: Option, +} + +const PURGED_MESSAGE: &str = "The document contents have been purged"; + +fn gone(message: &'static str) -> Response { + (StatusCode::GONE, Json(json!({ "error": message }))).into_response() +} + +fn missing_blob_invariant(document_id: Uuid, sha256: &str) -> AppError { + AppError::Other(anyhow::anyhow!( + "document {document_id} ledger references unavailable blob {sha256}" + )) +} + +fn content_disposition(filename: &str) -> String { + const FILENAME: &AsciiSet = &NON_ALPHANUMERIC.remove(b'-').remove(b'.').remove(b'_'); + + let fallback: String = filename + .chars() + .map(|c| { + if c.is_ascii_alphanumeric() || matches!(c, '-' | '.' | '_') { + c + } else { + '_' + } + }) + .collect(); + let encoded = percent_encode(filename.as_bytes(), FILENAME); + format!("attachment; filename=\"{fallback}\"; filename*=UTF-8''{encoded}") +} + +fn content_headers( + document: &Document, + version: &utopia_store::documents::DocumentVersion, + byte_count: usize, +) -> ApiResult { + let mime = HeaderValue::from_str(&document.mime) + .map_err(|_| anyhow::anyhow!("document {} has an invalid MIME header", document.id))?; + let disposition = HeaderValue::from_str(&content_disposition(&document.filename)) + .map_err(|_| anyhow::anyhow!("document {} has an unsafe filename", document.id))?; + let mut headers = HeaderMap::new(); + headers.insert(header::CONTENT_TYPE, mime); + headers.insert( + header::CONTENT_LENGTH, + HeaderValue::from_str(&byte_count.to_string()) + .map_err(|_| anyhow::anyhow!("content length is not a valid header"))?, + ); + headers.insert( + header::ETAG, + HeaderValue::from_str(&format!("\"{}\"", version.sha256)) + .map_err(|_| anyhow::anyhow!("document digest is not a valid header"))?, + ); + headers.insert(header::CONTENT_DISPOSITION, disposition); + Ok(headers) +} + +async fn require_reader_kb( + state: &AppState, + reader: &DocumentReader, + kb_id: Uuid, +) -> ApiResult<()> { + utopia_store::access::require_kb(&state.pool, &reader.user, kb_id, Role::Viewer).await?; + if let Some(pat) = &reader.pat { + if !pat.covers(kb_id) { + return Err(AppError::NotFound.into()); + } + } + Ok(()) +} + +/// Serve the retained original named by the document ledger. +pub async fn content( + State(state): State, + reader: DocumentReader, + Path(id): Path, + Query(query): Query, +) -> ApiResult { + if query.version.is_some_and(|version| version < 1) { + return Err(AppError::invalid("bad_version", "Version must be 1 or greater").into()); + } + let document = utopia_store::documents::get(&state.pool, id).await?; + require_reader_kb(&state, &reader, document.kb_id).await?; + if document.purged_at.is_some() { + return Ok(gone(PURGED_MESSAGE)); + } + + // 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. + 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)); + } + let version: utopia_store::documents::DocumentVersion = match query.version { + Some(requested) => sqlx::query_as( + "SELECT version, sha256, size_bytes, ingested_at + FROM document_versions WHERE document_id = $1 AND version = $2", + ) + .bind(id) + .bind(requested) + .fetch_optional(&mut *tx) + .await? + .ok_or(AppError::NotFound)?, + None => sqlx::query_as( + "SELECT version, sha256, size_bytes, ingested_at + FROM document_versions WHERE document_id = $1 AND sha256 = $2 + ORDER BY version DESC LIMIT 1", + ) + .bind(id) + .bind(&document.sha256) + .fetch_optional(&mut *tx) + .await? + .ok_or_else(|| { + anyhow::anyhow!( + "document {} has no ledger version for its current digest", + id + ) + })?, + }; + let bytes = state + .blob + .get(&version.sha256) + .await + .map_err(|_| missing_blob_invariant(id, &version.sha256))?; + tx.commit().await?; + + if version.size_bytes != bytes.len() as i64 { + return Err(anyhow::anyhow!( + "document {} version {} has an inaccurate ledger size", + id, + version.version + ) + .into()); + } + let headers = content_headers(&document, &version, bytes.len())?; + Ok((StatusCode::OK, headers, bytes).into_response()) +} + +/// Name the exact retained versions the content route can address. +pub async fn versions( + State(state): State, + reader: DocumentReader, + Path(id): Path, +) -> ApiResult { + let document = utopia_store::documents::get(&state.pool, id).await?; + require_reader_kb(&state, &reader, document.kb_id).await?; + if document.purged_at.is_some() { + return Ok(gone(PURGED_MESSAGE)); + } + let versions = utopia_store::documents::versions(&state.pool, id).await?; + Ok((StatusCode::OK, Json(json!({ "versions": versions }))).into_response()) +} + /// 批量上传（multipart，可多文件）。重复内容（同 KB 同 sha256）跳过。 pub async fn upload( State(state): State, diff --git a/crates/utopia-server/src/api/documents_routes_tests.rs b/crates/utopia-server/src/api/documents_routes_tests.rs index 682b524b2..fa3c847ed 100644 --- a/crates/utopia-server/src/api/documents_routes_tests.rs +++ b/crates/utopia-server/src/api/documents_routes_tests.rs @@ -1,6 +1,6 @@ use super::content_time; use axum::body::{to_bytes, Body}; -use axum::http::{Request, StatusCode}; +use axum::http::{HeaderMap, Request, StatusCode}; use chrono::{DateTime, Utc}; use serde_json::{json, Value}; use std::sync::Arc; @@ -50,6 +50,238 @@ fn only_a_complete_opening_dateline_sets_the_date() { } } +impl Fixture { + async fn get_raw( + &self, + path: &str, + token: Option<&str>, + ) -> anyhow::Result<(StatusCode, HeaderMap, Vec)> { + let mut request = Request::get(path); + if let Some(token) = token { + request = request.header("Authorization", format!("Bearer {token}")); + } + let response = self + .app + .clone() + .oneshot(request.body(Body::empty())?) + .await?; + let status = response.status(); + let headers = response.headers().clone(); + let bytes = to_bytes(response.into_body(), 128 * 1024 * 1024) + .await? + .to_vec(); + Ok((status, headers, bytes)) + } + + async fn get_json( + &self, + path: &str, + token: Option<&str>, + ) -> anyhow::Result<(StatusCode, Value)> { + let (status, _, bytes) = self.get_raw(path, token).await?; + Ok((status, serde_json::from_slice(&bytes)?)) + } +} + +fn content_headers(headers: &HeaderMap, sha256: &str, size: usize) { + assert_eq!( + headers["content-type"], "application/x-audit-record", + "the ledger's MIME, not a guessed type" + ); + assert_eq!(headers["content-length"], size.to_string()); + assert_eq!(headers["etag"], format!("\"{sha256}\"")); + assert_eq!( + headers["content-disposition"], + "attachment; filename=\"audit.bin\"; filename*=UTF-8''audit.bin" + ); +} + +#[tokio::test] +async fn document_content_serves_the_current_and_recorded_versions() -> anyhow::Result<()> { + let Some(f) = Fixture::new().await? else { + return Ok(()); + }; + let original = b"generation one".to_vec(); + let doc = f.create_retained(f.kb, &original).await?; + + let path = format!("/api/v1/documents/{}/content", doc.id); + let (status, headers, bytes) = f.get_raw(&path, Some(&f.token)).await?; + assert_eq!( + status, + StatusCode::OK, + "{}", + String::from_utf8_lossy(&bytes) + ); + assert_eq!(bytes, original); + content_headers(&headers, &doc.sha256, original.len()); + + let revised = b"generation two has grown".to_vec(); + use sha2::{Digest, Sha256}; + let revised_sha = super::hex(&Sha256::digest(&revised)); + f.state.blob.put(&revised_sha, &revised).await?; + documents::replace_content_and_enqueue_processing( + &f.pool, + doc.id, + &doc.filename, + &doc.mime, + revised.len() as i64, + &revised_sha, + None, + ) + .await?; + let (status, headers, bytes) = f + .get_raw(&format!("{path}?version=2"), Some(&f.token)) + .await?; + assert_eq!( + status, + StatusCode::OK, + "{}", + String::from_utf8_lossy(&bytes) + ); + assert_eq!(bytes, revised); + content_headers(&headers, &revised_sha, revised.len()); + + let (status, headers, bytes) = f + .get_raw(&format!("{path}?version=1"), Some(&f.token)) + .await?; + assert_eq!( + status, + StatusCode::OK, + "{}", + String::from_utf8_lossy(&bytes) + ); + assert_eq!(bytes, original); + content_headers(&headers, &doc.sha256, original.len()); + f.cleanup().await +} + +#[tokio::test] +async fn versions_ledger_names_what_content_can_serve() -> anyhow::Result<()> { + let Some(f) = Fixture::new().await? else { + return Ok(()); + }; + let original = b"auditable history"; + let doc = f.create_retained(f.kb, original).await?; + + let (status, body) = f + .get_json( + &format!("/api/v1/documents/{}/versions", doc.id), + Some(&f.token), + ) + .await?; + assert_eq!(status, StatusCode::OK, "{body}"); + let versions = body["versions"].as_array().expect("version ledger"); + assert_eq!(versions.len(), 1); + assert_eq!(versions[0]["version"], 1); + assert_eq!(versions[0]["sha256"], doc.sha256); + assert_eq!(versions[0]["size_bytes"], original.len() as i64); + assert!(versions[0]["ingested_at"].is_string()); + + let path = format!("/api/v1/documents/{}/content?version=99", doc.id); + let (status, _, bytes) = f.get_raw(&path, Some(&f.token)).await?; + assert_eq!(status, StatusCode::NOT_FOUND, "{:?}", bytes); + f.cleanup().await +} + +#[tokio::test] +async fn deleted_bytes_stay_readable_and_purged_tombstones_answer_gone() -> anyhow::Result<()> { + let Some(f) = Fixture::new().await? else { + return Ok(()); + }; + let original = b"retained after deletion"; + let doc = f.create_retained(f.kb, original).await?; + documents::delete(&f.pool, f.kb, doc.id, None).await?; + + let path = format!("/api/v1/documents/{}/content", doc.id); + let (status, _, bytes) = f.get_raw(&path, Some(&f.token)).await?; + assert_eq!( + status, + StatusCode::OK, + "{}", + String::from_utf8_lossy(&bytes) + ); + assert_eq!(bytes, original); + + documents::purge(&f.pool, f.kb, doc.id).await?; + let (status, _, bytes) = f.get_raw(&path, Some(&f.token)).await?; + assert_eq!(status, StatusCode::GONE, "{:?}", bytes); + let versions_path = format!("/api/v1/documents/{}/versions", doc.id); + let (status, _, bytes) = f.get_raw(&versions_path, Some(&f.token)).await?; + assert_eq!(status, StatusCode::GONE, "{:?}", bytes); + f.cleanup().await +} + +#[tokio::test] +async fn a_ledger_referenced_missing_blob_is_an_invariant_failure() -> anyhow::Result<()> { + let Some(f) = Fixture::new().await? else { + return Ok(()); + }; + let (status, created) = f + .upload(f.kb, "", &[("audit.bin", "still promised")]) + .await?; + assert_eq!(status, StatusCode::OK, "{created}"); + let doc = f.created_docs(&created).await?.remove(0); + f.state.blob.delete(&doc.sha256).await?; + + let path = format!("/api/v1/documents/{}/content", doc.id); + let (status, _, bytes) = f.get_raw(&path, Some(&f.token)).await?; + assert_eq!(status, StatusCode::INTERNAL_SERVER_ERROR, "{:?}", bytes); + f.cleanup().await +} + +#[tokio::test] +async fn content_reads_keep_viewer_access_pat_scope_and_reject_source_tokens() -> anyhow::Result<()> +{ + let Some(f) = Fixture::new().await? else { + return Ok(()); + }; + let doc = f.create_retained(f.kb, b"scoped bytes").await?; + let content_path = format!("/api/v1/documents/{}/content", doc.id); + let versions_path = format!("/api/v1/documents/{}/versions", doc.id); + + sqlx::query("UPDATE kb_members SET role='viewer' WHERE kb_id=$1 AND user_id=$2") + .bind(f.kb) + .bind(f.user) + .execute(&f.pool) + .await?; + let (status, _, bytes) = f.get_raw(&content_path, Some(&f.token)).await?; + assert_eq!( + status, + StatusCode::OK, + "{}", + String::from_utf8_lossy(&bytes) + ); + + let (_, pat) = + utopia_store::tokens::issue(&f.pool, f.user, "audit", "read", None, None).await?; + for path in [&content_path, &versions_path] { + let (status, _, bytes) = f.get_raw(path, Some(&pat)).await?; + assert_eq!( + status, + StatusCode::OK, + "{}", + String::from_utf8_lossy(&bytes) + ); + } + + let (_, scoped_pat) = utopia_store::tokens::issue( + &f.pool, + f.user, + "other base only", + "read", + Some(&[f.other_kb]), + None, + ) + .await?; + let (status, _, bytes) = f.get_raw(&content_path, Some(&scoped_pat)).await?; + assert_eq!(status, StatusCode::NOT_FOUND, "{:?}", bytes); + + let source_token = crate::api::sources_routes::new_ingest_token(); + let (status, _, bytes) = f.get_raw(&content_path, Some(&source_token)).await?; + assert_eq!(status, StatusCode::UNAUTHORIZED, "{:?}", bytes); + f.cleanup().await +} + #[test] fn header_decoding_is_bounded_and_never_accepts_a_truncated_line() { let expected = "2024-02-29T00:00:00Z".parse::>().unwrap(); @@ -160,6 +392,25 @@ impl Fixture { })) } + async fn create_retained(&self, kb: Uuid, bytes: &[u8]) -> anyhow::Result { + use sha2::{Digest, Sha256}; + + let sha256 = super::hex(&Sha256::digest(bytes)); + self.state.blob.put(&sha256, bytes).await?; + Ok(documents::create_with_version_and_processing( + &self.pool, + kb, + "audit.bin", + "application/x-audit-record", + bytes.len() as i64, + &sha256, + None, + None, + None, + ) + .await?) + } + async fn upload( &self, kb: Uuid, diff --git a/crates/utopia-server/src/api/mod.rs b/crates/utopia-server/src/api/mod.rs index b5aad5e14..53e7c1b38 100644 --- a/crates/utopia-server/src/api/mod.rs +++ b/crates/utopia-server/src/api/mod.rs @@ -400,6 +400,8 @@ pub fn router(state: AppState, cfg: &AppConfig) -> Router { "/documents/{id}", get(documents_routes::detail).delete(documents_routes::delete), ) + .route("/documents/{id}/content", get(documents_routes::content)) + .route("/documents/{id}/versions", get(documents_routes::versions)) // 撤销删除（#268）：删除是墓碑，所以有得撤 .route("/documents/{id}/restore", post(documents_routes::restore)) // 真删（#268 下半）：只对已删除的开放，库管理员 diff --git a/crates/utopia-server/src/api/sources_routes.rs b/crates/utopia-server/src/api/sources_routes.rs index 19f38c892..105600cc0 100644 --- a/crates/utopia-server/src/api/sources_routes.rs +++ b/crates/utopia-server/src/api/sources_routes.rs @@ -15,7 +15,7 @@ use crate::error::ApiResult; use crate::state::AppState; /// 生成 api 来源的推送密钥。 -fn new_ingest_token() -> String { +pub(crate) fn new_ingest_token() -> String { format!("utp_{}{}", Uuid::new_v4().simple(), Uuid::new_v4().simple()) } diff --git a/crates/utopia-store/src/documents.rs b/crates/utopia-store/src/documents.rs index 168585199..f64637fd1 100644 --- a/crates/utopia-store/src/documents.rs +++ b/crates/utopia-store/src/documents.rs @@ -1,5 +1,6 @@ use chrono::{DateTime, Utc}; use pgvector::Vector; +use serde::Serialize; use sqlx::{PgPool, Postgres, Transaction}; use utopia_core::models::{ChunkView, Document, DocumentPage}; use utopia_core::{AppError, AppResult}; @@ -544,6 +545,29 @@ pub async fn get(pool: &PgPool, id: Uuid) -> AppResult { .ok_or(AppError::NotFound) } +/// A recorded original, with only the facts the ingestion ledger guarantees. +#[derive(Debug, Clone, Serialize, sqlx::FromRow)] +pub struct DocumentVersion { + pub version: i32, + pub sha256: String, + pub size_bytes: i64, + pub ingested_at: DateTime, +} + +/// Every retained original version, oldest first. +/// +/// This deliberately includes soft-deleted documents: their bytes remain part +/// of the auditable record until the separate purge action removes them. +pub async fn versions(pool: &PgPool, id: Uuid) -> AppResult> { + Ok(sqlx::query_as( + "SELECT version, sha256, size_bytes, ingested_at + FROM document_versions WHERE document_id = $1 ORDER BY version", + ) + .bind(id) + .fetch_all(pool) + .await?) +} + /// 按 kb 收窄的取文档。**id 由模型给出时只能走这一支**：`get` 只按 id 查， /// 一个别的库的 id 照样查得到。 pub async fn find_in_kb(pool: &PgPool, kb_id: Uuid, id: Uuid) -> AppResult> { diff --git a/crates/utopia-store/src/tokens.rs b/crates/utopia-store/src/tokens.rs index ce165c4d3..06d0f7608 100644 --- a/crates/utopia-store/src/tokens.rs +++ b/crates/utopia-store/src/tokens.rs @@ -19,7 +19,7 @@ use uuid::Uuid; /// 明文令牌的前缀。与 `sources.ingest_token` 的 `utp_` 区分开—— /// 两者能干的事差很远，在日志或配置文件里一眼要认得出是哪一种 -const PREFIX: &str = "utp_pat_"; +pub const PREFIX: &str = "utp_pat_"; /// 列表里给人认的那一小截（含前缀）。够对上配置文件里那一串，又不足以复原 const SHOWN: usize = 16; diff --git a/docs/decisions/0052-document-content-is-a-read-contract.md b/docs/decisions/0052-document-content-is-a-read-contract.md new file mode 100644 index 000000000..2860fe108 --- /dev/null +++ b/docs/decisions/0052-document-content-is-a-read-contract.md @@ -0,0 +1,59 @@ +# 0052 · Document content is a read contract over the retained ledger + +- **Status**: proposed for review +- **Written**: 2026-09-21 +- **Related**: [#859](https://github.com/deeplethe/utopia/issues/859); [0014](0014-identity-from-the-person-scope-from-the-token.md); [0040](0040-a-chunk-says-where-its-words-came-from.md) + +## Problem + +Ingestion retains content-addressed originals and records their SHA-256 +digests, but the HTTP surface can expose derived text, chunks, and facts. A +client therefore cannot download the exact bytes that a document's digest +describes, compare those bytes to the ledger, or replay a named historical +version. The omission also turns an auditable invariant into an internal +assumption: nothing on the public boundary says whether a recorded digest can +still be served. + +## Decision + +Add two Viewer-level reads: + +* `GET /api/v1/documents/{id}/content[?version=N]` serves one retained + original. No query means the current version; a version number addresses a + recorded ledger row. +* `GET /api/v1/documents/{id}/versions` returns the ledger's `version`, + `sha256`, `size_bytes`, and `ingested_at`. + +Content is addressed by document identity, not blob identity. The handler +selects and locks the document plus its ledger row, reads the immutable blob +inside that window, and only then releases the database transaction. This +closes the replacement/purge race rather than asking the client to retry a +claim that was briefly true. + +The response carries the ledger MIME, actual byte length, a strong SHA-derived +`ETag` in quoted form, and an RFC 5987/6266 `Content-Disposition`. Historical +bytes reuse the document's current display metadata because the version ledger +records content identity and size, not a frozen historical display name or +MIME. This is an explicit compatibility boundary, not a claim that old uploads +carried metadata history. + +Deletion is reversible and bytes remain readable. Purge is final: its +tombstone answers `410 Gone` on both routes. A ledger row whose blob is absent +is not a normal missing resource; it is an internal invariant failure and +answers `500`. This keeps a storage fault distinguishable from a bad document +ID or version. + +## Access + +Both routes reuse the Viewer authorization rule. They accept a web session or +a `utp_pat_` personal access token. Token KB scoping is still a separate +narrowing check; a scoped token receives the same `404` as an inaccessible +document. Source ingest tokens remain rejected as credentials. + +## Limits + +The route buffers within the existing upload cap. It deliberately does not add +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.