diff --git a/.env.example b/.env.example index fac1903f..4e222419 100644 --- a/.env.example +++ b/.env.example @@ -8,6 +8,10 @@ REQUEST_TIMEOUT=30s SHUTDOWN_TIMEOUT=30s HEALTH_CACHE_DURATION=30s +# Request body size limits (4 MiB and 100 MiB) +MAX_BODY_BYTES=4194304 +MAX_FILE_BODY_BYTES=104857600 + # TLS (optional) TLS_CERT_PATH=./cert.pem TLS_KEY_PATH=./key.pem @@ -32,7 +36,7 @@ OPENAPI_SCALAR_PATH=/api/scalar # PostgreSQL POSTGRES_URL=postgresql://postgres:postgres@localhost:5432/postgres POSTGRES_MAX_CONNECTIONS=10 -POSTGRES_CONNECTION_TIMEOUT=30s +POSTGRES_CONNECTION_TIMEOUT=10s POSTGRES_IDLE_TIMEOUT=10m # NATS diff --git a/Cargo.lock b/Cargo.lock index 9cec5b79..021bd7bc 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2879,7 +2879,7 @@ checksum = "252afb9ae5eaa683babdc6a068b3f5726eb19e05070c731f9b2a23a7c3e8ed34" [[package]] name = "elide" version = "0.1.0" -source = "git+https://github.com/nvisycom/elide?branch=main#7da9b52534d1024602872816ce27feaad54c6a40" +source = "git+https://github.com/nvisycom/elide?branch=main#4c1d378db831441c2b6f9877fcd3f35dc75a7d89" dependencies = [ "async-trait", "elide-codec", @@ -2898,9 +2898,9 @@ dependencies = [ ] [[package]] -name = "elide-bento" +name = "elide-bentoml" version = "0.1.0" -source = "git+https://github.com/nvisycom/elide-bento?branch=main#391c8ec6c1f5ff614d32c8a9888eacdb8f546ed5" +source = "git+https://github.com/nvisycom/elide-provider?branch=main#17b3fb2282a5a5aec7dab3cf2d9ebc6d3d0ad5e7" dependencies = [ "async-trait", "base64 0.23.1", @@ -2917,7 +2917,7 @@ dependencies = [ [[package]] name = "elide-codec" version = "0.1.0" -source = "git+https://github.com/nvisycom/elide?branch=main#7da9b52534d1024602872816ce27feaad54c6a40" +source = "git+https://github.com/nvisycom/elide?branch=main#4c1d378db831441c2b6f9877fcd3f35dc75a7d89" dependencies = [ "async-trait", "bytes", @@ -2929,7 +2929,7 @@ dependencies = [ "hound", "image", "imageproc", - "quick-xml", + "quick-xml 0.42.0", "serde_json", "sha2 0.11.0", "symphonia", @@ -2939,7 +2939,7 @@ dependencies = [ [[package]] name = "elide-context" version = "0.1.0" -source = "git+https://github.com/nvisycom/elide?branch=main#7da9b52534d1024602872816ce27feaad54c6a40" +source = "git+https://github.com/nvisycom/elide?branch=main#4c1d378db831441c2b6f9877fcd3f35dc75a7d89" dependencies = [ "async-trait", "elide-core", @@ -2950,7 +2950,7 @@ dependencies = [ [[package]] name = "elide-core" version = "0.1.0" -source = "git+https://github.com/nvisycom/elide?branch=main#7da9b52534d1024602872816ce27feaad54c6a40" +source = "git+https://github.com/nvisycom/elide?branch=main#4c1d378db831441c2b6f9877fcd3f35dc75a7d89" dependencies = [ "async-trait", "blake3", @@ -2963,14 +2963,13 @@ dependencies = [ "oxilangtag", "schemars 1.2.2", "serde", - "type-map", "uuid", ] [[package]] name = "elide-detection" version = "0.1.0" -source = "git+https://github.com/nvisycom/elide?branch=main#7da9b52534d1024602872816ce27feaad54c6a40" +source = "git+https://github.com/nvisycom/elide?branch=main#4c1d378db831441c2b6f9877fcd3f35dc75a7d89" dependencies = [ "elide-core", "futures", @@ -2982,7 +2981,7 @@ dependencies = [ [[package]] name = "elide-engine" version = "0.1.0" -source = "git+https://github.com/nvisycom/elide?branch=main#7da9b52534d1024602872816ce27feaad54c6a40" +source = "git+https://github.com/nvisycom/elide?branch=main#4c1d378db831441c2b6f9877fcd3f35dc75a7d89" dependencies = [ "bytes", "elide-codec", @@ -3000,7 +2999,7 @@ dependencies = [ [[package]] name = "elide-export" version = "0.1.0" -source = "git+https://github.com/nvisycom/elide-runtime?branch=main#c50e46715bda0dcab13910bcac6f480ed1426e40" +source = "git+https://github.com/nvisycom/elide-runtime?branch=main#fd4072388e1dd5b195862d7641eeb8d770146cfa" dependencies = [ "csv", "elide", @@ -3012,7 +3011,7 @@ dependencies = [ [[package]] name = "elide-fake" version = "0.1.0" -source = "git+https://github.com/nvisycom/elide?branch=main#7da9b52534d1024602872816ce27feaad54c6a40" +source = "git+https://github.com/nvisycom/elide?branch=main#4c1d378db831441c2b6f9877fcd3f35dc75a7d89" dependencies = [ "async-trait", "elide-core", @@ -3023,7 +3022,7 @@ dependencies = [ [[package]] name = "elide-governance" version = "0.1.0" -source = "git+https://github.com/nvisycom/elide-runtime?branch=main#c50e46715bda0dcab13910bcac6f480ed1426e40" +source = "git+https://github.com/nvisycom/elide-runtime?branch=main#fd4072388e1dd5b195862d7641eeb8d770146cfa" dependencies = [ "elide-core", "elide-operator", @@ -3037,7 +3036,7 @@ dependencies = [ [[package]] name = "elide-lingua" version = "0.1.0" -source = "git+https://github.com/nvisycom/elide?branch=main#7da9b52534d1024602872816ce27feaad54c6a40" +source = "git+https://github.com/nvisycom/elide?branch=main#4c1d378db831441c2b6f9877fcd3f35dc75a7d89" dependencies = [ "async-trait", "elide-core", @@ -3048,7 +3047,7 @@ dependencies = [ [[package]] name = "elide-llm" version = "0.1.0" -source = "git+https://github.com/nvisycom/elide?branch=main#7da9b52534d1024602872816ce27feaad54c6a40" +source = "git+https://github.com/nvisycom/elide?branch=main#4c1d378db831441c2b6f9877fcd3f35dc75a7d89" dependencies = [ "async-trait", "derive_builder", @@ -3069,7 +3068,7 @@ dependencies = [ [[package]] name = "elide-ner" version = "0.1.0" -source = "git+https://github.com/nvisycom/elide?branch=main#7da9b52534d1024602872816ce27feaad54c6a40" +source = "git+https://github.com/nvisycom/elide?branch=main#4c1d378db831441c2b6f9877fcd3f35dc75a7d89" dependencies = [ "async-trait", "derive_builder", @@ -3083,7 +3082,7 @@ dependencies = [ [[package]] name = "elide-ocr" version = "0.1.0" -source = "git+https://github.com/nvisycom/elide?branch=main#7da9b52534d1024602872816ce27feaad54c6a40" +source = "git+https://github.com/nvisycom/elide?branch=main#4c1d378db831441c2b6f9877fcd3f35dc75a7d89" dependencies = [ "async-trait", "derive_builder", @@ -3095,11 +3094,11 @@ dependencies = [ [[package]] name = "elide-office" version = "0.1.0" -source = "git+https://github.com/nvisycom/elide?branch=main#7da9b52534d1024602872816ce27feaad54c6a40" +source = "git+https://github.com/nvisycom/elide?branch=main#4c1d378db831441c2b6f9877fcd3f35dc75a7d89" dependencies = [ "bytes", "hipstr", - "quick-xml", + "quick-xml 0.42.0", "thiserror", "zip", ] @@ -3107,7 +3106,7 @@ dependencies = [ [[package]] name = "elide-operator" version = "0.1.0" -source = "git+https://github.com/nvisycom/elide?branch=main#7da9b52534d1024602872816ce27feaad54c6a40" +source = "git+https://github.com/nvisycom/elide?branch=main#4c1d378db831441c2b6f9877fcd3f35dc75a7d89" dependencies = [ "aes-gcm", "async-trait", @@ -3129,7 +3128,7 @@ dependencies = [ [[package]] name = "elide-pattern" version = "0.1.0" -source = "git+https://github.com/nvisycom/elide?branch=main#7da9b52534d1024602872816ce27feaad54c6a40" +source = "git+https://github.com/nvisycom/elide?branch=main#4c1d378db831441c2b6f9877fcd3f35dc75a7d89" dependencies = [ "aho-corasick", "async-trait", @@ -3149,7 +3148,7 @@ dependencies = [ [[package]] name = "elide-pdf" version = "0.1.0" -source = "git+https://github.com/nvisycom/elide?branch=main#7da9b52534d1024602872816ce27feaad54c6a40" +source = "git+https://github.com/nvisycom/elide?branch=main#4c1d378db831441c2b6f9877fcd3f35dc75a7d89" dependencies = [ "bytes", "hipstr", @@ -3161,7 +3160,7 @@ dependencies = [ [[package]] name = "elide-pipeline" version = "0.1.0" -source = "git+https://github.com/nvisycom/elide-runtime?branch=main#c50e46715bda0dcab13910bcac6f480ed1426e40" +source = "git+https://github.com/nvisycom/elide-runtime?branch=main#fd4072388e1dd5b195862d7641eeb8d770146cfa" dependencies = [ "bytes", "elide", @@ -3182,10 +3181,10 @@ dependencies = [ [[package]] name = "elide-provider" version = "0.1.0" -source = "git+https://github.com/nvisycom/elide-runtime?branch=main#c50e46715bda0dcab13910bcac6f480ed1426e40" +source = "git+https://github.com/nvisycom/elide-runtime?branch=main#fd4072388e1dd5b195862d7641eeb8d770146cfa" dependencies = [ "elide", - "elide-bento", + "elide-bentoml", "elide-governance", "hipstr", "schemars 1.2.2", @@ -3196,7 +3195,7 @@ dependencies = [ [[package]] name = "elide-redaction" version = "0.1.0" -source = "git+https://github.com/nvisycom/elide?branch=main#7da9b52534d1024602872816ce27feaad54c6a40" +source = "git+https://github.com/nvisycom/elide?branch=main#4c1d378db831441c2b6f9877fcd3f35dc75a7d89" dependencies = [ "elide-core", "hipstr", @@ -3206,7 +3205,7 @@ dependencies = [ [[package]] name = "elide-review" version = "0.1.0" -source = "git+https://github.com/nvisycom/elide-runtime?branch=main#c50e46715bda0dcab13910bcac6f480ed1426e40" +source = "git+https://github.com/nvisycom/elide-runtime?branch=main#fd4072388e1dd5b195862d7641eeb8d770146cfa" dependencies = [ "elide", "elide-governance", @@ -3218,7 +3217,7 @@ dependencies = [ [[package]] name = "elide-stt" version = "0.1.0" -source = "git+https://github.com/nvisycom/elide?branch=main#7da9b52534d1024602872816ce27feaad54c6a40" +source = "git+https://github.com/nvisycom/elide?branch=main#4c1d378db831441c2b6f9877fcd3f35dc75a7d89" dependencies = [ "async-trait", "derive_builder", @@ -3230,7 +3229,7 @@ dependencies = [ [[package]] name = "elide-template" version = "0.1.0" -source = "git+https://github.com/nvisycom/elide-runtime?branch=main#c50e46715bda0dcab13910bcac6f480ed1426e40" +source = "git+https://github.com/nvisycom/elide-runtime?branch=main#fd4072388e1dd5b195862d7641eeb8d770146cfa" dependencies = [ "elide-core", "elide-governance", @@ -6386,7 +6385,7 @@ dependencies = [ "nix", "parking_lot", "percent-encoding", - "quick-xml", + "quick-xml 0.41.0", "rand 0.10.2", "reqwest 0.13.4", "rustls-pki-types", @@ -6718,7 +6717,7 @@ dependencies = [ "nom 7.1.3", "once_cell", "postcard", - "quick-xml", + "quick-xml 0.41.0", "regex", "regex-cache", "serde", @@ -7105,6 +7104,15 @@ dependencies = [ "serde", ] +[[package]] +name = "quick-xml" +version = "0.42.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "41b1177fdf999d2321d3fb46ff47159d9c1fb9ad66a4879f8c50a0b504615e9b" +dependencies = [ + "memchr", +] + [[package]] name = "quinn" version = "0.11.11" @@ -9481,15 +9489,6 @@ dependencies = [ "rand 0.10.2", ] -[[package]] -name = "type-map" -version = "0.5.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "cb30dbbd9036155e74adad6812e9898d03ec374946234fbcebd5dfc7b9187b90" -dependencies = [ - "rustc-hash", -] - [[package]] name = "typed-path" version = "0.12.3" diff --git a/crates/nvisy-postgres/src/model/workspace_detection.rs b/crates/nvisy-postgres/src/model/workspace_detection.rs index 66b51d39..ce829f4a 100644 --- a/crates/nvisy-postgres/src/model/workspace_detection.rs +++ b/crates/nvisy-postgres/src/model/workspace_detection.rs @@ -28,6 +28,10 @@ pub struct WorkspaceDetection { /// Audit file (`file_kind = audit`) holding the encrypted analysis. `None` /// until analysis writes it. pub audit_file_id: Option, + /// Intermediates file (`file_kind = intermediate`) holding the encrypted + /// enrichment (OCR layout, transcript, tokenized text). `None` until analysis + /// writes it, and stays `None` when the analysis ran no enricher. + pub intermediates_file_id: Option, /// How the detection was initiated. pub trigger_type: PipelineTriggerType, /// Current detection status. @@ -59,6 +63,9 @@ pub struct NewWorkspaceDetection { pub input_file_id: Uuid, /// Audit file holding the encrypted analysis (set once analyzed). pub audit_file_id: Option, + /// Intermediates file holding the encrypted enrichment (set once analyzed, if + /// the document produced any). + pub intermediates_file_id: Option, /// Trigger type. pub trigger_type: Option, /// Initial status. @@ -78,6 +85,8 @@ pub struct UpdateWorkspaceDetection { pub status: Option, /// Audit file holding the encrypted analysis. pub audit_file_id: Option>, + /// Intermediates file holding the encrypted enrichment. + pub intermediates_file_id: Option>, /// Non-encrypted metadata for filtering/display. pub metadata: Option>, /// When a worker last claimed this detection (lease timestamp). diff --git a/crates/nvisy-postgres/src/schema.rs b/crates/nvisy-postgres/src/schema.rs index ae81a1c8..76d1babd 100644 --- a/crates/nvisy-postgres/src/schema.rs +++ b/crates/nvisy-postgres/src/schema.rs @@ -289,6 +289,7 @@ diesel::table! { account_id -> Uuid, input_file_id -> Uuid, audit_file_id -> Nullable, + intermediates_file_id -> Nullable, trigger_type -> PipelineTriggerType, status -> DetectionStatus, idempotency_key -> Nullable, diff --git a/crates/nvisy-postgres/src/types/enums/file_kind.rs b/crates/nvisy-postgres/src/types/enums/file_kind.rs index 0a86dd9e..43ac893f 100644 --- a/crates/nvisy-postgres/src/types/enums/file_kind.rs +++ b/crates/nvisy-postgres/src/types/enums/file_kind.rs @@ -36,6 +36,14 @@ pub enum FileKind { #[db_rename = "review"] #[serde(rename = "review")] Review, + + /// Enrichment content a detection extracted from a non-text document — an + /// image's OCR layout, an audio clip's transcript — served to the client so a + /// reviewer can search it and add entities the analysis missed. Carries + /// document content, not shown in file lists. + #[db_rename = "intermediate"] + #[serde(rename = "intermediate")] + Intermediate, } impl FileKind { diff --git a/crates/nvisy-postgres/src/types/json/pipeline_metadata.rs b/crates/nvisy-postgres/src/types/json/pipeline_metadata.rs index 535412b1..c1a26cad 100644 --- a/crates/nvisy-postgres/src/types/json/pipeline_metadata.rs +++ b/crates/nvisy-postgres/src/types/json/pipeline_metadata.rs @@ -36,6 +36,8 @@ pub struct RetentionOverride { pub redacted_documents: Option, /// Overrides audit-blob retention when set. pub audit_logs: Option, + /// Overrides enrichment-intermediate retention when set. + pub intermediates: Option, } impl RetentionOverride { @@ -47,6 +49,7 @@ impl RetentionOverride { RetentionScope::OriginalDocuments => None, RetentionScope::RedactedDocuments => self.redacted_documents, RetentionScope::AuditLogs => self.audit_logs, + RetentionScope::Intermediates => self.intermediates, } } } diff --git a/crates/nvisy-postgres/src/types/json/retention.rs b/crates/nvisy-postgres/src/types/json/retention.rs index f7d40783..6fde66d3 100644 --- a/crates/nvisy-postgres/src/types/json/retention.rs +++ b/crates/nvisy-postgres/src/types/json/retention.rs @@ -62,6 +62,8 @@ pub enum RetentionScope { RedactedDocuments, /// Engine analysis (audit) blobs. AuditLogs, + /// Enrichment content extracted from a document (OCR layout, transcript). + Intermediates, } /// Retention for every scope. Missing fields default to [`Retention::Forever`], @@ -77,6 +79,8 @@ pub struct RetentionSettings { pub redacted_documents: Retention, /// Retention for audit blobs. pub audit_logs: Retention, + /// Retention for enrichment intermediates (OCR layout, transcript). + pub intermediates: Retention, } impl RetentionSettings { @@ -87,6 +91,7 @@ impl RetentionSettings { RetentionScope::OriginalDocuments => self.original_documents, RetentionScope::RedactedDocuments => self.redacted_documents, RetentionScope::AuditLogs => self.audit_logs, + RetentionScope::Intermediates => self.intermediates, } } diff --git a/crates/nvisy-server/src/handler/detection_audits.rs b/crates/nvisy-server/src/handler/detection_audits.rs index 43962dba..7f69c001 100644 --- a/crates/nvisy-server/src/handler/detection_audits.rs +++ b/crates/nvisy-server/src/handler/detection_audits.rs @@ -1,8 +1,11 @@ -//! Detection audit handlers: read and export a detection's analysis. +//! Detection audit handlers: read and export a detection's analysis, and read +//! the enrichment intermediates it extracted. //! //! Once a detection is complete, its `Audit` (the decrypted map of detected -//! findings) can be reviewed inline or downloaded as JSON or a zip of CSV tables. -//! The detection lifecycle itself (create, list, redact) lives in +//! findings) can be reviewed inline or downloaded as JSON or a zip of CSV tables, +//! and the enrichment intermediates (an image's OCR layout, an audio clip's +//! transcript) can be read for client-side search and entity addition. The +//! detection lifecycle itself (create, list, redact) lives in //! [`detections`](super::detections). use aide::axum::ApiRouter; @@ -11,8 +14,8 @@ use aide::transform::TransformOperation; use axum::body::Body; use axum::extract::State; use axum::http::{HeaderMap, HeaderValue, StatusCode}; -use elide_pipeline::Audit; use elide_pipeline::export::{ExportCsv, ExportJson}; +use elide_pipeline::{ArtifactSet, Audit}; use nvisy_postgres::PgClient; use super::detections::find_detection; @@ -84,6 +87,71 @@ fn get_detection_analysis_docs(op: TransformOperation) -> TransformOperation { .response::<409, Json>() } +/// Returns the detection's enrichment intermediates — an image's OCR layout, an +/// audio clip's transcript, tokenized text — as the content the analysis +/// extracted, so a client can search it and add entities the analysis missed. +/// +/// Available only for a detection whose analysis enriched (ran an enricher for a +/// group); a detection with no enrichment has none — 404. Requires +/// `ViewPipelines`. +#[tracing::instrument( + skip_all, + fields( + account_id = %auth_state.account_id, + workspace_id = %workspace.id, + detection_id = %path_params.detection_id, + ) +)] +async fn get_detection_intermediates( + State(pg_client): State, + State(blob): State, + State(engine): State, + AuthState(auth_state): AuthState, + WorkspaceContext(workspace): WorkspaceContext, + Path(path_params): Path, +) -> Result<(StatusCode, Json)> { + tracing::debug!(target: TRACING_TARGET, "Getting detection intermediates"); + + // Resolve the detection and its intermediates file row under a scoped + // connection, then release it before the object-store load so the pooled + // connection is not held across the NATS round-trip. + let intermediates_file = { + let mut conn = pg_client.get_connection().await?; + + auth_state + .authorize_workspace(&mut conn, workspace.id, Permission::ViewPipelines) + .await?; + + let (detection, _pipeline) = + find_detection(&mut conn, workspace.id, path_params.detection_id.as_uuid()).await?; + + blob.resolve_intermediates_file(&mut conn, workspace.id, &detection) + .await? + }; + + let intermediates = blob + .load_intermediates(&engine, workspace.id, &intermediates_file) + .await?; + + tracing::debug!(target: TRACING_TARGET, "Detection intermediates retrieved"); + + Ok((StatusCode::OK, Json(intermediates))) +} + +fn get_detection_intermediates_docs(op: TransformOperation) -> TransformOperation { + op.summary("Get detection intermediates") + .description( + "Returns the detection's enrichment intermediates — an image's OCR layout, an audio \ + clip's transcript, or tokenized text — as `{ body, parts }`, so a client can search \ + the extracted content and add entities the analysis missed. A detection whose \ + analysis ran no enricher has no intermediates (404).", + ) + .response::<200, Json>() + .response::<401, Json>() + .response::<403, Json>() + .response::<404, Json>() +} + /// Downloads a detection's audit as a file, in the requested `format`. /// /// `json` yields a pretty-printed JSON file with the full structure — body, @@ -208,5 +276,12 @@ pub fn routes() -> ApiRouter { "/workspaces/{workspaceSlug}/detections/{detectionId}/audit/", get_with(download_detection_audit, download_detection_audit_docs), ) + .api_route( + "/workspaces/{workspaceSlug}/detections/{detectionId}/intermediates/", + get_with( + get_detection_intermediates, + get_detection_intermediates_docs, + ), + ) .with_path_items(|item| item.tag("Detections")) } diff --git a/crates/nvisy-server/src/handler/pipelines.rs b/crates/nvisy-server/src/handler/pipelines.rs index 5c7d26a7..42eccb0e 100644 --- a/crates/nvisy-server/src/handler/pipelines.rs +++ b/crates/nvisy-server/src/handler/pipelines.rs @@ -274,6 +274,12 @@ async fn update_pipeline( (FileKind::Audit, audit_logs_expiry), // Review audits share the audit-logs scope with detection audits. (FileKind::Review, audit_logs_expiry), + ( + FileKind::Intermediate, + workspace_retention + .resolve(RetentionScope::Intermediates, Some(&over)) + .expires_at(now), + ), ] }); diff --git a/crates/nvisy-server/src/handler/workspaces.rs b/crates/nvisy-server/src/handler/workspaces.rs index 7162f7c2..ceb430e2 100644 --- a/crates/nvisy-server/src/handler/workspaces.rs +++ b/crates/nvisy-server/src/handler/workspaces.rs @@ -55,6 +55,7 @@ async fn backfill_retention( // Review audits share the audit-logs scope with detection audits; a // redaction stages them under `AuditLogs`, so they backfill under it too. (RetentionScope::AuditLogs, FileKind::Review), + (RetentionScope::Intermediates, FileKind::Intermediate), ] { let expires_at = retention.get(scope).expires_at(now); conn.backfill_files_expiry(workspace_id, kind, expires_at) diff --git a/crates/nvisy-server/src/service/detection/worker.rs b/crates/nvisy-server/src/service/detection/worker.rs index 0781b3cf..d4318d50 100644 --- a/crates/nvisy-server/src/service/detection/worker.rs +++ b/crates/nvisy-server/src/service/detection/worker.rs @@ -19,6 +19,7 @@ use nvisy_postgres::query::{ use nvisy_postgres::types::{DetectionStatus, Json, RasterPolicy, WorkspaceSettings}; use nvisy_postgres::{AsyncConnection, DieselError, PgError}; use tokio_util::sync::CancellationToken; +use uuid::Uuid; use super::job::DetectionJob; use super::service::DetectionQueue; @@ -252,19 +253,53 @@ impl DetectionWorker { JobOutcome::Done } - /// Best-effort reclaim of a staged audit object whose file row did not - /// commit. A failure only defers cleanup, so it is logged, never propagated. - async fn discard_staged_audit(&self, staged: &nvisy_postgres::model::NewWorkspaceFile) { + /// Best-effort reclaim of a staged object whose file row did not commit. A + /// failure only defers cleanup, so it is logged, never propagated. + async fn discard_staged(&self, staged: &nvisy_postgres::model::NewWorkspaceFile) { if let Err(err) = self.blob.discard_staged_object(staged).await { tracing::warn!( target: TRACING_TARGET, error = %err, storage_path = %staged.storage_path, - "Failed to reclaim orphaned audit object; left for a later sweep", + "Failed to reclaim orphaned staged object; left for a later sweep", ); } } + /// Stages a detection's enrichment intermediates, or returns `None` when the + /// analysis ran no enricher for any group (its artifact set serializes empty). + /// Skipping the empty case avoids an intermediates file that a client would + /// fetch only to find nothing. + async fn stage_intermediates( + &self, + pipeline: &WorkspacePipeline, + settings: &WorkspaceSettings, + account_id: Uuid, + artifacts: &T, + ) -> Result> { + // The set serializes to `{ body, parts }`; an un-enriched document has a + // null body and no parts, and there is nothing worth persisting. + let value = serde_json::to_value(artifacts).map_err(|err| { + ErrorKind::InternalServerError + .with_message("Failed to serialize intermediates") + .with_context(err.to_string()) + })?; + let has_body = value.get("body").is_some_and(|body| !body.is_null()); + let has_parts = value + .get("parts") + .and_then(|parts| parts.as_object()) + .is_some_and(|parts| !parts.is_empty()); + if !has_body && !has_parts { + return Ok(None); + } + + let file = self + .blob + .stage_intermediates(pipeline, &settings.retention, account_id, artifacts) + .await?; + Ok(Some(file)) + } + /// Performs the analysis and records the detection as `Complete`. /// /// Manages its own connection lifecycle in three phases so a pooled @@ -326,17 +361,27 @@ impl DetectionWorker { // staging — runs with no DB connection held. let document = self.blob.build_document(&file, detection.id).await?; let analyzed = self.engine.analyze(document, &policies, &request).await?; + let audit = &analyzed.audit; // Write the (non-transactional) audit object first, then commit its file // row together with the detection's usage and status in one transaction // below. let audit_file = self .blob - .stage_analyzed_document( + .stage_analyzed_document(pipeline, &settings.retention, detection.account_id, audit) + .await?; + + // Stage the enrichment intermediates (OCR layout, transcript, tokenized + // text) beside the audit, so the client can read them and add entities the + // analysis missed. An analysis that ran no enricher produces an empty + // artifact set — nothing is stored and the detection carries no + // intermediates reference. + let intermediates_file = self + .stage_intermediates( pipeline, - &settings.retention, + &settings, detection.account_id, - &analyzed, + &analyzed.artifacts, ) .await?; @@ -345,7 +390,7 @@ impl DetectionWorker { // metadata for drill-down. Absent for a purely deterministic detection. // The report is layered onto the detection's existing metadata so tags and // any recorded error survive the write. - let usage = extract_detection_usage(detection.id, &analyzed); + let usage = extract_detection_usage(detection.id, audit); let metadata = usage.as_ref().map(|u| { let mut current = detection.metadata.or_default(); current.usage = Some(u.report.clone()); @@ -385,10 +430,18 @@ impl DetectionWorker { // Phase 3: re-acquire a connection only for the fenced finalize // transaction, so the pool was free during the analysis above. let mut conn = self.infra.postgres.get_connection().await?; + // Kept to reclaim the staged objects if the transaction does not commit: + // on rollback their `workspace_files` rows never land, so the row-driven + // reaper could never find the objects otherwise. let staged_audit = audit_file.clone(); + let staged_intermediates = intermediates_file.clone(); let finalized = conn .transaction(async |conn| { let audit_file_id = conn.create_workspace_file(audit_file).await?.id; + let intermediates_file_id = match intermediates_file { + Some(file) => Some(conn.create_workspace_file(file).await?.id), + None => None, + }; if let Some(usage) = &usage { conn.record_detection_usage(&usage.per_model).await?; } @@ -398,15 +451,16 @@ impl DetectionWorker { claim_token, UpdateWorkspaceDetection { audit_file_id: Some(Some(audit_file_id)), + intermediates_file_id: Some(intermediates_file_id), metadata, ..Default::default() }, ) .await?; if !finalized { - // Abort the audit-file and usage inserts: the detection is no - // longer ours to finalize. `RollbackTransaction` unwinds the - // writes without being a real error; it is matched below. + // Abort the file and usage inserts: the detection is no longer + // ours to finalize. `RollbackTransaction` unwinds the writes + // without being a real error; it is matched below. return Err(PgError::Query(DieselError::RollbackTransaction)); } conn.insert_event_outbox(outbox_row).await?; @@ -417,14 +471,20 @@ impl DetectionWorker { match finalized { Ok(()) => {} Err(PgError::Query(DieselError::RollbackTransaction)) => { - self.discard_staged_audit(&staged_audit).await; + self.discard_staged(&staged_audit).await; + if let Some(staged) = &staged_intermediates { + self.discard_staged(staged).await; + } tracing::warn!(target: TRACING_TARGET, detection_id = %detection.id, "Claim went stale before finalize; another worker owns the detection"); return Ok(()); } Err(err) => { - // The transaction rolled back, so the audit row never committed; - // reclaim its object before surfacing the failure. - self.discard_staged_audit(&staged_audit).await; + // The transaction rolled back, so the file rows never committed; + // reclaim their objects before surfacing the failure. + self.discard_staged(&staged_audit).await; + if let Some(staged) = &staged_intermediates { + self.discard_staged(staged).await; + } return Err(err.into()); } } diff --git a/crates/nvisy-server/src/service/run_blob_store.rs b/crates/nvisy-server/src/service/run_blob_store.rs index 45c386e2..4172adb0 100644 --- a/crates/nvisy-server/src/service/run_blob_store.rs +++ b/crates/nvisy-server/src/service/run_blob_store.rs @@ -12,7 +12,7 @@ use std::str::FromStr; use bytes::Bytes; use elide_pipeline::file::Document; -use elide_pipeline::{Audit, Engine}; +use elide_pipeline::{ArtifactSet, Audit, Engine}; use nvisy_nats::object::{AuditBucket, AuditKey, FileKey, FilesBucket, ObjectBucket}; use nvisy_postgres::PgConn; use nvisy_postgres::model::{ @@ -20,6 +20,7 @@ use nvisy_postgres::model::{ }; use nvisy_postgres::query::WorkspaceFileRepository; use nvisy_postgres::types::{FileKind, RetentionScope, RetentionSettings}; +use serde::Serialize; use sha2::{Digest, Sha256}; use tokio::io::AsyncReadExt; use uuid::Uuid; @@ -241,6 +242,60 @@ impl RunBlobStore { }) } + /// Serializes a detection's enrichment intermediates (OCR layout, transcript), + /// encrypts them with the workspace key, writes them to the audit bucket, and + /// builds the `intermediate`-kind [`WorkspaceFile`] row that will point at it — + /// without inserting the row (staged like the analysis, reclaimed on rollback). + /// + /// The intermediates carry document content, so they are encrypted at rest and + /// governed by their own retention scope, resolved here (workspace baseline, + /// pipeline override if set). + pub async fn stage_intermediates( + &self, + pipeline: &WorkspacePipeline, + workspace_settings: &RetentionSettings, + account_id: Uuid, + artifacts: &T, + ) -> Result { + let workspace_id = pipeline.workspace_id; + let plaintext = serde_json::to_vec(artifacts).map_err(analysis_serde_error)?; + let hash = Sha256::digest(&plaintext).to_vec(); + let size = plaintext.len() as i64; + let ciphertext = self + .infra + .crypto + .encrypt(workspace_id, &plaintext) + .map_err(|err| { + ErrorKind::InternalServerError + .with_message("Failed to encrypt intermediates") + .with_context(err.to_string()) + })?; + + let store = self.infra.nats.object_store::().await?; + let key = AuditKey::generate(workspace_id); + store.put(&key, Cursor::new(ciphertext)).await?; + + let over = pipeline.metadata.or_default().retention; + let expires_at = workspace_settings + .resolve(RetentionScope::Intermediates, over.as_ref()) + .expires_at(jiff::Timestamp::now()); + + Ok(NewWorkspaceFile { + workspace_id, + account_id, + display_name: Some("analysis.intermediates".to_owned()), + original_filename: Some("analysis.intermediates".to_owned()), + file_extension: Some("json".to_owned()), + file_kind: Some(FileKind::Intermediate), + file_size_bytes: size, + file_hash_sha256: hash, + storage_path: key.to_string(), + storage_bucket: store.bucket().to_owned(), + expires_at: expires_at.map(Into::into), + ..Default::default() + }) + } + /// Deletes a staged object whose file row was never committed. /// /// The `stage_*` methods write an object before its `workspace_files` row; if @@ -335,6 +390,83 @@ impl RunBlobStore { }) } + /// Resolves the `workspace_files` row holding a detection's enrichment + /// intermediates. + /// + /// The connection-bound step; pair with [`load_intermediates`](Self::load_intermediates) + /// to release the connection before the object-store round-trip. A detection + /// whose modality produced no enrichment (text, tabular) has none — a `None` + /// reference maps to a 404, distinct from a reference to a since-deleted file. + pub async fn resolve_intermediates_file( + &self, + conn: &mut PgConn, + workspace_id: Uuid, + detection: &WorkspaceDetection, + ) -> Result { + let file_id = detection.intermediates_file_id.ok_or_else(|| { + ErrorKind::NotFound + .with_message("Detection has no enrichment intermediates") + .with_resource("detection") + })?; + conn.find_file_in_workspace(workspace_id, file_id) + .await? + .ok_or_else(|| { + ErrorKind::NotFound + .with_message("The intermediates for this detection have been deleted") + .with_resource("detection") + }) + } + + /// Loads and decodes a detection's enrichment intermediates from its + /// already-resolved file row. Holds no database connection: only object-store + /// I/O and decryption. + /// + /// The `engine` reconstructs the [`ArtifactSet`] from its serialized form: it + /// serializes but does not `Deserialize`, since each group is tagged by + /// modality name and only the engine's registry can map those back to concrete + /// artifact types. + pub async fn load_intermediates( + &self, + engine: &Engine, + workspace_id: Uuid, + intermediates_file: &WorkspaceFile, + ) -> Result { + let key = AuditKey::from_str(&intermediates_file.storage_path).map_err(|err| { + ErrorKind::InternalServerError + .with_message("Invalid intermediates storage key") + .with_context(err.to_string()) + })?; + + let store = self.infra.nats.object_store::().await?; + let data = store.get(&key).await?.ok_or_else(|| { + ErrorKind::InternalServerError.with_message("Intermediates are missing from storage") + })?; + let mut reader = data.into_reader(); + let mut ciphertext = Vec::new(); + reader.read_to_end(&mut ciphertext).await.map_err(|err| { + ErrorKind::InternalServerError + .with_message("Failed to read intermediates") + .with_context(err.to_string()) + })?; + + let plaintext = self + .infra + .crypto + .decrypt(workspace_id, &ciphertext) + .map_err(|err| { + ErrorKind::InternalServerError + .with_message("Failed to decrypt intermediates") + .with_context(err.to_string()) + })?; + engine + .deserialize_artifacts(&mut serde_json::Deserializer::from_slice(&plaintext)) + .map_err(|err| { + ErrorKind::InternalServerError + .with_message("Failed to decode intermediates") + .with_context(err.to_string()) + }) + } + /// Encrypts a redaction's review audit, writes it to the audit bucket, and /// builds the `review`-kind [`WorkspaceFile`] row that will point at it — /// without inserting the row. diff --git a/deny.toml b/deny.toml index fe36194b..7f2e553b 100644 --- a/deny.toml +++ b/deny.toml @@ -97,5 +97,5 @@ allow-registry = ["https://github.com/rust-lang/crates.io-index"] allow-git = [ "https://github.com/nvisycom/elide-runtime", "https://github.com/nvisycom/elide", - "https://github.com/nvisycom/elide-bento", + "https://github.com/nvisycom/elide-provider", ] diff --git a/migrations/2025-05-27-011852_files/up.sql b/migrations/2025-05-27-011852_files/up.sql index 1c612eef..83883c98 100644 --- a/migrations/2025-05-27-011852_files/up.sql +++ b/migrations/2025-05-27-011852_files/up.sql @@ -7,10 +7,11 @@ CREATE TYPE FILE_KIND AS ENUM ( 'original', -- Source document (uploaded or imported) 'redacted', -- Redacted output produced by a redaction 'audit', -- Engine detection analysis blob (not shown in file lists) - 'review' -- Engine analysis after reviewer edits + redaction (not shown in file lists) + 'review', -- Engine analysis after reviewer edits + redaction (not shown in file lists) + 'intermediate' -- Enrichment extracted from a document: OCR layout, transcript (not shown in file lists) ); -COMMENT ON TYPE FILE_KIND IS 'The role of a file: original document, redacted output, detection audit, or review (redaction) audit.'; +COMMENT ON TYPE FILE_KIND IS 'The role of a file: original document, redacted output, detection audit, review (redaction) audit, or enrichment intermediate.'; -- Workspace files table: one stored document, with version tracking and dedup. CREATE TABLE workspace_files ( diff --git a/migrations/2026-01-19-045016_detections/up.sql b/migrations/2026-01-19-045016_detections/up.sql index 3285d591..62392012 100644 --- a/migrations/2026-01-19-045016_detections/up.sql +++ b/migrations/2026-01-19-045016_detections/up.sql @@ -47,6 +47,7 @@ CREATE TABLE workspace_detections ( -- (redundant here, but the correct action for a produced artifact). input_file_id UUID NOT NULL REFERENCES workspace_files (id) ON DELETE CASCADE, audit_file_id UUID DEFAULT NULL REFERENCES workspace_files (id) ON DELETE SET NULL, + intermediates_file_id UUID DEFAULT NULL REFERENCES workspace_files (id) ON DELETE SET NULL, -- Detection attributes trigger_type PIPELINE_TRIGGER_TYPE NOT NULL DEFAULT 'user', @@ -100,6 +101,12 @@ CREATE INDEX workspace_detections_audit_file_idx ON workspace_detections (audit_file_id) WHERE audit_file_id IS NOT NULL; +-- Same shape for the enrichment intermediates file, NULL until (and unless) a +-- detection's analysis runs an enricher (an analysis with none produces no file). +CREATE INDEX workspace_detections_intermediates_file_idx + ON workspace_detections (intermediates_file_id) + WHERE intermediates_file_id IS NOT NULL; + -- Idempotent detect: at most one detection per (pipeline, idempotency key). CREATE UNIQUE INDEX workspace_detections_idempotency_idx ON workspace_detections (pipeline_id, idempotency_key) @@ -118,6 +125,7 @@ COMMENT ON COLUMN workspace_detections.pipeline_id IS 'Pipeline whose config dro COMMENT ON COLUMN workspace_detections.account_id IS 'Account that triggered the detection'; COMMENT ON COLUMN workspace_detections.input_file_id IS 'Source document the detection analyzes'; COMMENT ON COLUMN workspace_detections.audit_file_id IS 'Audit file (file_kind=audit) holding the analysis between detect and redact'; +COMMENT ON COLUMN workspace_detections.intermediates_file_id IS 'Intermediates file (file_kind=intermediate) holding the enrichment (OCR layout, transcript) served to the client'; COMMENT ON COLUMN workspace_detections.trigger_type IS 'How the detection was initiated'; COMMENT ON COLUMN workspace_detections.status IS 'Current detection status'; COMMENT ON COLUMN workspace_detections.idempotency_key IS 'Detect idempotency key (dedupes retries)';