nils/papercrate
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Skip uploads of existing documents and support checksum preflight in importer

Browse Source
This commit is contained in:
nils
2025-10-26 02:56:35 +02:00
parent 9b7ca3d692
commit 157f03b655
4 changed files with 207 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/src/openapi.rs
+48 -1
View File
@@ -13,6 +13,7 @@ use uuid::Uuid;
doc::me,
doc::select_tenant,
doc::list_documents,
doc::check_document,
doc::upload_document,
doc::get_document,
doc::update_document,
@@ -81,6 +82,8 @@ use uuid::Uuid;
schemas::ReanalyzeResponse,
schemas::DocumentAssetRequestParams,
schemas::AssetObjectsQuery,
schemas::DocumentCheckQuery,
schemas::DocumentCheckResponse,
schemas::UploadDocumentForm,
schemas::CreateFolderRequest,
schemas::EnsureFolderPathRequest,
@@ -189,11 +192,24 @@ mod doc {
post,
path = "/api/documents",
request_body = UploadDocumentForm,
responses((status = 201, description = "Document uploaded", body = DocumentDetailResponse)),
responses(
(status = 201, description = "Document created", body = DocumentDetailResponse),
(status = 200, description = "Existing document reused", body = DocumentDetailResponse),
(status = 204, description = "Upload skipped because the document already exists")
),
tag = "Documents"
)]
pub(super) fn upload_document() {}
#[utoipa::path(
get,
path = "/api/documents/check",
params(DocumentCheckQuery),
responses((status = 200, description = "Checksum lookup", body = DocumentCheckResponse)),
tag = "Documents"
)]
pub(super) fn check_document() {}
#[utoipa::path(
get,
path = "/api/documents/{id}",
@@ -785,6 +801,29 @@ pub mod schemas {
pub limit: Option<i32>,
}
#[derive(Serialize, Deserialize, IntoParams, ToSchema)]
#[into_params(parameter_in = Query)]
pub struct DocumentCheckQuery {
pub checksum: String,
}
#[derive(Serialize, Deserialize, ToSchema)]
pub struct DocumentCheckResponse {
pub exists: bool,
#[schema(nullable)]
pub document_id: Option<Uuid>,
#[schema(nullable)]
pub title: Option<String>,
#[schema(nullable)]
pub filename: Option<String>,
#[schema(nullable)]
pub version_id: Option<Uuid>,
#[schema(nullable)]
pub version_number: Option<i32>,
#[schema(nullable)]
pub uploaded_at: Option<String>,
}
#[derive(Serialize, Deserialize, ToSchema)]
pub struct UploadDocumentForm {
#[schema(value_type = String, format = Binary)]
@@ -795,6 +834,14 @@ pub mod schemas {
pub metadata: Option<Value>,
#[schema(nullable)]
pub title: Option<String>,
#[schema(nullable, value_type = Vec<Uuid>)]
pub tag_ids: Option<Vec<Uuid>>,
#[schema(nullable, value_type = Vec<CorrespondentAssignment>)]
pub correspondents: Option<Vec<CorrespondentAssignment>>,
#[schema(nullable, example = "2024-01-01T00:00:00Z")]
pub issued_at: Option<String>,
#[schema(nullable)]
pub skip_existing: Option<bool>,
}
#[derive(Serialize, Deserialize, ToSchema)]
backend/src/routes/documents.rs
+148 -28
View File
@@ -76,6 +76,29 @@ pub struct AssetRequestQuery {
pub force: bool,
}
#[derive(Deserialize, IntoParams, ToSchema)]
#[into_params(parameter_in = Query)]
pub struct DocumentCheckQuery {
pub checksum: String,
}
#[derive(Serialize, ToSchema)]
pub struct DocumentCheckResponse {
pub exists: bool,
#[serde(skip_serializing_if = "Option::is_none")]
pub document_id: Option<Uuid>,
#[serde(skip_serializing_if = "Option::is_none")]
pub title: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub filename: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub version_id: Option<Uuid>,
#[serde(skip_serializing_if = "Option::is_none")]
pub version_number: Option<i32>,
#[serde(skip_serializing_if = "Option::is_none")]
pub uploaded_at: Option<String>,
}
#[derive(Serialize, ToSchema)]
pub struct TagResponse {
pub id: Uuid,
@@ -297,11 +320,13 @@ struct UploadRequest {
tag_ids: Vec<Uuid>,
correspondents: Vec<CorrespondentAssignmentInput>,
issued_at_override: Option<NaiveDateTime>,
skip_if_existing: bool,
}
struct UploadOutcome {
detail: DocumentDetailResponse,
created: bool,
enum UploadOutcome {
Created(DocumentDetailResponse),
Reused(DocumentDetailResponse),
Skipped { document_id: Uuid },
}
#[derive(ToSchema)]
@@ -312,6 +337,16 @@ pub struct UploadDocumentForm {
pub folder_id: Option<Uuid>,
#[schema(nullable)]
pub metadata: Option<Value>,
#[schema(nullable)]
pub title: Option<String>,
#[schema(nullable, value_type = Vec<Uuid>)]
pub tag_ids: Option<Vec<Uuid>>,
#[schema(nullable, value_type = Vec<CorrespondentAssignmentInput>)]
pub correspondents: Option<Vec<CorrespondentAssignmentInput>>,
#[schema(nullable, example = "2024-01-01T00:00:00Z")]
pub issued_at: Option<String>,
#[schema(nullable)]
pub skip_existing: Option<bool>,
}
#[derive(Deserialize, ToSchema)]
@@ -586,6 +621,59 @@ pub async fn list_documents(
Ok(Json(response))
}
pub async fn check_document(
Query(query): Query<DocumentCheckQuery>,
TenantScopedConn {
mut conn,
tenant_id,
..
}: TenantScopedConn,
) -> AppResult<Json<DocumentCheckResponse>> {
let checksum_raw = query.checksum.trim();
if checksum_raw.is_empty() {
return Err(AppError::bad_request("checksum must not be empty"));
}
let checksum = checksum_raw.to_ascii_lowercase();
if !checksum.chars().all(|ch| ch.is_ascii_hexdigit()) {
return Err(AppError::bad_request(
"checksum must be a hex-encoded string",
));
}
let record: Option<(Document, DocumentVersion)> = documents::table
.inner_join(
document_versions::table.on(document_versions::id.eq(documents::current_version_id)),
)
.filter(documents::tenant_id.eq(tenant_id))
.filter(document_versions::checksum.eq(&checksum))
.select((documents::all_columns, document_versions::all_columns))
.first(&mut conn)
.optional()?;
if let Some((document, version)) = record {
Ok(Json(DocumentCheckResponse {
exists: true,
document_id: Some(document.id),
title: Some(document.title.clone()),
filename: Some(document.filename.clone()),
version_id: Some(version.id),
version_number: Some(version.version_number),
uploaded_at: Some(to_iso(document.uploaded_at)),
}))
} else {
Ok(Json(DocumentCheckResponse {
exists: false,
document_id: None,
title: None,
filename: None,
version_id: None,
version_number: None,
uploaded_at: None,
}))
}
}
pub async fn get_document(
State(state): State<AppState>,
Path(document_id): Path<Uuid>,
@@ -634,7 +722,7 @@ pub async fn upload_document(
tenant_id, user_id, ..
}: TenantScopedConn,
mut multipart: Multipart,
) -> AppResult<(StatusCode, Json<DocumentDetailResponse>)> {
) -> AppResult<impl IntoResponse> {
let mut file_bytes: Option<Vec<u8>> = None;
let mut original_name: Option<String> = None;
let mut content_type: Option<String> = None;
@@ -643,6 +731,7 @@ pub async fn upload_document(
let mut tag_ids: Vec<Uuid> = Vec::new();
let mut correspondents: Vec<CorrespondentAssignmentInput> = Vec::new();
let mut issued_at_override: Option<NaiveDateTime> = None;
let mut skip_if_existing = false;
let mut title_override: Option<String> = None;
while let Some(field) = multipart.next_field().await.map_err(|err| {
@@ -751,6 +840,17 @@ pub async fn upload_document(
issued_at_override = Some(parsed.naive_utc());
}
}
Some("skip_existing") => {
let value = field.text().await.map_err(|err| {
let msg = format!("invalid skip_existing flag: {err}");
error!(error = %err, "invalid skip_existing payload");
AppError::bad_request(msg)
})?;
skip_if_existing = matches!(
value.trim().to_ascii_lowercase().as_str(),
"1" | "true" | "yes"
);
}
_ => {}
}
}
@@ -780,31 +880,45 @@ pub async fn upload_document(
tag_ids,
correspondents,
issued_at_override,
skip_if_existing,
};
let outcome = match process_upload(&state, request, tenant_id, user_id).await {
Ok(outcome) => {
info!(
document_id = %outcome.detail.document.id,
original_name = %outcome.detail.document.original_name,
created = outcome.created,
reused_existing = !outcome.created,
"document upload succeeded"
);
outcome
}
Ok(outcome) => outcome,
Err(err) => {
error!(error = ?err, original_name = %original_name_for_log, "document upload failed");
return Err(err);
}
};
let status = if outcome.created {
StatusCode::CREATED
} else {
StatusCode::OK
let response = match outcome {
UploadOutcome::Created(detail) => {
info!(
document_id = %detail.document.id,
original_name = %detail.document.original_name,
created = true,
reused_existing = false,
"document upload succeeded",
);
(StatusCode::CREATED, Json(detail)).into_response()
}
UploadOutcome::Reused(detail) => {
info!(
document_id = %detail.document.id,
original_name = %detail.document.original_name,
created = false,
reused_existing = true,
"document upload succeeded",
);
(StatusCode::OK, Json(detail)).into_response()
}
UploadOutcome::Skipped { document_id } => {
info!(document_id = %document_id, "document upload skipped by client request");
StatusCode::NO_CONTENT.into_response()
}
};
Ok((status, Json(outcome.detail)))
Ok(response)
}
pub async fn request_document_assets(
@@ -1692,6 +1806,7 @@ async fn process_upload(
tag_ids,
correspondents,
issued_at_override,
skip_if_existing,
} = request;
if let Some(folder) = folder_id {
@@ -1729,6 +1844,17 @@ async fn process_upload(
.optional()?;
if let Some((mut document, version)) = existing {
if skip_if_existing {
info!(
document_id = %document.id,
checksum = %checksum_hex,
"upload skipped existing document due to skip flag",
);
return Ok(UploadOutcome::Skipped {
document_id: document.id,
});
}
if let Some(issued_at) = issued_at_override {
if document.issued_at != Some(issued_at) {
diesel::update(
@@ -1782,8 +1908,7 @@ async fn process_upload(
"upload deduplicated existing document"
);
return Ok(UploadOutcome {
detail: DocumentDetailResponse {
return Ok(UploadOutcome::Reused(DocumentDetailResponse {
document: to_document_response(
state,
user_id,
@@ -1792,9 +1917,7 @@ async fn process_upload(
correspondents,
Some((version_response, assets)),
)?,
},
created: false,
});
}));
}
}
@@ -1917,10 +2040,7 @@ async fn process_upload(
warn!(document_id = %doc_id, "failed to enqueue analyze job due to pool error");
}
Ok(UploadOutcome {
detail,
created: true,
})
Ok(UploadOutcome::Created(detail))
}
fn assign_tags_internal(
backend/src/routes/mod.rs
+1
View File
@@ -57,6 +57,7 @@ pub fn create_router(state: AppState) -> Router<()> {
.route("/me", get(auth::me));
let documents_routes = Router::new()
.route("/check", get(documents::check_document))
.route(
"/",
get(documents::list_documents).post(documents::upload_document),
docs/api.txt
+2 -1
View File
@@ -17,7 +17,8 @@ Health
Documents
---------
- GET /api/documents - List or search documents. Optional filters: `folder_id` (defaults to root when omitted), `include_deleted`, `include_descendants` (defaults to true when a `folder_id` is provided and no other override is supplied), `query` (Quickwit full-text), `tags` (comma-separated tag UUIDs), and `correspondents` (comma-separated correspondent UUIDs). Each entry includes tags, correspondent assignments, and current version info.
- POST /api/documents - Upload a document via multipart form-data (`file`, optional `title`, `folder_id`, and JSON `metadata`). When `title` is supplied, the stored filename is set to `<title><original_extension>` automatically.
- GET /api/documents/check?checksum=<sha256> - Lightweight checksum preflight. Returns `exists=false` when no document with the supplied SHA-256 checksum is present; otherwise returns `exists=true` plus the current document metadata.
- POST /api/documents - Upload a document via multipart form-data. Required field: `file`. Optional fields: `title`, `folder_id`, JSON `metadata`, JSON array `tag_ids`, JSON array `correspondents` (each with `correspondent_id` and `role`), and `issued_at` (RFC3339). When `title` is supplied, the stored filename becomes `<title><original_extension>`. Include `skip_existing=true` to receive `204 No Content` instead of reusing a matching document.
- POST /api/documents/bulk/move - Move multiple documents to a target folder.
- POST /api/documents/bulk/tags - Add or remove tags across multiple documents.
- POST /api/documents/bulk/correspondents - Bulk correspondent actions. Default `action=add` replaces existing assignments for the provided roles before adding the supplied correspondents; `action=remove` drops the specified correspondent/role pairs.