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 upload_redirect_error(msg: &str) -> Redirect {
let encoded: String = msg
.as_bytes()
.iter()
.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();
Redirect::to(&format!("/uploads?error={}", encoded))
}
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(),
})
.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())
{
if len > state.config.max_upload_bytes {
return upload_redirect_error(&format!(
"Upload too large ({}). Maximum is {}.",
human_bytes(len),
human_bytes(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(&format!(
"Upload too large ({}). Maximum is {}.",
human_bytes(actual),
human_bytes(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(&format!(
"Upload too large ({}). Maximum is {}.",
human_bytes(actual),
human_bytes(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_str(&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,
}
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,
})
.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())
{
if 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,
})),
)
.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,
})),
)
.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,
"date_uploaded": upload.date_uploaded,
"tags": tag_strings,
})))
}
pub enum IngestError {
TooLarge { actual: usize, limit: usize },
BadRequest(String),
Status(StatusCode),
}
impl From<StatusCode> for IngestError {
fn from(c: StatusCode) -> Self { IngestError::Status(c) }
}
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) => {
// Map Axum's "body too large" rejection into our friendly
// TooLarge variant so the user sees a nice message.
let s = e.to_string();
error!(error = %s, "multipart next_field failed");
if s.to_ascii_lowercase().contains("length limit") || s.to_ascii_lowercase().contains("too large") {
return Err(IngestError::TooLarge {
actual: state.config.max_upload_bytes + 1,
limit: state.config.max_upload_bytes,
});
}
return Err(IngestError::BadRequest("Malformed multipart upload.".into()));
}
};
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| {
let s = e.to_string();
error!(error = %s, "multipart read file bytes failed");
if s.to_ascii_lowercase().contains("length limit") || s.to_ascii_lowercase().contains("too large") {
IngestError::TooLarge {
actual: state.config.max_upload_bytes + 1,
limit: state.config.max_upload_bytes,
}
} else {
IngestError::BadRequest("Could not read uploaded file.".into())
}
})?;
if bytes.len() > state.config.max_upload_bytes {
return Err(IngestError::TooLarge {
actual: 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 ingest path: persist to storage, dedupe per user,
/// insert the row, attach tags, and enqueue processing jobs.
async fn store_upload(
state: &AppState,
user: &UserContext,
data: StoredUpload,
) -> Result<Uuid, 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)?
{
info!(upload_id = %existing.id, "duplicate upload short-circuited");
return Ok(existing.id);
}
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
})?;
if let Some(raw) = tags_raw.as_deref() {
let parsed = tagsvc::parse_tags(raw);
if let Err(e) = tagsvc::attach_tags(state.db.as_ref(), inserted.id, &parsed, None).await {
error!(error = %e, "attach user tags failed");
}
}
enqueue_default_jobs(state, &inserted).await;
Ok(inserted.id)
}
/// 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() {
if len as usize > state.config.max_upload_bytes {
return Err(IngestError::TooLarge {
actual: 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: 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() {
if 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
}
async fn enqueue_default_jobs(state: &AppState, upload: &uploads::Model) {
let is_image = upload
.content_type
.as_deref()
.map(|ct| ct.starts_with("image/"))
.unwrap_or(false);
let mut kinds: Vec<JobKind> = Vec::new();
if is_image {
kinds.push(JobKind::Thumbnail);
kinds.push(JobKind::Ocr);
kinds.push(JobKind::Describe);
}
if kinds.is_empty() {
return;
}
// Create the pending rows BEFORE publishing so the aggregate computed by
// workers sees the full set up front.
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 _ = 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 = 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 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");
}
}
}
// ============================================================================
// 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,
"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"));
}
if !sources.desc {
other = other.add(tags::Column::Namespace.ne("desc"));
} else {
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 = page.max(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 = qs.page.unwrap_or(1).max(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.
50.2 KiB · text
History
2e5cc52