From cbd5e42f4fcfd2b81225fa46195fc7b91261331a Mon Sep 17 00:00:00 2001 From: Oleh Martsokha Date: Mon, 31 Aug 2026 19:34:59 +0200 Subject: [PATCH] Add file content-hash dedup, expose file hash, harden upload - Expose `fileHash` (hex SHA-256) on the File response so a client can identify a file's content. Add a self-validating FileHash newtype in handler/utility (deserializes a 64-char hex string, rejects bad hex/length at the extractor boundary, carries its own OpenAPI schema). - Let clients de-duplicate before uploading: the file list endpoint gains a `hash` filter (GET /workspaces/{slug}/files/?hash=) that returns the workspace's live files with that exact content. A non-empty result means the file already exists, so the upload can be skipped. Threads a content-hash facet through FileFilter and both list queries, hitting the existing (file_hash_sha256, file_size_bytes) index. - Fix upload-path connection starvation surfaced under load: upload_file no longer holds a pooled DB connection across the NATS streaming phase. It does the quick pre-flight DB work (auth, uploader, settings) under a connection, drops it, streams every file with no connection held, then re-acquires only for the final commit transaction. - Remove the RouteCategory metrics categorizer: its prefix list had drifted out of sync with the routes (so e.g. /notifications/ logged category="unknown"), and most routes are workspace-scoped and collapse to one category anyway. The request-metrics middleware keeps method, uri, status, duration, and body sizes. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_018bKk1YEG4tZ69jzYVQvQL8 --- .../src/query/workspace_file.rs | 16 +++ .../src/types/filtering/files.rs | 17 +++ crates/nvisy-server/src/handler/files.rs | 52 +++++---- .../nvisy-server/src/handler/request/files.rs | 9 ++ .../src/handler/response/files.rs | 3 + .../src/handler/utility/file_hash.rs | 107 ++++++++++++++++++ .../nvisy-server/src/handler/utility/mod.rs | 2 + crates/nvisy-server/src/middleware/mod.rs | 2 - .../src/middleware/observability.rs | 14 +-- .../src/middleware/route_category.rs | 80 ------------- 10 files changed, 189 insertions(+), 113 deletions(-) create mode 100644 crates/nvisy-server/src/handler/utility/file_hash.rs delete mode 100644 crates/nvisy-server/src/middleware/route_category.rs diff --git a/crates/nvisy-postgres/src/query/workspace_file.rs b/crates/nvisy-postgres/src/query/workspace_file.rs index 5a31ebde..c718a728 100644 --- a/crates/nvisy-postgres/src/query/workspace_file.rs +++ b/crates/nvisy-postgres/src/query/workspace_file.rs @@ -654,6 +654,11 @@ impl WorkspaceFileRepository for PgConnection { query = query.filter(dsl::file_extension.eq_any(extensions.to_vec())); } + // Apply the exact content-hash constraint (dedup lookup). + if let Some(hash) = filter.hash() { + query = query.filter(dsl::file_hash_sha256.eq(hash.to_vec())); + } + // Apply sorting let query = match (sort_by.field, sort_by.order) { (FileSortField::Name, SortOrder::Asc) => query.order(dsl::display_name.asc()), @@ -687,6 +692,7 @@ impl WorkspaceFileRepository for PgConnection { // Precompute filter values let search_term = filter.search_term().map(|s| s.to_string()); let extensions: Option> = filter.extensions().map(|e| e.to_vec()); + let hash: Option> = filter.hash().map(|h| h.to_vec()); // Build base query with filters let mut base_query = workspace_files::table @@ -711,6 +717,11 @@ impl WorkspaceFileRepository for PgConnection { base_query = base_query.filter(dsl::file_extension.eq_any(extensions)); } + // Apply the exact content-hash constraint (dedup lookup). + if let Some(ref hash) = hash { + base_query = base_query.filter(dsl::file_hash_sha256.eq(hash)); + } + let total = if pagination.include_count { Some( base_query @@ -749,6 +760,11 @@ impl WorkspaceFileRepository for PgConnection { query = query.filter(dsl::file_extension.eq_any(extensions)); } + // Apply the exact content-hash constraint (dedup lookup). + if let Some(ref hash) = hash { + query = query.filter(dsl::file_hash_sha256.eq(hash)); + } + let limit = pagination.fetch_limit(); // Apply cursor filter if present diff --git a/crates/nvisy-postgres/src/types/filtering/files.rs b/crates/nvisy-postgres/src/types/filtering/files.rs index e3503d70..8b9a67dd 100644 --- a/crates/nvisy-postgres/src/types/filtering/files.rs +++ b/crates/nvisy-postgres/src/types/filtering/files.rs @@ -14,6 +14,10 @@ pub struct FileFilter { /// only these extensions — including `Some(empty)`, which matches nothing /// (an active facet resolved to an empty set). extensions: Option>, + /// Exact SHA-256 content hash (32 raw bytes). `None` imposes no constraint; + /// `Some(hash)` matches only files with this exact content. Lets a client + /// check whether identical content already exists before uploading it. + hash: Option>, } impl FileFilter { @@ -63,4 +67,17 @@ impl FileFilter { pub fn extensions(&self) -> Option<&[String]> { self.extensions.as_deref() } + + /// Constrains to files whose content hash is exactly `hash` (32 raw bytes). + #[inline] + pub fn with_hash(mut self, hash: Vec) -> Self { + self.hash = Some(hash); + self + } + + /// Returns the content-hash constraint, if one is set. + #[inline] + pub fn hash(&self) -> Option<&[u8]> { + self.hash.as_deref() + } } diff --git a/crates/nvisy-server/src/handler/files.rs b/crates/nvisy-server/src/handler/files.rs index 826cf43d..7cef23d4 100644 --- a/crates/nvisy-server/src/handler/files.rs +++ b/crates/nvisy-server/src/handler/files.rs @@ -112,9 +112,10 @@ async fn list_files( fn list_files_docs(op: TransformOperation) -> TransformOperation { op.summary("List files") .description( - "Lists files in a workspace with cursor-based pagination. Use the `after` parameter with the `nextCursor` value from the response to fetch subsequent pages.", + "Lists files in a workspace with cursor-based pagination. Use the `after` parameter with the `nextCursor` value from the response to fetch subsequent pages. Pass `hash` (a hex SHA-256) to find files with identical content — a non-empty result means the file already exists, so an upload can be skipped.", ) .response::<200, Json>() + .response::<400, Json>() .response::<401, Json>() .response::<403, Json>() } @@ -312,26 +313,31 @@ async fn upload_file( ) -> Result<(StatusCode, Json)> { tracing::info!(target: TRACING_TARGET, "Uploading files"); - let mut conn = pg_client.get_connection().await?; + let file_store = nats_client.object_store::().await?; - auth_claims - .authorize_workspace(&mut conn, workspace.id, Permission::UploadFiles) - .await?; + // Do the quick pre-flight DB work under a connection, then release it: auth, + // resolve the uploader's identity, and read the workspace's upload settings. + // Holding a pooled connection across the streaming below would pin it for the + // whole upload and starve the pool under load, so this scope drops it before + // streaming begins. + let (uploaded_by, expires_at, max_upload_bytes) = { + let mut conn = pg_client.get_connection().await?; - let file_store = nats_client.object_store::().await?; + auth_claims + .authorize_workspace(&mut conn, workspace.id, Permission::UploadFiles) + .await?; + + let uploaded_by = resolve_account_ref(&mut conn, auth_claims.account_id).await?; - // The uploader is the caller; resolve their identity once for every file below. - let uploaded_by = resolve_account_ref(&mut conn, auth_claims.account_id).await?; + let settings = workspace.settings.or_default(); + let expires_at = settings + .retention + .original_documents + .expires_at(jiff::Timestamp::now()); + let max_upload_bytes = settings.effective_max_upload_bytes(upload.max_file_bytes()); - // Read workspace settings once for the whole batch: the retention expiry for - // uploaded originals, and the effective per-file upload cap (the workspace's - // soft cap clamped to the server-wide hard limit). - let settings = workspace.settings.or_default(); - let expires_at = settings - .retention - .original_documents - .expires_at(jiff::Timestamp::now()); - let max_upload_bytes = settings.effective_max_upload_bytes(upload.max_file_bytes()); + (uploaded_by, expires_at, max_upload_bytes) + }; let ctx = FileUploadContext { workspace_id: workspace.id, @@ -343,16 +349,20 @@ async fn upload_file( max_upload_bytes, }; - // Stream every file to storage first, then persist all their rows and events - // in one transaction. The upload is atomic: it either records every file or, - // on any failure, records none and reclaims every staged object — never a - // partial batch, and never an object left behind with no row. + // Stream every file to storage first (no DB connection held), then persist all + // their rows and events in one transaction. The upload is atomic: it either + // records every file or, on any failure, records none and reclaims every + // staged object — never a partial batch, and never an object left behind with + // no row. let staged = ctx.stage_all(&mut multipart).await?; if staged.is_empty() { return Err(ErrorKind::BadRequest.with_message("No files provided in multipart request")); } + // Re-acquire a connection only for the final commit, so the pool is free + // during the streaming above. + let mut conn = pg_client.get_connection().await?; let origin = EventOrigin { workspace_id: workspace.id, account_id: auth_claims.account_id, diff --git a/crates/nvisy-server/src/handler/request/files.rs b/crates/nvisy-server/src/handler/request/files.rs index ee523d11..764a7a96 100644 --- a/crates/nvisy-server/src/handler/request/files.rs +++ b/crates/nvisy-server/src/handler/request/files.rs @@ -12,6 +12,7 @@ use serde::{Deserialize, Serialize}; use uuid::Uuid; use validator::Validate; +use crate::handler::utility::FileHash; use crate::service::{EngineService, UnknownFormatToken}; /// Request to update file metadata. @@ -114,6 +115,11 @@ pub struct ListFiles { /// Filter by modality (`text`, `tabular`, `image`, `audio`). #[serde(skip_serializing_if = "Option::is_none")] pub modality: Option>, + /// Filter to files whose content is exactly this SHA-256. Lets a client check + /// whether identical content already exists in the workspace before uploading + /// it. + #[serde(skip_serializing_if = "Option::is_none")] + pub hash: Option, } impl ListFiles { @@ -138,6 +144,9 @@ impl ListFiles { if let Some(extensions) = intersect_facets(formats, modality) { filter = filter.with_extensions(extensions); } + if let Some(hash) = &self.hash { + filter = filter.with_hash(hash.to_bytes()); + } Ok(filter) } } diff --git a/crates/nvisy-server/src/handler/response/files.rs b/crates/nvisy-server/src/handler/response/files.rs index 40442a84..7580a3ab 100644 --- a/crates/nvisy-server/src/handler/response/files.rs +++ b/crates/nvisy-server/src/handler/response/files.rs @@ -26,6 +26,8 @@ pub struct File { pub file_extension: String, /// File size in bytes. pub file_size: i64, + /// Lowercase hex-encoded SHA-256 of the file's plaintext content. + pub file_hash: String, /// The file's role (original, redacted, audit). pub file_kind: FileKind, /// Account that uploaded/created the file. @@ -50,6 +52,7 @@ impl File { original_filename: file.original_filename, file_extension: file.file_extension, file_size: file.file_size_bytes, + file_hash: hex::encode(&file.file_hash_sha256), file_kind: file.file_kind, uploaded_by, version_number: file.version_number, diff --git a/crates/nvisy-server/src/handler/utility/file_hash.rs b/crates/nvisy-server/src/handler/utility/file_hash.rs new file mode 100644 index 00000000..1f11d404 --- /dev/null +++ b/crates/nvisy-server/src/handler/utility/file_hash.rs @@ -0,0 +1,107 @@ +//! A validated file content hash for request parameters. + +use std::borrow::Cow; +use std::fmt; +use std::str::FromStr; + +use schemars::{JsonSchema, Schema, SchemaGenerator}; +use serde::{Deserialize, Serialize}; + +/// Length of a SHA-256 digest in bytes. +const SHA256_LEN: usize = 32; + +/// A file content hash: a SHA-256 digest carried as a 64-character hex string. +/// +/// Validates on deserialization — a value that is not exactly 32 bytes of hex is +/// rejected — so a handler holding one never has to re-check it. +#[must_use] +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct FileHash([u8; SHA256_LEN]); + +impl FileHash { + /// The digest as raw bytes, for matching against the stored column. + pub fn to_bytes(&self) -> Vec { + self.0.to_vec() + } +} + +impl FromStr for FileHash { + type Err = FileHashError; + + fn from_str(s: &str) -> Result { + let mut bytes = [0u8; SHA256_LEN]; + hex::decode_to_slice(s, &mut bytes).map_err(|_| FileHashError)?; + Ok(Self(bytes)) + } +} + +impl fmt::Display for FileHash { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "{}", hex::encode(self.0)) + } +} + +impl Serialize for FileHash { + fn serialize(&self, serializer: S) -> Result { + serializer.serialize_str(&self.to_string()) + } +} + +impl<'de> Deserialize<'de> for FileHash { + fn deserialize>(deserializer: D) -> Result { + let s = String::deserialize(deserializer)?; + s.parse().map_err(serde::de::Error::custom) + } +} + +impl JsonSchema for FileHash { + fn schema_name() -> Cow<'static, str> { + "FileHash".into() + } + + fn json_schema(_generator: &mut SchemaGenerator) -> Schema { + schemars::json_schema!({ + "type": "string", + "description": "A SHA-256 content hash as a 64-character hex string.", + "pattern": "^[0-9a-fA-F]{64}$", + }) + } +} + +/// The `hash` value was not a valid 64-character hex SHA-256. +#[derive(Debug)] +pub struct FileHashError; + +impl fmt::Display for FileHashError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "hash must be a 64-character hex SHA-256") + } +} + +impl std::error::Error for FileHashError {} + +#[cfg(test)] +mod tests { + use super::FileHash; + + #[test] + fn parses_valid_sha256_and_round_trips() { + let hex = "a".repeat(64); + let hash: FileHash = hex.parse().expect("64 hex chars is a valid sha256"); + assert_eq!(hash.to_bytes().len(), 32); + assert_eq!(hash.to_string(), hex); + } + + #[test] + fn rejects_bad_length_and_non_hex() { + assert!("abcd".parse::().is_err(), "too short"); + assert!( + "a".repeat(66).parse::().is_err(), + "too long (33 bytes)" + ); + assert!( + "zz".repeat(32).parse::().is_err(), + "non-hex characters" + ); + } +} diff --git a/crates/nvisy-server/src/handler/utility/mod.rs b/crates/nvisy-server/src/handler/utility/mod.rs index b0153b42..6acc2710 100644 --- a/crates/nvisy-server/src/handler/utility/mod.rs +++ b/crates/nvisy-server/src/handler/utility/mod.rs @@ -3,9 +3,11 @@ mod accounts; mod custom_routes; mod download; +mod file_hash; mod sse_response; pub use accounts::{ActorFilter, build_password_user_inputs, resolve_account_ref, resolve_actor}; pub use custom_routes::{BuiltinModule, CustomRoutes, RouterMapFn}; pub use download::{DownloadResponseExt, attachment_headers}; +pub use file_hash::FileHash; pub use sse_response::SseResponse; diff --git a/crates/nvisy-server/src/middleware/mod.rs b/crates/nvisy-server/src/middleware/mod.rs index b286d2aa..d873e319 100644 --- a/crates/nvisy-server/src/middleware/mod.rs +++ b/crates/nvisy-server/src/middleware/mod.rs @@ -51,7 +51,6 @@ mod constants; mod counting_body; mod observability; mod recovery; -mod route_category; mod security; mod specification; mod sunset; @@ -61,7 +60,6 @@ pub use authorization::require_admin; pub use constants::{DEFAULT_MAX_BODY_SIZE, DEFAULT_MAX_FILE_BODY_SIZE}; pub use observability::RouterObservabilityExt; pub use recovery::{RecoveryConfig, RouterRecoveryExt}; -pub use route_category::RouteCategory; pub use security::{ CorsConfig, FrameOptions, ReferrerPolicy, RouterSecurityExt, SecurityHeadersConfig, UploadConfig, diff --git a/crates/nvisy-server/src/middleware/observability.rs b/crates/nvisy-server/src/middleware/observability.rs index 4a91c915..9d66ddbb 100644 --- a/crates/nvisy-server/src/middleware/observability.rs +++ b/crates/nvisy-server/src/middleware/observability.rs @@ -17,7 +17,6 @@ use tower_http::request_id::{MakeRequestUuid, PropagateRequestIdLayer, SetReques use tower_http::sensitive_headers::SetSensitiveRequestHeadersLayer; use tower_http::trace::TraceLayer; -use super::RouteCategory; use super::counting_body::CountingBody; /// Tracing target for request metrics. @@ -53,8 +52,7 @@ pub trait RouterObservabilityExt { /// Layers metrics middleware for request tracking and performance monitoring. /// - /// This middleware tracks request counts by category, response times, - /// request/response body sizes, and client IP addresses. + /// This middleware tracks response times and request/response body sizes. fn with_metrics(self) -> Self; } @@ -78,16 +76,15 @@ where } fn with_metrics(self) -> Self { - self.layer(ServiceBuilder::new().layer(from_fn(track_categorized_metrics))) + self.layer(ServiceBuilder::new().layer(from_fn(track_request_metrics))) } } -/// Request metrics middleware with categorization and timing. -pub async fn track_categorized_metrics(request: Request, next: Next) -> Response { +/// Request metrics middleware with timing and body sizes. +pub async fn track_request_metrics(request: Request, next: Next) -> Response { let start_time = Instant::now(); let method = request.method().clone(); let uri = request.uri().clone(); - let category = RouteCategory::from_uri(&uri); let request_size = request .headers() @@ -100,7 +97,6 @@ pub async fn track_categorized_metrics(request: Request, next: Next) -> Response target: TRACING_TARGET_METRICS, method = %method, uri = %uri, - category = category.as_str(), request_size = request_size, "request started" ); @@ -113,7 +109,6 @@ pub async fn track_categorized_metrics(request: Request, next: Next) -> Response // is not known here. Wrap the body to count bytes as they flow and emit the // "request completed" line when the stream ends (or is dropped) — which for // a streamed response is after this function returns. - let category_str = category.as_str(); let span = tracing::Span::current(); let (parts, body) = response.into_parts(); let body = CountingBody::new(body, move |response_size| { @@ -122,7 +117,6 @@ pub async fn track_categorized_metrics(request: Request, next: Next) -> Response target: TRACING_TARGET_METRICS, method = %method, uri = %uri, - category = category_str, status = %status, duration_ms = duration.as_millis() as u64, request_size = request_size, diff --git a/crates/nvisy-server/src/middleware/route_category.rs b/crates/nvisy-server/src/middleware/route_category.rs deleted file mode 100644 index f7b4ab6d..00000000 --- a/crates/nvisy-server/src/middleware/route_category.rs +++ /dev/null @@ -1,80 +0,0 @@ -//! Route categorization for metrics and logging. -//! -//! This module provides a categorization system for routes based on their -//! URI path, enabling aggregated metrics and monitoring by functional area. - -use axum::http::Uri; - -/// Route classification for metrics grouping. -/// -/// Categorizes routes based on their URI path for aggregated metrics -/// and monitoring purposes. Each category represents a distinct -/// functional area of the API. -#[derive(Debug, Clone, PartialEq, Eq, Hash)] -pub enum RouteCategory { - /// Authentication routes (`/auth/*`). - Authentication, - /// Account management routes (`/accounts/*`). - Accounts, - /// Workspace management routes (`/workspaces/*`). - Workspaces, - /// Document routes (`/documents/*`). - Documents, - /// File operations routes (`/files/*`). - Files, - /// Connection routes (`/connections/*`). - Connections, - /// Webhook routes (`/webhooks/*`). - Webhooks, - /// Health and monitoring routes (`/monitors/*`). - Monitors, - /// API documentation routes (`/api/*`). - Api, - /// Unknown or uncategorized routes. - Unknown, -} - -impl RouteCategory { - /// Categorizes a route based on its URI path. - pub fn from_uri(uri: &Uri) -> Self { - let path = uri.path(); - - if path.starts_with("/auth/") { - Self::Authentication - } else if path.starts_with("/accounts/") { - Self::Accounts - } else if path.starts_with("/workspaces/") { - Self::Workspaces - } else if path.starts_with("/documents/") { - Self::Documents - } else if path.starts_with("/files/") { - Self::Files - } else if path.starts_with("/connections/") { - Self::Connections - } else if path.starts_with("/webhooks/") { - Self::Webhooks - } else if path.starts_with("/monitors/") { - Self::Monitors - } else if path.starts_with("/api/") { - Self::Api - } else { - Self::Unknown - } - } - - /// Returns the string representation for logging and metrics. - pub fn as_str(&self) -> &'static str { - match self { - Self::Authentication => "auth", - Self::Accounts => "accounts", - Self::Workspaces => "workspaces", - Self::Documents => "documents", - Self::Files => "files", - Self::Connections => "connections", - Self::Webhooks => "webhooks", - Self::Monitors => "monitors", - Self::Api => "api", - Self::Unknown => "unknown", - } - } -}