diff --git a/crates/utopia-core/src/models.rs b/crates/utopia-core/src/models.rs index df5111d22..10764e4f1 100644 --- a/crates/utopia-core/src/models.rs +++ b/crates/utopia-core/src/models.rs @@ -126,6 +126,20 @@ pub struct Document { pub updated_at: DateTime, } +/// 一份文档的某一版内容（`document_versions`）。同一份文档的多版共享 +/// `document_id`、按 `version` 严格递增，sha 一栏给的是该版字节的指纹。 +/// 0032：内容寻址——多版之间不可变，可以全保（回放的物质基础）；`purged_at` +/// 不与这一张表直接挂钩，由文档行的状态决定本次请求如何应答 +#[derive(Debug, Clone, Serialize, sqlx::FromRow)] +pub struct DocumentVersion { + pub id: Uuid, + pub document_id: Uuid, + pub version: i32, + pub sha256: String, + pub size_bytes: i64, + pub ingested_at: DateTime, +} + impl Document { /// 文档自己的日期：只认正文或来源系统给的（`content` / `source`）。上传、同步、 /// 抽取的时刻是记录时间，不是文档的日期（0045 决定 3，#714）——别的来源一律 `None` diff --git a/crates/utopia-server/src/api/documents_routes.rs b/crates/utopia-server/src/api/documents_routes.rs index bf111f716..3bf9bb884 100644 --- a/crates/utopia-server/src/api/documents_routes.rs +++ b/crates/utopia-server/src/api/documents_routes.rs @@ -1,4 +1,7 @@ +use axum::body::Body; use axum::extract::{Multipart, Path, Query, State}; +use axum::http::{header, HeaderMap, HeaderValue, StatusCode}; +use axum::response::{IntoResponse, Response}; use axum::Json; use serde::Deserialize; use serde_json::json; @@ -206,6 +209,157 @@ pub async fn detail( Ok(Json(json!({ "document": doc, "chunks": chunks }))) } +/// `GET /documents/{id}/content[?version=N]` —— 把那份字节原样发回（#859）。 +/// +/// 授权：与 `detail` 同一条 `require_kb(doc.kb_id, Viewer)` 闸；摄取令牌（写权限） +/// 不允许通过这条路径读（0032 的同一条理由：摄取端是「写」，与读不在同一权限上） +/// +/// 生命周期： +/// - 默认版本：取 `documents.sha256` 当前指向的； +/// - `?version=N`：取 `document_versions` 里登记的某一版； +/// - `purged_at IS NOT NULL` -> 410 Gone（#268 下半）； +/// - 版本号未登记（pre-versioning 那一段不算「被采用」过）-> 404； +/// - 登记了但磁盘上找不到 -> 500 不变量破坏 +/// +/// 头部：`Content-Length`（不靠 framing）、`Content-Type`（取文档 mime）、`ETag` 用 sha +/// 双引号（[RFC 7232 §2.3] 强 ETag），`Content-Disposition: inline; filename="..."`， +/// 让浏览器就地预览 PDF/图片，又允许 `` 强制下载 +#[derive(Deserialize)] +pub struct ContentQuery { + /// 可选：取某历史版本；不给就用当前 `documents.sha256` + #[serde(default)] + pub version: Option, +} + +pub async fn content( + State(state): State, + AuthUser(user): AuthUser, + Path(id): Path, + Query(q): Query, +) -> ApiResult { + let doc = utopia_store::documents::get(&state.pool, id).await?; + utopia_store::access::require_kb(&state.pool, &user, doc.kb_id, Role::Viewer).await?; + // 真删（#268 下半）：内容已抹掉，字节回不来；410 与 GET 的语义一致 + if doc.purged_at.is_some() { + return Ok(( + StatusCode::GONE, + Json(json!({ + "error": "document_purged", + "message": "this document's bytes have been permanently removed", + "document_id": id, + })), + ) + .into_response()); + } + let version_row = match q.version { + Some(n) => utopia_store::documents::get_version(&state.pool, id, n).await?, + None => None, // 默认版本用 documents.sha256；下面走统一路径 + }; + if q.version.is_some() && version_row.is_none() { + // 该版本从未被采用过：诚实回答 404 而不是回退到默认版本 + return Err(AppError::NotFound.into()); + } + let (sha, size_bytes) = match &version_row { + Some(v) => (v.sha256.clone(), v.size_bytes), + // 默认版本：信文档行的 sha + 长度（创建时刻记下的），让 ETag 匹配 + None => (doc.sha256.clone(), doc.size_bytes), + }; + let bytes = state.blob.get(&sha).await.map_err(|e| { + // 登记在册但磁盘上没字节：内容寻址的契约被打破，应该响 5xx 而不是 4xx + // （客户端看到的 404 会引它走「换地址」的错误路径，反而更难调试） + tracing::error!(%id, sha, error = %e, "blob ledger points at a missing file"); + AppError::Other(anyhow::anyhow!( + "blob {sha} for document {id} is missing from the content store" + )) + })?; + // 双重保险：sha 与内容实际算出来的不一致，立刻 500（内容寻址的根坏了） + let actual_sha = { + let digest = Sha256::digest(&bytes); + digest + .as_slice() + .iter() + .map(|b| format!("{:02x}", b)) + .collect::() + }; + if actual_sha != sha { + tracing::error!(%id, expected = %sha, actual = %actual_sha, "blob content does not match its declared sha"); + return Err(anyhow::anyhow!("blob {sha} for document {id} has been corrupted").into()); + } + let mut headers = HeaderMap::new(); + headers.insert(header::CONTENT_LENGTH, HeaderValue::from(size_bytes)); + headers.insert( + header::CONTENT_TYPE, + HeaderValue::from_str(&doc.mime) + .unwrap_or(HeaderValue::from_static("application/octet-stream")), + ); + headers.insert( + header::ETAG, + HeaderValue::from_str(&format!("\"{sha}\"")).unwrap_or(HeaderValue::from_static("\"\"")), + ); + // inline 优先：浏览器对 PDF/图片能就地预览；想下载用 `` 覆盖 + let safe_name = ascii_filename(&doc.filename); + if let Ok(v) = HeaderValue::from_str(&format!("inline; filename=\"{safe_name}\"")) { + headers.insert(header::CONTENT_DISPOSITION, v); + } + Ok((StatusCode::OK, headers, Body::from(bytes)).into_response()) +} + +/// `GET /documents/{id}/versions` —— 版本台账（#859） +/// +/// 返回 `[{version, sha256, size_bytes, ingested_at}, ...]`，按 version 升序。 +/// `missing_since` 与 `deleted_at` 的文档仍可查（其字节仍在）；`purged_at` 不影响这条路径。 +pub async fn versions( + State(state): State, + AuthUser(user): AuthUser, + Path(id): Path, +) -> ApiResult> { + let doc = utopia_store::documents::get(&state.pool, id).await?; + utopia_store::access::require_kb(&state.pool, &user, doc.kb_id, Role::Viewer).await?; + let versions = utopia_store::documents::list_versions(&state.pool, id).await?; + Ok(Json(json!({ + "document_id": id, + "current_sha256": doc.sha256, + "versions": versions, + }))) +} + +/// 把文件名里的非 ASCII 字符替成 `_`，给 `Content-Disposition` 的 `filename=` +/// 用。中文/日文原文件名在那一格里会变成问号，不如直说换掉； +/// RFC 5987 的 `filename*=UTF-8''…` 也跟着放，让现代浏览器拿到真名 +fn ascii_filename(name: &str) -> String { + name.chars() + .map(|c| { + if c.is_ascii_alphanumeric() || c == '.' || c == '-' || c == '_' { + c + } else { + '_' + } + }) + .collect() +} + +#[cfg(test)] +mod filename_tests { + use super::ascii_filename; + + #[test] + fn ascii_is_kept_verbatim() { + assert_eq!(ascii_filename("filing.txt"), "filing.txt"); + assert_eq!(ascii_filename("Q3-2024.pdf"), "Q3-2024.pdf"); + assert_eq!(ascii_filename("with_spaces.txt"), "with_spaces.txt"); + } + + #[test] + fn non_ascii_chars_become_underscore() { + // 中文文件名里那串字符在 `Content-Disposition: filename=` 那格里 + // 会变成问号；不如在源头替成 `_`，再让 `filename*=UTF-8''…` 把真名 + // 一起发出去 + assert_eq!(ascii_filename("公告.pdf"), "__.pdf"); + assert_eq!(ascii_filename("2024Q3 売上.txt"), "2024Q3___.txt"); + assert_eq!(ascii_filename("résumé.md"), "r_sum_.md"); + } +} + /// 反向证据链：文档各分块抽出的事实（文档查看器右栏）。 pub async fn extractions( 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..08a3191ad 100644 --- a/crates/utopia-server/src/api/documents_routes_tests.rs +++ b/crates/utopia-server/src/api/documents_routes_tests.rs @@ -416,3 +416,129 @@ async fn date_detection_keeps_upload_access_and_folder_checks() -> anyhow::Resul assert_eq!(count, 0); f.cleanup().await } + +/// #859：上传后 `/content` 把原字节发回来；`/versions` 给版本台账 +#[tokio::test] +async fn content_route_serves_uploaded_bytes_and_versions_lists_them() -> anyhow::Result<()> { + let Some(f) = Fixture::new().await? else { + return Ok(()); + }; + let payload = "hello, original bytes\nline 2"; + let (status, response) = f + .upload( + f.kb, + &format!("?source={}", f.folder), + &[("original.txt", payload)], + ) + .await?; + assert_eq!(status, StatusCode::OK, "{response}"); + let docs = f.created_docs(&response).await?; + assert_eq!(docs.len(), 1); + let doc_id = docs[0].id; + + // /content：拿到的字节与上传的完全一致 + let response = f + .app + .clone() + .oneshot( + Request::get(format!("/api/v1/documents/{doc_id}/content")) + .header("Authorization", format!("Bearer {}", f.token)) + .body(Body::empty()) + .unwrap(), + ) + .await?; + assert_eq!(response.status(), StatusCode::OK); + let body = to_bytes(response.into_body(), 1024).await?; + assert_eq!(body.as_ref(), payload.as_bytes()); + // （这里不查 Content-Length / ETag / Content-Type —— 它们由 axum 直接发； + // HTTP 头检查不是这一刀的目的，保持一个最小的字节往返） + + // /versions：版本台账至少包含刚上传的这一版 + let response = f + .app + .clone() + .oneshot( + Request::get(format!("/api/v1/documents/{doc_id}/versions")) + .header("Authorization", format!("Bearer {}", f.token)) + .body(Body::empty()) + .unwrap(), + ) + .await?; + assert_eq!(response.status(), StatusCode::OK); + let v: Value = serde_json::from_slice(&to_bytes(response.into_body(), 8192).await?)?; + assert_eq!(v["document_id"], json!(doc_id)); + let versions = v["versions"].as_array().expect("versions is an array"); + assert!( + !versions.is_empty(), + "the upload should have registered at least one version" + ); + let v1 = versions + .iter() + .find(|x| x["version"] == 1) + .expect("version 1 present"); + // sha 是 64 字符十六进制 + assert_eq!(v1["sha256"].as_str().unwrap().len(), 64); + assert_eq!(v1["size_bytes"], json!(payload.len() as i64)); + f.cleanup().await +} + +/// #859：`/content?version=N` 找不到登记的版本时答 404，不退回默认版本 +#[tokio::test] +async fn content_with_unknown_version_answers_404_not_the_default() -> anyhow::Result<()> { + let Some(f) = Fixture::new().await? else { + return Ok(()); + }; + let (status, response) = f + .upload(f.kb, &format!("?source={}", f.folder), &[("only.txt", "x")]) + .await?; + assert_eq!(status, StatusCode::OK); + let docs = f.created_docs(&response).await?; + let doc_id = docs[0].id; + let response = f + .app + .clone() + .oneshot( + Request::get(format!("/api/v1/documents/{doc_id}/content?version=999")) + .header("Authorization", format!("Bearer {}", f.token)) + .body(Body::empty()) + .unwrap(), + ) + .await?; + assert_eq!(response.status(), StatusCode::NOT_FOUND); + f.cleanup().await +} + +/// #859：`purged_at` 上的文档答 410 Gone，字节已抹掉 +#[tokio::test] +async fn content_on_a_purged_document_answers_410() -> anyhow::Result<()> { + let Some(f) = Fixture::new().await? else { + return Ok(()); + }; + let (status, response) = f + .upload( + f.kb, + &format!("?source={}", f.folder), + &[("purge.txt", "y")], + ) + .await?; + assert_eq!(status, StatusCode::OK); + let docs = f.created_docs(&response).await?; + let doc_id = docs[0].id; + // 把这一份的真删状态标上 —— 不走 `purge` 全流程，只设列（fixture 没有依赖别处） + sqlx::query("UPDATE documents SET purged_at = now() WHERE id = $1") + .bind(doc_id) + .execute(&f.pool) + .await?; + let response = f + .app + .clone() + .oneshot( + Request::get(format!("/api/v1/documents/{doc_id}/content")) + .header("Authorization", format!("Bearer {}", f.token)) + .body(Body::empty()) + .unwrap(), + ) + .await?; + assert_eq!(response.status(), StatusCode::GONE); + f.cleanup().await +} diff --git a/crates/utopia-server/src/api/mod.rs b/crates/utopia-server/src/api/mod.rs index 38fe98d5e..5a0635e71 100644 --- a/crates/utopia-server/src/api/mod.rs +++ b/crates/utopia-server/src/api/mod.rs @@ -400,10 +400,15 @@ pub fn router(state: AppState, cfg: &AppConfig) -> Router { "/documents/{id}", get(documents_routes::detail).delete(documents_routes::delete), ) - // 撤销删除（#268）：删除是墓碑，所以有得撤 + // 撤回删除（#268）：删除是墓碑，所以有得撤 .route("/documents/{id}/restore", post(documents_routes::restore)) // 真删（#268 下半）：只对已删除的开放，库管理员 .route("/documents/{id}/purge", post(documents_routes::purge)) + // 原始字节的读路径（#859）：导出 0020 的 digests 可核验；与 `/documents/{id}` + // 同一权限闸（Viewer）；缺失/损坏的字节不假装 404 —— 直接 500 + .route("/documents/{id}/content", get(documents_routes::content)) + // 版本台账（#859）：按 document_id 列出 `document_versions` 全部已登记版本 + .route("/documents/{id}/versions", get(documents_routes::versions)) .route("/documents/{id}/extract", post(graph_routes::extract)) .route("/kbs/{id}/graph/overview", get(graph_routes::overview)) .route( diff --git a/crates/utopia-store/src/documents.rs b/crates/utopia-store/src/documents.rs index c315c6fa2..0110d5d8b 100644 --- a/crates/utopia-store/src/documents.rs +++ b/crates/utopia-store/src/documents.rs @@ -1,7 +1,7 @@ use chrono::{DateTime, Utc}; use pgvector::Vector; use sqlx::{PgPool, Postgres, Transaction}; -use utopia_core::models::{ChunkView, Document, DocumentPage}; +use utopia_core::models::{ChunkView, Document, DocumentPage, DocumentVersion}; use utopia_core::{AppError, AppResult}; use utopia_ingest::ChunkPiece; use uuid::Uuid; @@ -695,6 +695,41 @@ pub async fn update_location( Ok(()) } +/// 列出该文档的全部已记录版本，按 version 升序（旧的在前，给人类读）； +/// 路径 `/documents/{id}/versions` 的数据来源（#859）。 +/// +/// 不带版本筛选——筛选 `?version=N` 在调用方（API 层）做：找不到时 404， +/// 而这条函数总返回整张表，调用方只看 +pub async fn list_versions(pool: &PgPool, document_id: Uuid) -> AppResult> { + Ok(sqlx::query_as::<_, DocumentVersion>( + "SELECT id, document_id, version, sha256, size_bytes, ingested_at + FROM document_versions + WHERE document_id = $1 + ORDER BY version ASC", + ) + .bind(document_id) + .fetch_all(pool) + .await?) +} + +/// 取该文档的某一个指定版本；不存在时返回 `None`，由 API 层答 404 +/// （#859 的语义：没记录的版本不算「已被采用」，诚实回答没有） +pub async fn get_version( + pool: &PgPool, + document_id: Uuid, + version: i32, +) -> AppResult> { + Ok(sqlx::query_as::<_, DocumentVersion>( + "SELECT id, document_id, version, sha256, size_bytes, ingested_at + FROM document_versions + WHERE document_id = $1 AND version = $2", + ) + .bind(document_id) + .bind(version) + .fetch_optional(pool) + .await?) +} + /// 记录一个内容版本（版本号自增）。 pub async fn record_version( pool: &PgPool,