use axum::{
extract::{Multipart, Path, Query, State},
http::{header::CONTENT_LENGTH, HeaderMap, StatusCode},
response::{Html, IntoResponse, Redirect, Response},
Json,
};
use minijinja::context;
use sea_orm::*;
use serde::{Deserialize, Serialize};
use tracing::{debug, error, info};
use uuid::Uuid;
use crate::entities::{ai_descriptions, ocr_results, tags, upload_tags, uploads};
use crate::jobs;
use crate::middleware::UserContext;
use crate::queue::{publish_job, JobKind, JobMessage};
use crate::storage::{LocalStorage, Storage};
use crate::tags as tagsvc;
use crate::AppState;
#[derive(Debug, Deserialize)]
pub struct UploadQuery {
/// Free-text tag filter, e.g. "desc:dog Series:Friends"
pub tags: Option<String>,
/// Restrict to a single namespace
pub namespace: Option<String>,
pub q: Option<String>,
pub page: Option<u64>,
/// Surfaced in the listing UI as a banner — populated by the upload
/// handler after a redirect on user-fixable failures (too large, etc).
pub error: Option<String>,
pub notice: Option<String>,
}
fn human_bytes(n: usize) -> String {
const MIB: usize = 1024 * 1024;
const KIB: usize = 1024;
if n >= MIB {
format!("{:.1} MiB", n as f64 / MIB as f64)
} else if n >= KIB {
format!("{:.1} KiB", n as f64 / KIB as f64)
} else {
format!("{} B", n)
}
}
fn urlencode(s: &str) -> String {
s.bytes()
.map(|b| match b {
b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'-' | b'_' | b'.' | b'~' => (b as char).to_string(),
b' ' => "+".to_string(),
_ => format!("%{:02X}", b),
})
.collect()
}
/// The user-facing "too large" sentence, with or without a known size.
fn too_large_message(noun: &str, actual: Option<usize>, limit: usize) -> String {
match actual {
Some(n) => format!(
"{noun} too large ({}). Maximum is {}.",
human_bytes(n),
human_bytes(limit)
),
None => format!("{noun} exceeds the maximum of {}.", human_bytes(limit)),
}
}
fn upload_redirect_error(msg: &str) -> Redirect {
Redirect::to(&format!("/uploads?error={}", urlencode(msg)))
}
fn upload_redirect_notice(msg: &str) -> Redirect {
Redirect::to(&format!("/uploads?notice={}", urlencode(msg)))
}
const PAGE_SIZE: u64 = 24;
pub async fn list_uploads(
State(state): State<AppState>,
user: UserContext,
Query(qs): Query<UploadQuery>,
) -> Result<Html<String>, StatusCode> {
let (rows, total) = query_uploads(&state, &user, &qs)
.await
.map_err(|e| {
error!(error = %e, "list_uploads query failed");
StatusCode::INTERNAL_SERVER_ERROR
})?;
let template = state
.env
.get_template("uploads/list.html")
.map_err(|e| { error!(error = %e, "load uploads/list.html failed"); StatusCode::INTERNAL_SERVER_ERROR })?;
let rendered = template
.render(context! {
title => "Uploads",
user => user.user,
uploads => rows,
total => total,
page => qs.page.unwrap_or(1),
page_size => PAGE_SIZE,
filter_tags => qs.tags.clone().unwrap_or_default(),
filter_namespace => qs.namespace.clone().unwrap_or_default(),
filter_q => qs.q.clone().unwrap_or_default(),
error => qs.error.clone(),
notice => qs.notice.clone(),
// Surfaced to the upload forms so the browser can reject an
// oversize file before sending a byte of it.
max_upload_bytes => state.config.max_upload_bytes,
})
.map_err(|e| {
error!(error = %e, "list_uploads render failed");
StatusCode::INTERNAL_SERVER_ERROR
})?;
Ok(Html(rendered))
}
pub async fn upload_file(
State(state): State<AppState>,
user: UserContext,
headers: HeaderMap,
multipart: Multipart,
) -> Response {
// Cheap pre-check: trust Content-Length for an early bailout. The handler
// also re-checks the actual bytes — Content-Length is hint-only.
if let Some(len) = headers
.get(CONTENT_LENGTH)
.and_then(|v| v.to_str().ok())
.and_then(|s| s.parse::<usize>().ok())
&& len > state.config.max_upload_bytes
{
return upload_redirect_error(&too_large_message(
"Upload",
Some(len),
state.config.max_upload_bytes,
))
.into_response();
}
match ingest_multipart(&state, &user, multipart).await {
Ok(id) => Redirect::to(&format!("/uploads/{}", id)).into_response(),
Err(IngestError::TooLarge { actual, limit }) => {
upload_redirect_error(&too_large_message("Upload", actual, limit)).into_response()
}
Err(IngestError::BadRequest(msg)) => upload_redirect_error(&msg).into_response(),
Err(IngestError::Status(code)) => code.into_response(),
}
}
#[derive(Debug, Deserialize)]
pub struct UrlUploadForm {
pub url: String,
pub item_name: Option<String>,
pub item_description: Option<String>,
pub tags: Option<String>,
}
/// POST /uploads/url — fetch a remote file server-side and ingest it. Lets the
/// client paste a link (e.g. an animated GIF or video URL) and keep the
/// original bytes, which a clipboard image-paste would otherwise flatten to PNG.
pub async fn upload_from_url(
State(state): State<AppState>,
user: UserContext,
axum::Form(form): axum::Form<UrlUploadForm>,
) -> Response {
match ingest_url(&state, &user, form).await {
Ok(id) => Redirect::to(&format!("/uploads/{}", id)).into_response(),
Err(IngestError::TooLarge { actual, limit }) => {
upload_redirect_error(&too_large_message("Upload", actual, limit)).into_response()
}
Err(IngestError::BadRequest(msg)) => upload_redirect_error(&msg).into_response(),
Err(IngestError::Status(code)) => code.into_response(),
}
}
/// POST /uploads/zip — ingest every image inside an uploaded zip archive.
/// Each entry becomes its own upload row and runs the usual job pipeline, so a
/// batch of photos can be dropped in without picking them one by one.
pub async fn upload_zip(
State(state): State<AppState>,
user: UserContext,
headers: HeaderMap,
multipart: Multipart,
) -> Response {
if let Some(len) = headers
.get(CONTENT_LENGTH)
.and_then(|v| v.to_str().ok())
.and_then(|s| s.parse::<usize>().ok())
&& len > state.config.max_upload_bytes
{
return upload_redirect_error(&too_large_message(
"Zip",
Some(len),
state.config.max_upload_bytes,
))
.into_response();
}
match ingest_zip(&state, &user, multipart).await {
Ok(summary) => upload_redirect_notice(&summary.message()).into_response(),
Err(IngestError::TooLarge { actual, limit }) => {
upload_redirect_error(&too_large_message("Zip", actual, limit)).into_response()
}
Err(IngestError::BadRequest(msg)) => upload_redirect_error(&msg).into_response(),
Err(IngestError::Status(code)) => code.into_response(),
}
}
#[derive(Debug, Deserialize)]
pub struct AddTagsForm {
pub tags: Option<String>,
}
async fn add_tags_inner(
state: &AppState,
user: &UserContext,
upload_id: Uuid,
raw: &str,
) -> Result<Uuid, StatusCode> {
let upload = uploads::Entity::find_by_id(upload_id)
.filter(uploads::Column::UserId.eq(user.user.id))
.one(state.db.as_ref())
.await
.map_err(|e| { error!(error = %e, "add_tags: find upload failed"); StatusCode::INTERNAL_SERVER_ERROR })?
.ok_or(StatusCode::NOT_FOUND)?;
let parsed = tagsvc::parse_tags(raw);
if parsed.is_empty() {
return Ok(upload.id);
}
if let Err(e) = tagsvc::attach_tags(state.db.as_ref(), upload.id, &parsed, None).await {
error!(error = %e, "attach user tags failed");
return Err(StatusCode::INTERNAL_SERVER_ERROR);
}
info!(upload_id = %upload.id, count = parsed.len(), "user tags attached");
Ok(upload.id)
}
/// POST /uploads/{id}/tags — HTML form path; redirects back to the detail page.
pub async fn add_tags(
State(state): State<AppState>,
user: UserContext,
Path(id): Path<Uuid>,
axum::extract::Form(form): axum::extract::Form<AddTagsForm>,
) -> Result<Redirect, StatusCode> {
let id = add_tags_inner(&state, &user, id, form.tags.as_deref().unwrap_or("")).await?;
Ok(Redirect::to(&format!("/uploads/{}", id)))
}
/// POST /api/uploads/{id}/tags — JSON path; returns the updated upload.
pub async fn api_add_tags(
State(state): State<AppState>,
user: UserContext,
Path(id): Path<Uuid>,
Json(form): Json<AddTagsForm>,
) -> Result<Json<serde_json::Value>, StatusCode> {
add_tags_inner(&state, &user, id, form.tags.as_deref().unwrap_or("")).await?;
// Reuse the existing detail JSON shape so the client can swap in the
// response without a second round-trip.
api_show_upload(State(state), user, Path(id)).await
}
#[derive(Debug, Deserialize)]
pub struct EnqueueJobForm {
pub kind: String,
/// When "on" / "true", clear the existing result row for this kind before
/// enqueuing so the worker doesn't short-circuit via its idempotency check.
pub force: Option<String>,
}
/// POST /uploads/{id}/jobs — enqueue a categorizer job for this upload.
pub async fn enqueue_job(
State(state): State<AppState>,
user: UserContext,
Path(id): Path<Uuid>,
axum::extract::Form(form): axum::extract::Form<EnqueueJobForm>,
) -> Result<Redirect, StatusCode> {
let upload = uploads::Entity::find_by_id(id)
.filter(uploads::Column::UserId.eq(user.user.id))
.one(state.db.as_ref())
.await
.map_err(|e| { error!(error = %e, "enqueue_job: find upload failed"); StatusCode::INTERNAL_SERVER_ERROR })?
.ok_or(StatusCode::NOT_FOUND)?;
let kind = JobKind::from_name(&form.kind).ok_or(StatusCode::BAD_REQUEST)?;
let force = form
.force
.as_deref()
.map(|v| matches!(v.to_ascii_lowercase().as_str(), "on" | "1" | "true" | "yes"))
.unwrap_or(false);
if force {
match kind {
JobKind::Ocr => {
let _ = ocr_results::Entity::delete_many()
.filter(ocr_results::Column::UploadId.eq(upload.id))
.exec(state.db.as_ref())
.await;
}
JobKind::Describe => {
let _ = ai_descriptions::Entity::delete_many()
.filter(ai_descriptions::Column::UploadId.eq(upload.id))
.exec(state.db.as_ref())
.await;
}
_ => {}
}
}
// Reset (or create) the upload_jobs row for this kind. Always — manual
// re-trigger should reflect a fresh attempt regardless of force.
if let Err(e) = jobs::upsert_pending(state.db.as_ref(), upload.id, kind).await {
error!(error = %e, "upsert pending job row failed");
return Err(StatusCode::INTERNAL_SERVER_ERROR);
}
let _ = uploads::Entity::update_many()
.filter(uploads::Column::Id.eq(upload.id))
.col_expr(uploads::Column::ProcessingStatus, sea_orm::sea_query::Expr::value(jobs::UPLOAD_STATUS_PROCESSING))
.exec(state.db.as_ref())
.await;
let conn = crate::queue::connect(&state.config.amqp_url)
.await
.map_err(|e| { error!(error = %e, "amqp connect failed"); StatusCode::INTERNAL_SERVER_ERROR })?;
let msg = JobMessage::new(kind, upload.id, upload.user_id);
publish_job(&conn, &msg)
.await
.map_err(|e| { error!(error = %e, "publish job failed"); StatusCode::INTERNAL_SERVER_ERROR })?;
info!(upload_id = %upload.id, kind = ?kind, force, "job enqueued");
Ok(Redirect::to(&format!("/uploads/{}", upload.id)))
}
pub async fn show_upload(
State(state): State<AppState>,
user: UserContext,
Path(id): Path<Uuid>,
) -> Result<Html<String>, StatusCode> {
let upload = uploads::Entity::find_by_id(id)
.filter(uploads::Column::UserId.eq(user.user.id))
.one(state.db.as_ref())
.await
.map_err(|e| { error!(error = %e, "show_upload: find failed"); StatusCode::INTERNAL_SERVER_ERROR })?
.ok_or(StatusCode::NOT_FOUND)?;
let attached_tags = load_tags_for_upload(state.db.as_ref(), id)
.await
.map_err(|e| { error!(error = %e, "show_upload: load tags failed"); StatusCode::INTERNAL_SERVER_ERROR })?;
let ocr_rows = ocr_results::Entity::find()
.filter(ocr_results::Column::UploadId.eq(id))
.order_by_desc(ocr_results::Column::ProcessedAt)
.all(state.db.as_ref())
.await
.map_err(|e| { error!(error = %e, "load ocr rows failed"); StatusCode::INTERNAL_SERVER_ERROR })?;
let description_rows = ai_descriptions::Entity::find()
.filter(ai_descriptions::Column::UploadId.eq(id))
.order_by_desc(ai_descriptions::Column::ProcessedAt)
.all(state.db.as_ref())
.await
.map_err(|e| { error!(error = %e, "load description rows failed"); StatusCode::INTERNAL_SERVER_ERROR })?;
let job_rows = jobs::jobs_for_upload(state.db.as_ref(), id)
.await
.map_err(|e| { error!(error = %e, "load upload_jobs failed"); StatusCode::INTERNAL_SERVER_ERROR })?;
let mut job_by_kind: std::collections::BTreeMap<String, _> = std::collections::BTreeMap::new();
for row in &job_rows {
job_by_kind.insert(row.kind.clone(), row.clone());
}
let template = state
.env
.get_template("uploads/show.html")
.map_err(|e| { error!(error = %e, "load uploads/show.html failed"); StatusCode::INTERNAL_SERVER_ERROR })?;
let rendered = template
.render(context! {
title => upload.item_name.clone().unwrap_or_else(|| upload.file_name_original.clone()),
user => user.user,
upload => upload,
tags => attached_tags,
ocr_rows => ocr_rows,
description_rows => description_rows,
job_rows => job_rows,
job_by_kind => job_by_kind,
})
.map_err(|e| { error!(error = %e, "render uploads/show.html failed"); StatusCode::INTERNAL_SERVER_ERROR })?;
Ok(Html(rendered))
}
#[derive(Debug, Serialize)]
pub struct UploadJson {
pub id: Uuid,
pub file_name_original: String,
pub item_path: String,
pub processing_status: String,
/// Lets a client tell a video from an image without guessing from the
/// extension: a video's thumbnail is `.thumb.jpg`, not `.thumb.mp4`.
pub content_type: Option<String>,
}
pub async fn api_list_uploads(
State(state): State<AppState>,
user: UserContext,
Query(qs): Query<UploadQuery>,
) -> Result<Json<serde_json::Value>, StatusCode> {
let (rows, total) = query_uploads(&state, &user, &qs)
.await
.map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
let json: Vec<_> = rows
.into_iter()
.map(|u| UploadJson {
id: u.id,
file_name_original: u.file_name_original,
item_path: u.item_path,
processing_status: u.processing_status,
content_type: u.content_type,
})
.collect();
Ok(Json(serde_json::json!({
"total": total,
"page": qs.page.unwrap_or(1),
"page_size": PAGE_SIZE,
"items": json,
})))
}
pub async fn api_upload_file(
State(state): State<AppState>,
user: UserContext,
headers: HeaderMap,
multipart: Multipart,
) -> Response {
if let Some(len) = headers
.get(CONTENT_LENGTH)
.and_then(|v| v.to_str().ok())
.and_then(|s| s.parse::<usize>().ok())
&& len > state.config.max_upload_bytes
{
return (
StatusCode::PAYLOAD_TOO_LARGE,
Json(serde_json::json!({
"error": "payload_too_large",
"max_bytes": state.config.max_upload_bytes,
"received_bytes": len,
})),
)
.into_response();
}
match ingest_multipart(&state, &user, multipart).await {
Ok(id) => Json(serde_json::json!({ "id": id })).into_response(),
Err(IngestError::TooLarge { actual, limit }) => (
StatusCode::PAYLOAD_TOO_LARGE,
Json(serde_json::json!({
"error": "payload_too_large",
"max_bytes": limit,
"received_bytes": actual,
"message": too_large_message("Upload", actual, limit),
})),
)
.into_response(),
Err(IngestError::BadRequest(msg)) => (
StatusCode::BAD_REQUEST,
Json(serde_json::json!({"error": msg})),
)
.into_response(),
Err(IngestError::Status(code)) => code.into_response(),
}
}
/// POST /api/uploads/url — JSON path; fetch a remote file server-side and
/// return the new upload id. Body: `{"url": "...", "item_name"?: "...", ...}`.
pub async fn api_upload_from_url(
State(state): State<AppState>,
user: UserContext,
Json(form): Json<UrlUploadForm>,
) -> Response {
match ingest_url(&state, &user, form).await {
Ok(id) => Json(serde_json::json!({ "id": id })).into_response(),
Err(IngestError::TooLarge { actual, limit }) => (
StatusCode::PAYLOAD_TOO_LARGE,
Json(serde_json::json!({
"error": "payload_too_large",
"max_bytes": limit,
"received_bytes": actual,
"message": too_large_message("Upload", actual, limit),
})),
)
.into_response(),
Err(IngestError::BadRequest(msg)) => (
StatusCode::BAD_REQUEST,
Json(serde_json::json!({"error": msg})),
)
.into_response(),
Err(IngestError::Status(code)) => code.into_response(),
}
}
/// POST /api/uploads/zip — JSON sibling of `upload_zip`. Returns the ids that
/// were created plus the counts for everything that was not.
pub async fn api_upload_zip(
State(state): State<AppState>,
user: UserContext,
headers: HeaderMap,
multipart: Multipart,
) -> Response {
if let Some(len) = headers
.get(CONTENT_LENGTH)
.and_then(|v| v.to_str().ok())
.and_then(|s| s.parse::<usize>().ok())
&& len > state.config.max_upload_bytes
{
return (
StatusCode::PAYLOAD_TOO_LARGE,
Json(serde_json::json!({
"error": "payload_too_large",
"max_bytes": state.config.max_upload_bytes,
"received_bytes": len,
})),
)
.into_response();
}
match ingest_zip(&state, &user, multipart).await {
Ok(summary) => Json(serde_json::json!({
"imported": summary.imported,
"imported_count": summary.imported.len(),
"duplicates": summary.duplicates,
"retagged": summary.retagged,
"skipped_unsupported": summary.skipped_unsupported,
"skipped_too_large": summary.skipped_too_large,
"failed": summary.failed,
"truncated": summary.truncated,
"message": summary.message(),
}))
.into_response(),
Err(IngestError::TooLarge { actual, limit }) => (
StatusCode::PAYLOAD_TOO_LARGE,
Json(serde_json::json!({
"error": "payload_too_large",
"max_bytes": limit,
"received_bytes": actual,
"message": too_large_message("Upload", actual, limit),
})),
)
.into_response(),
Err(IngestError::BadRequest(msg)) => (
StatusCode::BAD_REQUEST,
Json(serde_json::json!({"error": msg})),
)
.into_response(),
Err(IngestError::Status(code)) => code.into_response(),
}
}
pub async fn api_show_upload(
State(state): State<AppState>,
user: UserContext,
Path(id): Path<Uuid>,
) -> Result<Json<serde_json::Value>, StatusCode> {
let upload = uploads::Entity::find_by_id(id)
.filter(uploads::Column::UserId.eq(user.user.id))
.one(state.db.as_ref())
.await
.map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?
.ok_or(StatusCode::NOT_FOUND)?;
let attached_tags = load_tags_for_upload(state.db.as_ref(), id)
.await
.map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
let tag_strings: Vec<String> = attached_tags
.iter()
.map(|t| match &t.namespace {
Some(ns) => format!("{}:{}", ns, t.tag_value),
None => t.tag_value.clone(),
})
.collect();
Ok(Json(serde_json::json!({
"id": upload.id,
"file_name_original": upload.file_name_original,
"item_path": upload.item_path,
"item_name": upload.item_name,
"item_description": upload.item_description,
"processing_status": upload.processing_status,
"content_type": upload.content_type,
"date_uploaded": upload.date_uploaded,
"tags": tag_strings,
})))
}
pub enum IngestError {
/// `actual` is None when the limit tripped mid-stream and the real size is
/// unknown — reporting a synthetic "limit + 1" would read as nonsense
/// ("400.0 MiB. Maximum is 400.0 MiB.").
TooLarge { actual: Option<usize>, limit: usize },
BadRequest(String),
Status(StatusCode),
}
impl From<StatusCode> for IngestError {
fn from(c: StatusCode) -> Self { IngestError::Status(c) }
}
/// Flatten an error and everything it wraps into one string.
///
/// Axum's `MultipartError` displays as a bland "Error parsing multipart/form-data
/// request" and keeps the interesting part — including the body-limit rejection —
/// in its source chain, so matching on the top-level message alone misses it.
fn error_chain(err: &dyn std::error::Error) -> String {
let mut out = err.to_string();
let mut source = err.source();
while let Some(cause) = source {
out.push_str(": ");
out.push_str(&cause.to_string());
source = cause.source();
}
out
}
/// Map a multipart read failure onto our friendly variants. The body-limit
/// rejection arrives as a wrapped error rather than a distinct type, so match
/// the flattened chain to turn "body too large" into the same TooLarge the
/// explicit byte checks produce.
fn multipart_err(state: &AppState, err: String, message: &str) -> IngestError {
error!(error = %err, context = message, "multipart read failed");
let lower = err.to_ascii_lowercase();
if lower.contains("length limit") || lower.contains("too large") || lower.contains("body limit") {
IngestError::TooLarge {
actual: None,
limit: state.config.max_upload_bytes,
}
} else {
IngestError::BadRequest(message.to_string())
}
}
async fn ingest_multipart(
state: &AppState,
user: &UserContext,
mut multipart: Multipart,
) -> Result<Uuid, IngestError> {
let mut file_bytes: Option<Vec<u8>> = None;
let mut original_name: Option<String> = None;
let mut content_type: Option<String> = None;
let mut item_name: Option<String> = None;
let mut item_description: Option<String> = None;
let mut tags_raw: Option<String> = None;
loop {
let field = match multipart.next_field().await {
Ok(Some(f)) => f,
Ok(None) => break,
Err(e) => {
return Err(multipart_err(state, error_chain(&e), "Malformed multipart upload."));
}
};
let name = field.name().unwrap_or("").to_string();
match name.as_str() {
"file" => {
original_name = field.file_name().map(|s| s.to_string());
content_type = field.content_type().map(|s| s.to_string());
let bytes = field
.bytes()
.await
.map_err(|e| multipart_err(state, error_chain(&e), "Could not read uploaded file."))?;
if bytes.len() > state.config.max_upload_bytes {
return Err(IngestError::TooLarge {
actual: Some(bytes.len()),
limit: state.config.max_upload_bytes,
});
}
file_bytes = Some(bytes.to_vec());
}
"item_name" => {
item_name = field.text().await.ok();
}
"item_description" => {
item_description = field.text().await.ok();
}
"tags" => {
tags_raw = field.text().await.ok();
}
_ => {
let _ = field.bytes().await;
}
}
}
let bytes = file_bytes.ok_or_else(|| IngestError::BadRequest("No file selected.".into()))?;
let original_name = original_name.unwrap_or_else(|| "upload.bin".to_string());
store_upload(
state,
user,
StoredUpload {
bytes,
original_name,
content_type,
item_name,
item_description,
tags_raw,
},
)
.await
}
/// Bytes plus metadata ready to be persisted — the common shape every ingest
/// path (multipart upload, URL fetch) converges on.
struct StoredUpload {
bytes: Vec<u8>,
original_name: String,
content_type: Option<String>,
item_name: Option<String>,
item_description: Option<String>,
tags_raw: Option<String>,
}
/// Shared tail of every single-file ingest path: persist, then enqueue the
/// default job set. Duplicates short-circuit without re-enqueuing.
async fn store_upload(
state: &AppState,
user: &UserContext,
data: StoredUpload,
) -> Result<Uuid, IngestError> {
let result = persist_upload(state, user, data).await?;
if !result.duplicate {
enqueue_default_jobs(state, &result.upload).await;
}
Ok(result.upload.id)
}
/// Outcome of persisting one file.
struct Persisted {
upload: uploads::Model,
/// These bytes were already in this user's archive.
duplicate: bool,
/// Tags from this request were attached to a row that already existed.
retagged: bool,
}
/// Attach the request's user tags to an upload. Returns whether anything was
/// actually attached, which is only interesting when the row already existed.
/// `attach_tags` is idempotent, so re-tagging never doubles a tag up.
async fn attach_request_tags(state: &AppState, upload_id: Uuid, tags_raw: Option<&str>) -> bool {
let Some(raw) = tags_raw else { return false };
let parsed = tagsvc::parse_tags(raw);
if parsed.is_empty() {
return false;
}
match tagsvc::attach_tags(state.db.as_ref(), upload_id, &parsed, None).await {
Ok(()) => true,
Err(e) => {
error!(error = %e, "attach user tags failed");
false
}
}
}
/// Persist to storage, dedupe per user, insert the row and attach tags.
///
/// A duplicate short-circuits without re-storing anything, but still takes the
/// request's tags: re-uploading a set with a tag in the box is a reasonable way
/// to label an archive you already have, and it is the obvious move after a
/// partly-failed import.
async fn persist_upload(
state: &AppState,
user: &UserContext,
data: StoredUpload,
) -> Result<Persisted, IngestError> {
let StoredUpload {
bytes,
original_name,
content_type,
item_name,
item_description,
tags_raw,
} = data;
let extension = std::path::Path::new(&original_name)
.extension()
.and_then(|s| s.to_str())
.map(|s| s.to_lowercase());
let resolved_ct = content_type.clone().or_else(|| {
mime_guess::from_path(&original_name)
.first()
.map(|m| m.to_string())
});
let storage = LocalStorage::new(&state.config.storage_root);
let stored = storage
.put(&bytes, extension.as_deref())
.await
.map_err(|e| {
error!(error = %e, "storage put failed");
StatusCode::INTERNAL_SERVER_ERROR
})?;
// Dedupe per user via the unique (file_hash, user_id) index.
if let Some(existing) = uploads::Entity::find()
.filter(uploads::Column::UserId.eq(user.user.id))
.filter(uploads::Column::FileHash.eq(stored.sha256_hex.clone()))
.one(state.db.as_ref())
.await
.map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?
{
let retagged = attach_request_tags(state, existing.id, tags_raw.as_deref()).await;
info!(upload_id = %existing.id, retagged, "duplicate upload short-circuited");
return Ok(Persisted { upload: existing, duplicate: true, retagged });
}
let active = uploads::ActiveModel {
id: Set(Uuid::now_v7()),
user_id: Set(user.user.id),
file_name_original: Set(original_name.clone()),
file_name_generated: Set(stored.relative_path.clone()),
file_size_bytes: Set(stored.size_bytes as i64),
content_type: Set(resolved_ct),
file_hash: Set(Some(stored.sha256_hex.clone())),
item_name: Set(item_name),
item_description: Set(item_description),
item_path: Set(stored.relative_path.clone()),
storage_backend: Set("local".to_string()),
processing_status: Set("pending".to_string()),
processing_error: Set(None),
date_uploaded: Set(chrono::Utc::now().into()),
date_updated: Set(chrono::Utc::now().into()),
};
let inserted = active
.insert(state.db.as_ref())
.await
.map_err(|e| {
error!(error = %e, "upload insert failed");
StatusCode::INTERNAL_SERVER_ERROR
})?;
attach_request_tags(state, inserted.id, tags_raw.as_deref()).await;
Ok(Persisted { upload: inserted, duplicate: false, retagged: false })
}
/// Fetch a remote URL server-side and hand the bytes to [`store_upload`].
/// Fetching server-side (rather than relying on the clipboard) preserves the
/// original encoding — animated GIFs and videos arrive intact.
async fn ingest_url(
state: &AppState,
user: &UserContext,
form: UrlUploadForm,
) -> Result<Uuid, IngestError> {
let url = form.url.trim();
// Only fetch real web URLs. This is the SSRF guard surface — http(s) only,
// no file://, no scheme-relative or internal protocols.
if !(url.starts_with("http://") || url.starts_with("https://")) {
return Err(IngestError::BadRequest(
"Only http(s) URLs can be fetched.".into(),
));
}
let client = reqwest::Client::builder()
.timeout(std::time::Duration::from_secs(30))
.build()
.map_err(|e| {
error!(error = %e, "build http client failed");
IngestError::Status(StatusCode::INTERNAL_SERVER_ERROR)
})?;
let resp = client
.get(url)
.send()
.await
.map_err(|e| IngestError::BadRequest(format!("Could not fetch URL: {e}")))?;
if !resp.status().is_success() {
return Err(IngestError::BadRequest(format!(
"Remote server returned {} for that URL.",
resp.status().as_u16()
)));
}
// Early bail on advertised size before we buffer the body.
if let Some(len) = resp.content_length()
&& len as usize > state.config.max_upload_bytes
{
return Err(IngestError::TooLarge {
actual: Some(len as usize),
limit: state.config.max_upload_bytes,
});
}
// Content-Type header without any `; charset=…` parameters.
let content_type = resp
.headers()
.get(axum::http::header::CONTENT_TYPE)
.and_then(|v| v.to_str().ok())
.map(|s| s.split(';').next().unwrap_or(s).trim().to_string())
.filter(|s| !s.is_empty());
let bytes = resp
.bytes()
.await
.map_err(|e| IngestError::BadRequest(format!("Could not read remote body: {e}")))?
.to_vec();
if bytes.len() > state.config.max_upload_bytes {
return Err(IngestError::TooLarge {
actual: Some(bytes.len()),
limit: state.config.max_upload_bytes,
});
}
// Derive a filename from the URL path; backfill an extension from the
// content type when the path has none (common for query-driven URLs).
let mut original_name = url
.split(['?', '#'])
.next()
.unwrap_or(url)
.rsplit('/')
.next()
.unwrap_or("")
.trim()
.to_string();
if original_name.is_empty() {
original_name = "download".to_string();
}
if std::path::Path::new(&original_name).extension().is_none()
&& let Some(ext) = content_type
.as_deref()
.and_then(mime_guess::get_mime_extensions_str)
.and_then(|exts| exts.first())
{
original_name = format!("{original_name}.{ext}");
}
store_upload(
state,
user,
StoredUpload {
bytes,
original_name,
content_type,
item_name: form.item_name.filter(|s| !s.trim().is_empty()),
item_description: form.item_description.filter(|s| !s.trim().is_empty()),
tags_raw: form.tags,
},
)
.await
}
// ============================================================================
// Zip ingest — one archive of images in, many uploads out
// ============================================================================
/// A spooled upload that deletes itself when it goes out of scope, whichever
/// way the handler exits.
struct SpooledFile(std::path::PathBuf);
impl Drop for SpooledFile {
fn drop(&mut self) {
match std::fs::remove_file(&self.0) {
Ok(()) => {}
Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}
Err(e) => error!(error = %e, path = %self.0.display(), "spooled upload cleanup failed"),
}
}
}
/// Stream a multipart field to disk, enforcing `limit` as the bytes arrive.
///
/// Spooling rather than buffering is what keeps a multi-gigabyte archive off
/// the heap — `Field::bytes()` would hold the whole thing in memory at once.
/// The spool lives under STORAGE_ROOT, not `/tmp`, because `/tmp` is tmpfs on
/// most hosts and would put those gigabytes straight back into RAM.
async fn spool_field(
state: &AppState,
field: &mut axum::extract::multipart::Field<'_>,
limit: usize,
) -> Result<(SpooledFile, usize), IngestError> {
use tokio::io::AsyncWriteExt;
let dir = std::path::Path::new(&state.config.storage_root).join("tmp");
tokio::fs::create_dir_all(&dir).await.map_err(|e| {
error!(error = %e, "create spool dir failed");
IngestError::Status(StatusCode::INTERNAL_SERVER_ERROR)
})?;
let spooled = SpooledFile(dir.join(format!("upload-{}.part", Uuid::now_v7())));
let mut file = tokio::fs::File::create(&spooled.0).await.map_err(|e| {
error!(error = %e, "create spool file failed");
IngestError::Status(StatusCode::INTERNAL_SERVER_ERROR)
})?;
let mut written: usize = 0;
while let Some(chunk) = field
.chunk()
.await
.map_err(|e| multipart_err(state, error_chain(&e), "Could not read uploaded file."))?
{
written += chunk.len();
// Bail the moment the running total crosses the line — no reason to
// keep writing bytes we have already decided to reject.
if written > limit {
return Err(IngestError::TooLarge { actual: Some(written), limit });
}
file.write_all(&chunk).await.map_err(|e| {
error!(error = %e, "write to spool failed");
IngestError::Status(StatusCode::INTERNAL_SERVER_ERROR)
})?;
}
file.flush().await.map_err(|e| {
error!(error = %e, "flush spool failed");
IngestError::Status(StatusCode::INTERNAL_SERVER_ERROR)
})?;
Ok((spooled, written))
}
/// Tally of a zip import. Everything the import chose *not* to store is
/// counted so the user gets an honest report instead of a silent truncation.
#[derive(Debug, Default)]
pub struct ZipImportSummary {
pub imported: Vec<Uuid>,
/// Entries whose bytes this user had already uploaded.
pub duplicates: usize,
/// Of those duplicates, how many picked up this request's tags.
pub retagged: usize,
pub skipped_unsupported: usize,
pub skipped_too_large: usize,
pub failed: usize,
/// Hit MAX_ZIP_ENTRIES or MAX_ZIP_TOTAL_BYTES — the tail of the archive
/// was never read.
pub truncated: bool,
}
impl ZipImportSummary {
pub fn message(&self) -> String {
let n = self.imported.len();
// "files", not "images" — an archive of videos imports just as happily.
let mut msg = format!(
"Imported {} file{} from the zip",
n,
if n == 1 { "" } else { "s" }
);
let mut notes: Vec<String> = Vec::new();
if self.duplicates > 0 {
notes.push(format!("{} already in the archive", self.duplicates));
}
if self.retagged > 0 {
notes.push(format!("{} retagged", self.retagged));
}
if self.skipped_unsupported > 0 {
notes.push(format!("{} unsupported entries skipped", self.skipped_unsupported));
}
if self.skipped_too_large > 0 {
notes.push(format!("{} over the size limit", self.skipped_too_large));
}
if self.failed > 0 {
notes.push(format!("{} failed", self.failed));
}
if notes.is_empty() {
msg.push('.');
} else {
msg = format!("{} ({}).", msg, notes.join(", "));
}
if self.truncated {
msg.push_str(" Archive limits reached — the rest of the zip was not read.");
}
msg
}
}
/// One image lifted out of a zip, on its way to [`persist_upload`].
struct ZipEntryFile {
name: String,
content_type: Option<String>,
bytes: Vec<u8>,
}
/// Counters the blocking scan owns; merged into the summary once it finishes.
#[derive(Debug, Default)]
struct ZipScan {
skipped_unsupported: usize,
skipped_too_large: usize,
truncated: bool,
}
/// Walk the archive on a blocking thread, handing each image to `tx` as it is
/// decompressed. The archive is read from its spool on disk and entries stream
/// through a small channel, so peak memory is a handful of images regardless of
/// how large the zip is.
fn scan_zip_blocking(
zip_path: std::path::PathBuf,
max_entry_bytes: usize,
max_total_bytes: usize,
max_entries: usize,
tx: tokio::sync::mpsc::Sender<ZipEntryFile>,
) -> Result<ZipScan, String> {
use std::io::Read;
let file = std::fs::File::open(&zip_path).map_err(|e| {
error!(error = %e, path = %zip_path.display(), "open spooled zip failed");
"Could not read the uploaded zip file.".to_string()
})?;
// Seeking through the file on disk rather than a heap buffer is the whole
// point — peak memory is one entry, not the whole archive.
let mut archive = zip::ZipArchive::new(std::io::BufReader::new(file))
.map_err(|e| format!("Could not read that zip file: {e}"))?;
let mut scan = ZipScan::default();
let mut total_bytes: usize = 0;
let mut taken: usize = 0;
for i in 0..archive.len() {
if taken >= max_entries {
scan.truncated = true;
break;
}
let mut entry = match archive.by_index(i) {
Ok(e) => e,
Err(e) => {
error!(error = %e, index = i, "zip entry unreadable");
scan.skipped_unsupported += 1;
continue;
}
};
if entry.is_dir() {
continue;
}
// enclosed_name() rejects absolute paths and `..` traversal. Storage is
// content-addressed so the name never reaches the filesystem, but an
// untrustworthy name is reason enough to leave the entry alone.
let Some(path) = entry.enclosed_name() else {
scan.skipped_unsupported += 1;
continue;
};
// Finder resource forks ride along in most macOS-made zips; so do
// dotfiles. Neither is something the user meant to upload, so they
// don't count against the skipped tally.
if path.components().any(|c| c.as_os_str() == "__MACOSX") {
continue;
}
let name = match path.file_name().and_then(|n| n.to_str()) {
Some(n) if !n.starts_with('.') => n.to_string(),
_ => continue,
};
// Anything the pipeline can categorize: images directly, videos via an
// extracted poster frame. Everything else (an export's stray .json or
// .txt) is left out rather than becoming an upload row nobody wanted.
let content_type = mime_guess::from_path(&name).first().map(|m| m.to_string());
if !content_type
.as_deref()
.map(|ct| ct.starts_with("image/") || ct.starts_with("video/"))
.unwrap_or(false)
{
scan.skipped_unsupported += 1;
continue;
}
let declared = entry.size() as usize;
if declared > max_entry_bytes {
scan.skipped_too_large += 1;
continue;
}
if total_bytes.saturating_add(declared) > max_total_bytes {
scan.truncated = true;
break;
}
// Read through a Take: the central directory's size is a claim, not a
// guarantee, and a crafted zip can keep producing bytes past it.
let mut buf = Vec::with_capacity(declared);
if let Err(e) = (&mut entry)
.take(max_entry_bytes as u64 + 1)
.read_to_end(&mut buf)
{
error!(error = %e, entry = %name, "zip entry decompression failed");
scan.skipped_unsupported += 1;
continue;
}
if buf.len() > max_entry_bytes {
scan.skipped_too_large += 1;
continue;
}
total_bytes += buf.len();
taken += 1;
// A send error means the consumer went away — stop decompressing.
if tx
.blocking_send(ZipEntryFile { name, content_type, bytes: buf })
.is_err()
{
break;
}
}
Ok(scan)
}
/// Ingest every image in an uploaded zip. `tags` and `item_description` from
/// the form, when present, are applied to every image in the archive.
async fn ingest_zip(
state: &AppState,
user: &UserContext,
mut multipart: Multipart,
) -> Result<ZipImportSummary, IngestError> {
let mut spooled_zip: Option<SpooledFile> = None;
let mut item_description: Option<String> = None;
let mut tags_raw: Option<String> = None;
loop {
let mut field = match multipart.next_field().await {
Ok(Some(f)) => f,
Ok(None) => break,
Err(e) => return Err(multipart_err(state, error_chain(&e), "Malformed multipart upload.")),
};
match field.name().unwrap_or("") {
"file" => {
let (spooled, size) =
spool_field(state, &mut field, state.config.max_upload_bytes).await?;
debug!(bytes = size, path = %spooled.0.display(), "zip spooled to disk");
spooled_zip = Some(spooled);
}
"item_description" => {
item_description = field.text().await.ok().filter(|s| !s.trim().is_empty());
}
"tags" => {
tags_raw = field.text().await.ok();
}
_ => {
let _ = field.bytes().await;
}
}
}
// Held to the end of the function: dropping it deletes the spool, and the
// blocking scan below is reading from that exact path.
let spooled_zip =
spooled_zip.ok_or_else(|| IngestError::BadRequest("No zip file selected.".into()))?;
let (tx, mut rx) = tokio::sync::mpsc::channel::<ZipEntryFile>(4);
let zip_path = spooled_zip.0.clone();
let max_entry_bytes = state.config.max_upload_bytes;
let max_total_bytes = state.config.max_zip_total_bytes;
let max_entries = state.config.max_zip_entries;
let scanner = tokio::task::spawn_blocking(move || {
scan_zip_blocking(zip_path, max_entry_bytes, max_total_bytes, max_entries, tx)
});
let mut summary = ZipImportSummary::default();
let mut stored: Vec<uploads::Model> = Vec::new();
while let Some(entry) = rx.recv().await {
let name = entry.name.clone();
let result = persist_upload(
state,
user,
StoredUpload {
bytes: entry.bytes,
original_name: entry.name,
content_type: entry.content_type,
item_name: None,
item_description: item_description.clone(),
tags_raw: tags_raw.clone(),
},
)
.await;
match result {
Ok(p) if p.duplicate => {
summary.duplicates += 1;
if p.retagged {
summary.retagged += 1;
}
}
Ok(p) => {
summary.imported.push(p.upload.id);
stored.push(p.upload);
}
// One bad entry shouldn't abort the import — count it and carry on.
Err(_) => {
error!(entry = %name, "zip entry ingest failed");
summary.failed += 1;
}
}
}
let scan = match scanner.await {
Ok(Ok(scan)) => scan,
// A malformed archive fails before the first entry, so nothing has been
// stored yet and this is safe to surface as a plain bad request.
Ok(Err(msg)) => return Err(IngestError::BadRequest(msg)),
Err(e) => {
error!(error = %e, "zip scan task panicked");
return Err(IngestError::Status(StatusCode::INTERNAL_SERVER_ERROR));
}
};
summary.skipped_unsupported = scan.skipped_unsupported;
summary.skipped_too_large = scan.skipped_too_large;
summary.truncated = scan.truncated;
// One AMQP connection for the whole archive.
enqueue_default_jobs_many(state, &stored).await;
info!(
imported = summary.imported.len(),
duplicates = summary.duplicates,
skipped_unsupported = summary.skipped_unsupported,
skipped_too_large = summary.skipped_too_large,
failed = summary.failed,
truncated = summary.truncated,
"zip import finished"
);
drop(spooled_zip);
Ok(summary)
}
fn default_job_kinds(upload: &uploads::Model) -> Vec<JobKind> {
let ct = upload.content_type.as_deref().unwrap_or("");
let mut kinds = if ct.starts_with("image/") {
vec![JobKind::Thumbnail, JobKind::Ocr, JobKind::Describe]
} else if ct.starts_with("video/") {
// Poster extraction, not the still-image categorizers: none of them can
// decode a video, so the poster job runs first and chains them itself
// once there is a frame to work on.
vec![JobKind::Poster]
} else {
Vec::new()
};
// The preview reads the original directly, so it needn't wait for a poster.
if crate::workers::preview::wants_preview(upload) {
kinds.push(JobKind::Preview);
}
kinds
}
async fn enqueue_default_jobs(state: &AppState, upload: &uploads::Model) {
enqueue_default_jobs_many(state, std::slice::from_ref(upload)).await;
}
/// Dispatch the default job set for a batch of fresh uploads over a single
/// AMQP connection. A zip import can land hundreds of rows at once; connecting
/// per upload would be one handshake per image.
async fn enqueue_default_jobs_many(state: &AppState, batch: &[uploads::Model]) {
let planned: Vec<(&uploads::Model, Vec<JobKind>)> = batch
.iter()
.map(|u| (u, default_job_kinds(u)))
.filter(|(_, kinds)| !kinds.is_empty())
.collect();
if planned.is_empty() {
return;
}
// Create the pending rows BEFORE publishing so the aggregate computed by
// workers sees the full set up front.
for (upload, kinds) in &planned {
for kind in kinds {
if let Err(e) = jobs::upsert_pending(state.db.as_ref(), upload.id, *kind).await {
error!(error = %e, ?kind, "upsert pending job row failed");
}
}
}
// Bring uploads.processing_status in line with the pending rows.
let ids: Vec<Uuid> = planned.iter().map(|(u, _)| u.id).collect();
let _ = uploads::Entity::update_many()
.filter(uploads::Column::Id.is_in(ids))
.col_expr(uploads::Column::ProcessingStatus, sea_orm::sea_query::Expr::value(jobs::UPLOAD_STATUS_PROCESSING))
.exec(state.db.as_ref())
.await;
let conn = match crate::queue::connect(&state.config.amqp_url).await {
Ok(c) => c,
Err(e) => {
error!(error = %e, "amqp connect failed; jobs will not be dispatched");
return;
}
};
for (upload, kinds) in planned {
for kind in kinds {
let msg = JobMessage::new(kind, upload.id, upload.user_id);
if let Err(e) = publish_job(&conn, &msg).await {
error!(kind = ?kind, error = %e, "publish failed");
}
}
}
}
#[cfg(test)]
mod zip_unit_tests {
use super::*;
use std::io::Write;
/// Build an in-memory zip from (name, bytes) pairs.
fn make_zip(entries: &[(&str, &[u8])]) -> Vec<u8> {
let mut w = zip::ZipWriter::new(std::io::Cursor::new(Vec::new()));
let opts = zip::write::SimpleFileOptions::default()
.compression_method(zip::CompressionMethod::Stored);
for (name, bytes) in entries {
w.start_file(*name, opts).expect("start_file");
w.write_all(bytes).expect("write");
}
w.finish().expect("finish").into_inner()
}
/// The scan reads from disk, so spool the bytes the way the handler does.
/// SpooledFile cleans the path up when the test drops it.
fn spool(zip_bytes: &[u8]) -> SpooledFile {
let path = std::env::temp_dir().join(format!("dms-test-{}.zip", Uuid::now_v7()));
std::fs::write(&path, zip_bytes).expect("write test zip");
SpooledFile(path)
}
/// Run the scan and drain the channel. Capacity comfortably exceeds the
/// entry count so the blocking sends never park.
fn scan(zip_bytes: Vec<u8>, max_entry: usize, max_total: usize, max_entries: usize)
-> (Vec<ZipEntryFile>, ZipScan)
{
let spooled = spool(&zip_bytes);
let (tx, mut rx) = tokio::sync::mpsc::channel::<ZipEntryFile>(64);
let scan = scan_zip_blocking(spooled.0.clone(), max_entry, max_total, max_entries, tx)
.expect("scan should succeed");
let mut got = Vec::new();
while let Ok(entry) = rx.try_recv() {
got.push(entry);
}
(got, scan)
}
#[test]
fn picks_media_and_skips_the_rest() {
let zip = make_zip(&[
("a.png", b"fake-png"),
("notes.txt", b"hello"),
("sub/dir/b.jpg", b"fake-jpg"),
("__MACOSX/._a.png", b"resource fork"),
(".hidden.png", b"dotfile"),
]);
let (got, scan) = scan(zip, 1024, 1024 * 1024, 100);
let names: Vec<&str> = got.iter().map(|e| e.name.as_str()).collect();
assert_eq!(names, vec!["a.png", "b.jpg"], "nested paths flatten to the basename");
assert_eq!(got[0].content_type.as_deref(), Some("image/png"));
// notes.txt is the only genuine skip — Finder forks and dotfiles are
// junk the user never chose to upload, so they aren't tallied.
assert_eq!(scan.skipped_unsupported, 1);
assert_eq!(scan.skipped_too_large, 0);
assert!(!scan.truncated);
}
#[test]
fn videos_are_taken_alongside_images() {
// The shape of an imgur export: mostly mp4, the odd gif.
let zip = make_zip(&[
("1 - KVHchZH.gif", b"fake-gif"),
("2 - TtjZyX4.mp4", b"fake-mp4"),
("3 - k91N7NT.mp4", b"fake-mp4"),
("index.json", b"{}"),
]);
let (got, scan) = scan(zip, 1024, 1024 * 1024, 100);
let types: Vec<&str> = got.iter().filter_map(|e| e.content_type.as_deref()).collect();
assert_eq!(types, vec!["image/gif", "video/mp4", "video/mp4"]);
assert_eq!(scan.skipped_unsupported, 1, "only the json is unsupported");
}
/// Names that climb out of the archive are refused by enclosed_name() and
/// tallied as skips, never taken. A leading `/` is only made relative, so
/// that entry stays inside and is taken by its basename.
#[test]
fn escaping_names_are_skipped() {
let zip = make_zip(&[
("../escape.png", b"up one"),
("sub/../../escape.jpg", b"up two"),
("/abs.png", b"rooted"),
("ok.png", b"fine"),
]);
let (got, scan) = scan(zip, 1024, 1024 * 1024, 100);
let names: Vec<&str> = got.iter().map(|e| e.name.as_str()).collect();
assert_eq!(names, vec!["abs.png", "ok.png"]);
assert_eq!(scan.skipped_unsupported, 2);
}
#[test]
fn oversize_entries_are_skipped_not_fatal() {
let zip = make_zip(&[("small.png", b"tiny"), ("big.png", &[0u8; 512])]);
let (got, scan) = scan(zip, 64, 1024 * 1024, 100);
assert_eq!(got.len(), 1);
assert_eq!(got[0].name, "small.png");
assert_eq!(scan.skipped_too_large, 1);
}
#[test]
fn entry_cap_truncates() {
let zip = make_zip(&[("a.png", b"one"), ("b.png", b"two"), ("c.png", b"three")]);
let (got, scan) = scan(zip, 1024, 1024 * 1024, 2);
assert_eq!(got.len(), 2);
assert!(scan.truncated, "hitting MAX_ZIP_ENTRIES must be reported");
}
#[test]
fn total_byte_cap_truncates() {
let zip = make_zip(&[("a.png", &[0u8; 100]), ("b.png", &[0u8; 100])]);
// Room for the first entry only.
let (got, scan) = scan(zip, 1024, 150, 100);
assert_eq!(got.len(), 1);
assert!(scan.truncated);
}
#[test]
fn garbage_is_not_a_zip() {
let spooled = spool(b"definitely not a zip");
let (tx, _rx) = tokio::sync::mpsc::channel::<ZipEntryFile>(4);
let err = scan_zip_blocking(spooled.0.clone(), 1024, 1024, 10, tx)
.expect_err("garbage must be rejected");
assert!(err.starts_with("Could not read that zip file"), "got: {err}");
}
#[test]
fn spool_is_removed_on_drop() {
let path = {
let spooled = spool(b"transient");
let p = spooled.0.clone();
assert!(p.exists());
p
};
assert!(!path.exists(), "the spool must not outlive its guard");
}
#[test]
fn too_large_message_omits_an_unknown_size() {
// The mid-stream limit trip knows only that the request went over.
assert_eq!(
too_large_message("Zip", None, 50 * 1024 * 1024),
"Zip exceeds the maximum of 50.0 MiB."
);
assert_eq!(
too_large_message("Upload", Some(200 * 1024 * 1024), 100 * 1024 * 1024),
"Upload too large (200.0 MiB). Maximum is 100.0 MiB."
);
}
#[test]
fn summary_message_reports_every_bucket() {
let summary = ZipImportSummary {
imported: vec![Uuid::now_v7()],
duplicates: 2,
retagged: 2,
skipped_unsupported: 3,
skipped_too_large: 1,
failed: 1,
truncated: true,
};
let msg = summary.message();
assert!(msg.starts_with("Imported 1 file from the zip ("), "got: {msg}");
assert!(msg.contains("2 already in the archive"));
assert!(msg.contains("2 retagged"));
assert!(msg.contains("3 unsupported entries skipped"));
assert!(msg.contains("1 over the size limit"));
assert!(msg.contains("1 failed"));
assert!(msg.contains("the rest of the zip was not read"));
}
#[test]
fn summary_message_reports_a_pure_retag() {
// Re-importing an archive purely to label it: nothing new, everything
// tagged. The message has to make that legible.
let summary = ZipImportSummary {
duplicates: 469,
retagged: 469,
..Default::default()
};
assert_eq!(
summary.message(),
"Imported 0 files from the zip (469 already in the archive, 469 retagged)."
);
}
#[test]
fn summary_message_stays_plain_when_clean() {
let summary = ZipImportSummary {
imported: vec![Uuid::now_v7(), Uuid::now_v7()],
..Default::default()
};
assert_eq!(summary.message(), "Imported 2 files from the zip.");
}
}
// ============================================================================
// Dedicated search page
// ============================================================================
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum SearchSpec {
/// Bare value, e.g. "macro" — matches `tag_value LIKE %macro%` against
/// any selected tag-source namespace, plus any selected text fields.
Value(String),
/// `ns:value` — matches `namespace = ns AND tag_value LIKE %value%`.
NsValue { ns: String, val: String },
/// `ns:*` — matches every tag in the namespace, ignoring tag value.
/// Text-field sources are skipped entirely; only tag matches contribute.
NsWildcard(String),
}
/// Parse a query string into a SearchSpec.
pub fn parse_search_spec(q: &str) -> SearchSpec {
match q.split_once(':') {
Some((ns, val)) if !ns.trim().is_empty() => {
let ns = ns.trim().to_string();
let val = val.trim();
if val == "*" {
SearchSpec::NsWildcard(ns)
} else if val.is_empty() {
// "species:" — treat as bare so we don't surprise the user with
// an open-ended namespace wildcard.
SearchSpec::Value(q.to_string())
} else {
SearchSpec::NsValue { ns, val: val.to_string() }
}
}
_ => SearchSpec::Value(q.to_string()),
}
}
/// Backwards-compatible split kept for callers that only need the namespace
/// and value strings.
pub fn split_namespace_query(q: &str) -> (Option<String>, String) {
match parse_search_spec(q) {
SearchSpec::Value(v) => (None, v),
SearchSpec::NsValue { ns, val } => (Some(ns), val),
SearchSpec::NsWildcard(ns) => (Some(ns), "*".to_string()),
}
}
#[derive(Debug, Deserialize, Default)]
pub struct SearchQuery {
pub q: Option<String>,
/// Marker that the form was submitted, so we know to respect the absence
/// of an `s_*` checkbox as "this source is off". Without it, missing
/// fields default to all-on (first visit).
pub f: Option<String>,
pub s_filename: Option<String>,
pub s_item_name: Option<String>,
pub s_description: Option<String>,
pub s_user_tags: Option<String>,
pub s_ocr: Option<String>,
pub s_desc: Option<String>,
/// Any namespaced tag whose namespace isn't `ocr` or `desc` — covers tags
/// the user added by hand, future categorizers, etc.
pub s_other_ns: Option<String>,
pub page: Option<u64>,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct SearchSources {
pub filename: bool,
pub item_name: bool,
pub description: bool,
pub user_tags: bool,
pub ocr: bool,
pub desc: bool,
pub other_ns: bool,
}
impl SearchSources {
pub fn from_query(qs: &SearchQuery) -> Self {
let submitted = qs.f.as_deref().map(|v| !v.is_empty()).unwrap_or(false);
if !submitted {
return Self {
filename: true, item_name: true, description: true,
user_tags: true, ocr: true, desc: true, other_ns: true,
};
}
fn on(v: &Option<String>) -> bool {
v.as_deref().map(|s| !s.is_empty()).unwrap_or(false)
}
Self {
filename: on(&qs.s_filename),
item_name: on(&qs.s_item_name),
description: on(&qs.s_description),
user_tags: on(&qs.s_user_tags),
ocr: on(&qs.s_ocr),
desc: on(&qs.s_desc),
other_ns: on(&qs.s_other_ns),
}
}
pub fn any(&self) -> bool {
self.filename || self.item_name || self.description
|| self.user_tags || self.ocr || self.desc || self.other_ns
}
pub const fn all_on() -> Self {
Self {
filename: true, item_name: true, description: true,
user_tags: true, ocr: true, desc: true, other_ns: true,
}
}
}
#[cfg(test)]
mod search_unit_tests {
use super::*;
fn qs(s: SearchQuery) -> SearchQuery { s }
#[test]
fn split_bare_query() {
assert_eq!(split_namespace_query("macro"), (None, "macro".to_string()));
}
#[test]
fn split_namespaced_query() {
assert_eq!(split_namespace_query("species:macro"),
(Some("species".to_string()), "macro".to_string()));
}
#[test]
fn split_trims_whitespace() {
assert_eq!(split_namespace_query(" species : macro "),
(Some("species".to_string()), "macro".to_string()));
}
#[test]
fn split_empty_half_treated_as_bare() {
assert_eq!(split_namespace_query(":foo"), (None, ":foo".to_string()));
assert_eq!(split_namespace_query("foo:"), (None, "foo:".to_string()));
assert_eq!(split_namespace_query(":"), (None, ":".to_string()));
}
#[test]
fn split_only_first_colon() {
// Subsequent colons are part of the value half.
assert_eq!(
split_namespace_query("series:friends:s01"),
(Some("series".to_string()), "friends:s01".to_string())
);
}
#[test]
fn spec_bare() {
assert_eq!(parse_search_spec("macro"), SearchSpec::Value("macro".into()));
}
#[test]
fn spec_namespaced() {
assert_eq!(
parse_search_spec("species:macro"),
SearchSpec::NsValue { ns: "species".into(), val: "macro".into() }
);
}
#[test]
fn spec_wildcard() {
assert_eq!(parse_search_spec("species:*"), SearchSpec::NsWildcard("species".into()));
assert_eq!(parse_search_spec(" species : * "), SearchSpec::NsWildcard("species".into()));
}
#[test]
fn spec_trailing_colon_is_bare() {
// "species:" is treated as a bare query (against the literal "species:")
// rather than a namespace wildcard, so users can still search for
// text containing a colon without surprise.
assert!(matches!(parse_search_spec("species:"), SearchSpec::Value(_)));
}
#[test]
fn sources_default_all_on_when_unsubmitted() {
let s = SearchSources::from_query(&qs(SearchQuery::default()));
assert_eq!(s, SearchSources::all_on());
}
#[test]
fn sources_respect_explicit_off() {
// f=1 marks the form as submitted; missing checkboxes mean off.
let s = SearchSources::from_query(&qs(SearchQuery {
f: Some("1".into()),
s_filename: Some("on".into()),
// every other source omitted
..Default::default()
}));
assert!(s.filename);
assert!(!s.item_name);
assert!(!s.user_tags);
assert!(!s.ocr);
assert!(!s.desc);
assert!(!s.other_ns);
assert!(s.any());
}
#[test]
fn sources_no_box_means_no_search() {
let s = SearchSources::from_query(&qs(SearchQuery {
f: Some("1".into()),
..Default::default()
}));
assert!(!s.any());
}
}
pub async fn search_uploads(
State(state): State<AppState>,
user: UserContext,
Query(qs): Query<SearchQuery>,
) -> Result<Html<String>, StatusCode> {
let sources = SearchSources::from_query(&qs);
let q = qs.q.as_deref().map(str::trim).filter(|s| !s.is_empty()).map(String::from);
let result = if !sources.any() || q.is_none() {
SearchResult { rows: Vec::new(), total: 0, matched_tags: Default::default() }
} else {
run_search(&state, &user, q.as_deref().unwrap(), &sources, qs.page.unwrap_or(1))
.await
.map_err(|e| { error!(error = %e, "search query failed"); StatusCode::INTERNAL_SERVER_ERROR })?
};
let template = state.env.get_template("search.html")
.map_err(|e| { error!(error = %e, "load search.html failed"); StatusCode::INTERNAL_SERVER_ERROR })?;
let rendered = template
.render(context! {
title => "Search",
user => user.user,
uploads => result.rows,
total => result.total,
matched_tags => result.matched_tags,
page => qs.page.unwrap_or(1),
page_size => PAGE_SIZE,
q => q.clone().unwrap_or_default(),
sources => serde_json::json!({
"filename": sources.filename,
"item_name": sources.item_name,
"description": sources.description,
"user_tags": sources.user_tags,
"ocr": sources.ocr,
"desc": sources.desc,
"other_ns": sources.other_ns,
}),
submitted => qs.f.is_some(),
})
.map_err(|e| { error!(error = %e, "render search.html failed"); StatusCode::INTERNAL_SERVER_ERROR })?;
Ok(Html(rendered))
}
/// JSON sibling of `search_uploads` for the API surface. Same query params
/// (`q`, `f`, source toggles, `page`) so the Android client can hit the
/// same URL shape it builds for the web. Returns matched-tag info per row
/// so the client can render the same "matched on …" badge as the web.
pub async fn api_search_uploads(
State(state): State<AppState>,
user: UserContext,
Query(qs): Query<SearchQuery>,
) -> Result<Json<serde_json::Value>, StatusCode> {
let sources = SearchSources::from_query(&qs);
let q = qs.q.as_deref().map(str::trim).filter(|s| !s.is_empty()).map(String::from);
let result = if !sources.any() || q.is_none() {
SearchResult { rows: Vec::new(), total: 0, matched_tags: Default::default() }
} else {
run_search(&state, &user, q.as_deref().unwrap(), &sources, qs.page.unwrap_or(1))
.await
.map_err(|e| { error!(error = %e, "api search query failed"); StatusCode::INTERNAL_SERVER_ERROR })?
};
let items: Vec<_> = result
.rows
.iter()
.map(|u| {
let matched: Vec<String> = result
.matched_tags
.get(&u.id)
.map(|ts| {
ts.iter()
.map(|t| match &t.namespace {
Some(ns) => format!("{}:{}", ns, t.tag_value),
None => t.tag_value.clone(),
})
.collect()
})
.unwrap_or_default();
serde_json::json!({
"id": u.id,
"file_name_original": u.file_name_original,
"item_path": u.item_path,
"processing_status": u.processing_status,
"content_type": u.content_type,
"matched_tags": matched,
})
})
.collect();
Ok(Json(serde_json::json!({
"total": result.total,
"page": qs.page.unwrap_or(1),
"page_size": PAGE_SIZE,
"q": q,
"sources": {
"filename": sources.filename,
"item_name": sources.item_name,
"description": sources.description,
"user_tags": sources.user_tags,
"ocr": sources.ocr,
"desc": sources.desc,
"other_ns": sources.other_ns,
},
"items": items,
})))
}
#[derive(Debug, Clone)]
pub struct SearchResult {
pub rows: Vec<uploads::Model>,
pub total: u64,
/// upload_id → tags that triggered the match (empty if the upload only
/// matched on a text field like file_name).
pub matched_tags: std::collections::HashMap<Uuid, Vec<tags::Model>>,
}
/// Public for integration tests; production code goes through `search_uploads`.
pub async fn run_search_for_user(
db: &DatabaseConnection,
user_id: Uuid,
q: &str,
sources: &SearchSources,
page: u64,
) -> Result<SearchResult, DbErr> {
run_search_inner(db, user_id, q, sources, page).await
}
async fn run_search(
state: &AppState,
user: &UserContext,
q: &str,
sources: &SearchSources,
page: u64,
) -> Result<SearchResult, DbErr> {
run_search_inner(state.db.as_ref(), user.user.id, q, sources, page).await
}
async fn run_search_inner(
db: &DatabaseConnection,
user_id: Uuid,
q: &str,
sources: &SearchSources,
page: u64,
) -> Result<SearchResult, DbErr> {
let spec = parse_search_spec(q);
debug!(?spec, ?sources, "search input");
// OR over the selected sources. Each source contributes a Condition node.
let mut any = Condition::any();
let mut matched_anything = false;
let mut matched_tag_models: Vec<tags::Model> = Vec::new();
let mut tag_ids_by_upload: std::collections::HashMap<Uuid, Vec<Uuid>> =
std::collections::HashMap::new();
// Namespace-wildcard queries (`species:*`) only target tags. Skipping the
// text columns avoids matching `LIKE '%species:*%'` against file names.
let text_searchable = !matches!(spec, SearchSpec::NsWildcard(_));
if text_searchable {
let pat = format!("%{}%", q.replace('%', "\\%"));
if sources.filename {
any = any.add(uploads::Column::FileNameOriginal.like(pat.clone()));
matched_anything = true;
}
if sources.item_name {
any = any.add(uploads::Column::ItemName.like(pat.clone()));
matched_anything = true;
}
if sources.description {
any = any.add(uploads::Column::ItemDescription.like(pat.clone()));
matched_anything = true;
}
}
// Tag-based sources: resolve tag IDs once, then use a single EXISTS over
// upload_tags. Combine all selected tag namespaces into one tag-id pool.
if sources.user_tags || sources.ocr || sources.desc || sources.other_ns {
let mut ns_cond = Condition::any();
if sources.user_tags {
ns_cond = ns_cond.add(tags::Column::Namespace.is_null());
}
if sources.ocr {
ns_cond = ns_cond.add(tags::Column::Namespace.eq("ocr"));
}
if sources.desc {
ns_cond = ns_cond.add(tags::Column::Namespace.eq("desc"));
}
if sources.other_ns {
// Any namespace that isn't already covered above. Always require
// the namespace to be non-null (otherwise it's a "user_tag").
let mut other = Condition::all().add(tags::Column::Namespace.is_not_null());
if !sources.ocr {
// ocr is off; including ocr here would defeat the user's choice
other = other.add(tags::Column::Namespace.ne("ocr"));
} else {
// ocr is already covered above; exclude here to avoid duplicate matches
other = other.add(tags::Column::Namespace.ne("ocr"));
}
// desc is either off or already covered above; exclude it either way.
other = other.add(tags::Column::Namespace.ne("desc"));
ns_cond = ns_cond.add(other);
}
// Build the tag query based on the parsed spec. The user-selected
// namespace checkboxes (ns_cond) always gate, so typing `ocr:foo`
// with the OCR box unchecked still won't surface ocr rows.
let mut tag_query = tags::Entity::find().filter(ns_cond);
match &spec {
SearchSpec::Value(v) => {
let pat = format!("%{}%", v.replace('%', "\\%"));
tag_query = tag_query.filter(tags::Column::TagValue.like(pat));
}
SearchSpec::NsValue { ns, val } => {
let pat = format!("%{}%", val.replace('%', "\\%"));
tag_query = tag_query
.filter(tags::Column::TagValue.like(pat))
.filter(tags::Column::Namespace.eq(ns.as_str()));
}
SearchSpec::NsWildcard(ns) => {
tag_query = tag_query.filter(tags::Column::Namespace.eq(ns.as_str()));
}
}
matched_tag_models = tag_query.all(db).await?;
debug!(
tag_count = matched_tag_models.len(),
sample = ?matched_tag_models.iter().take(5).map(|t| (t.namespace.clone(), t.tag_value.clone())).collect::<Vec<_>>(),
"tag matches before subquery"
);
let tag_ids: Vec<Uuid> = matched_tag_models.iter().map(|t| t.id).collect();
if !tag_ids.is_empty() {
// sea-orm's Expr::col(uploads::Column::Id) inside a subquery emits
// an unqualified "id" that resolves against the subquery's own
// FROM (upload_tags), so the correlation silently breaks. Avoid
// the foot-gun: resolve matching upload_ids up front and use IN.
let attaches = upload_tags::Entity::find()
.filter(upload_tags::Column::TagId.is_in(tag_ids.clone()))
.all(db)
.await?;
for a in &attaches {
tag_ids_by_upload.entry(a.upload_id).or_default().push(a.tag_id);
}
let upload_ids: Vec<Uuid> = tag_ids_by_upload.keys().copied().collect();
if !upload_ids.is_empty() {
any = any.add(uploads::Column::Id.is_in(upload_ids));
matched_anything = true;
}
}
}
if !matched_anything {
return Ok(SearchResult { rows: Vec::new(), total: 0, matched_tags: Default::default() });
}
let query = uploads::Entity::find()
.filter(uploads::Column::UserId.eq(user_id))
.filter(any);
let total = query.clone().count(db).await?;
let page = Ord::max(page, 1);
let rows = query
.order_by_desc(uploads::Column::DateUploaded)
.paginate(db, PAGE_SIZE)
.fetch_page(page - 1)
.await?;
// Fan out matched_tag_models + tag_ids_by_upload into a per-upload tag list
// for just the rows on this page.
let tag_lookup: std::collections::HashMap<Uuid, tags::Model> = matched_tag_models
.into_iter()
.map(|t| (t.id, t))
.collect();
let mut matched_tags: std::collections::HashMap<Uuid, Vec<tags::Model>> =
std::collections::HashMap::new();
for row in &rows {
if let Some(ids) = tag_ids_by_upload.get(&row.id) {
let mut ts: Vec<tags::Model> = ids
.iter()
.filter_map(|id| tag_lookup.get(id).cloned())
.collect();
ts.sort_by(|a, b| {
(a.namespace.clone(), a.tag_value.clone())
.cmp(&(b.namespace.clone(), b.tag_value.clone()))
});
if !ts.is_empty() {
matched_tags.insert(row.id, ts);
}
}
}
Ok(SearchResult { rows, total, matched_tags })
}
async fn query_uploads(
state: &AppState,
user: &UserContext,
qs: &UploadQuery,
) -> Result<(Vec<uploads::Model>, u64), DbErr> {
let mut query = uploads::Entity::find().filter(uploads::Column::UserId.eq(user.user.id));
// Free-text on file name + item name + description
if let Some(q) = qs.q.as_deref().map(str::trim).filter(|s| !s.is_empty()) {
let pat = format!("%{}%", q.replace('%', "\\%"));
query = query.filter(
Condition::any()
.add(uploads::Column::FileNameOriginal.like(pat.clone()))
.add(uploads::Column::ItemName.like(pat.clone()))
.add(uploads::Column::ItemDescription.like(pat)),
);
}
// Tag filtering — for each parsed tag, the upload must have a matching attach.
let parsed_tags = qs
.tags
.as_deref()
.map(tagsvc::parse_tags)
.unwrap_or_default();
let ns_filter = qs
.namespace
.as_deref()
.map(str::trim)
.filter(|s| !s.is_empty());
if !parsed_tags.is_empty() || ns_filter.is_some() {
// Resolve tag IDs once. An empty matched set with a tag filter present
// means "no results".
let mut tag_id_groups: Vec<Vec<Uuid>> = Vec::new();
for tag in &parsed_tags {
let mut q = tags::Entity::find().filter(tags::Column::TagValue.eq(&tag.value));
q = match &tag.namespace {
Some(ns) => q.filter(tags::Column::Namespace.eq(ns)),
None => q,
};
let ids: Vec<Uuid> = q.all(state.db.as_ref()).await?.into_iter().map(|t| t.id).collect();
if ids.is_empty() {
return Ok((Vec::new(), 0));
}
tag_id_groups.push(ids);
}
if let Some(ns) = ns_filter {
let ids: Vec<Uuid> = tags::Entity::find()
.filter(tags::Column::Namespace.eq(ns))
.all(state.db.as_ref())
.await?
.into_iter()
.map(|t| t.id)
.collect();
if ids.is_empty() {
return Ok((Vec::new(), 0));
}
tag_id_groups.push(ids);
}
// Each group must have at least one match. The correlated EXISTS
// form silently breaks with sea-orm (column refs don't qualify into
// the subquery's FROM), so resolve upload_id sets up front and AND
// them with IN. Slight extra round-trip; obviously correct.
for group in tag_id_groups {
let ids: Vec<Uuid> = upload_tags::Entity::find()
.filter(upload_tags::Column::TagId.is_in(group))
.all(state.db.as_ref())
.await?
.into_iter()
.map(|ut| ut.upload_id)
.collect();
if ids.is_empty() {
return Ok((Vec::new(), 0));
}
query = query.filter(uploads::Column::Id.is_in(ids));
}
}
let total = query.clone().count(state.db.as_ref()).await?;
let page = Ord::max(qs.page.unwrap_or(1), 1);
let rows = query
.order_by_desc(uploads::Column::DateUploaded)
.paginate(state.db.as_ref(), PAGE_SIZE)
.fetch_page(page - 1)
.await?;
Ok((rows, total))
}
async fn load_tags_for_upload(
db: &DatabaseConnection,
upload_id: Uuid,
) -> Result<Vec<tags::Model>, DbErr> {
let attached = upload_tags::Entity::find()
.filter(upload_tags::Column::UploadId.eq(upload_id))
.find_also_related(tags::Entity)
.all(db)
.await?;
Ok(attached.into_iter().filter_map(|(_, t)| t).collect())
}
⭐
Log in to mark this repository.
77.9 KiB · text
History
8aa8e40