Files
ImageApi/src/enhance/handlers.rs
T
Cameron CordesandClaude Opus 5.5 37a2386458 List every photo's open Enhance jobs for a jobs manager
GET /enhance/jobs without a path now answers the jobs manager (FVR #99):
every photo's queued / running / unavailable jobs, plus ready / failed /
expired ones touched in the last 3 days, newest first, discarded and
approved hidden, at most 100. DB only; the poller keeps rows current.
An optional status=a,b filter applies to both forms.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-11-01 22:14:27 -05:00

623 lines
20 KiB
Rust

//! HTTP surface of Enhance (#192). ImageApi relays everything, previews
//! included: the apps never talk to ComfyUI.
//!
//! - `GET /enhance/templates`
//! - `POST /enhance`
//! - `GET /enhance/prompts?templates=&limit=`
//! - `GET /enhance/jobs?path=`
//! - `GET /enhance/jobs/{id}`
//! - `GET /enhance/jobs/{id}/preview/{n}`
//! - `POST /enhance/jobs/{id}/approve`
//! - `POST /enhance/jobs/{id}/discard`
use std::path::{Component, Path, PathBuf};
use std::sync::{Arc, Mutex};
use std::time::Duration;
use actix_web::{HttpResponse, Responder, get, post, web};
use log::warn;
use serde::{Deserialize, Serialize};
use serde_json::json;
use super::config::PromptMode;
use super::service::{EnhanceError, EnhanceService, SourceFile};
use crate::data::Claims;
use crate::database::ExifDao;
use crate::database::enhance_dao::{EnhanceJobRow, STATUS_APPROVED, STATUS_READY};
use crate::error::ApiError;
use crate::libraries::{self, Library};
use crate::state::AppState;
use crate::utils::normalize_path;
use crate::versions::handlers::build_response;
use actix_web::ResponseError;
type ExifData = web::Data<Mutex<Box<dyn ExifDao>>>;
/// Enhance errors in the shared format (#202). The codes predate it and stay
/// as they are: clients may already branch on them.
fn error_response(e: &EnhanceError) -> HttpResponse {
use actix_web::http::StatusCode;
let (status, code) = match e {
EnhanceError::Disabled => (StatusCode::SERVICE_UNAVAILABLE, "disabled"),
EnhanceError::Unavailable => (StatusCode::SERVICE_UNAVAILABLE, "unavailable"),
EnhanceError::UnknownTemplate(_) => (StatusCode::NOT_FOUND, "unknown_template"),
EnhanceError::BadRequest(_) => (StatusCode::BAD_REQUEST, "bad_request"),
EnhanceError::Rejected(_) => (StatusCode::UNPROCESSABLE_ENTITY, "rejected"),
EnhanceError::Expired => (StatusCode::GONE, "expired"),
EnhanceError::NotFound => (StatusCode::NOT_FOUND, "not_found"),
EnhanceError::Unplaceable(_) => (StatusCode::BAD_GATEWAY, "unplaceable"),
EnhanceError::Upstream(_) => (StatusCode::BAD_GATEWAY, "upstream"),
EnhanceError::Internal(_) => {
return ApiError::internal(e).error_response();
}
};
ApiError::new(status, code, e.message()).error_response()
}
fn bad(msg: &str) -> HttpResponse {
error_response(&EnhanceError::BadRequest(msg.to_string()))
}
// ---- views --------------------------------------------------------------
#[derive(Debug, Serialize)]
struct PendingView {
/// Index for `preview/{n}` and `approve { n }`.
n: usize,
/// Path of the preview on this API.
preview: String,
approved: bool,
}
#[derive(Debug, Serialize)]
struct SavedView {
n: usize,
path: Option<String>,
library_id: Option<i32>,
}
#[derive(Debug, Serialize)]
pub struct JobView {
id: String,
status: String,
#[serde(skip_serializing_if = "Option::is_none")]
queue_position: Option<i32>,
#[serde(skip_serializing_if = "Option::is_none")]
error: Option<String>,
path: String,
library_id: i32,
template: String,
#[serde(skip_serializing_if = "Option::is_none")]
prompt: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
reference_path: Option<String>,
created_at: i64,
updated_at: i64,
/// Milliseconds ComfyUI ran the job; null until it has finished.
run_ms: Option<i64>,
pending: Vec<PendingView>,
saved: Vec<SavedView>,
}
pub fn job_view(job: &EnhanceJobRow) -> JobView {
let saved = EnhanceService::saved_of(job);
// An expired job's staged files are gone; don't offer them.
let pending = if job.status == STATUS_READY || job.status == STATUS_APPROVED {
EnhanceService::pending_of(job)
.iter()
.enumerate()
.map(|(n, _)| PendingView {
n,
preview: format!("/enhance/jobs/{}/preview/{n}", job.id),
approved: saved.iter().any(|s| s.n == n),
})
.collect()
} else {
Vec::new()
};
let queued = matches!(job.status.as_str(), "queued" | "running");
JobView {
id: job.id.clone(),
status: job.status.clone(),
queue_position: job.queue_position.filter(|_| queued),
error: job.error.clone(),
path: job.rel_path.clone(),
library_id: job.library_id,
template: job.template.clone(),
prompt: job.prompt.clone(),
reference_path: job.reference_path.clone(),
created_at: job.created_at,
updated_at: job.updated_at,
run_ms: job.run_ms,
pending,
saved: saved
.into_iter()
.map(|s| SavedView {
n: s.n,
path: s.rel_path,
library_id: s.library_id,
})
.collect(),
}
}
// ---- file resolution ----------------------------------------------------
fn safe_rel(rel: &str) -> bool {
!rel.is_empty()
&& Path::new(rel)
.components()
.all(|c| matches!(c, Component::Normal(_)))
}
/// An indexed, hashed file in a library.
fn resolve_file(
app_state: &AppState,
exif_dao: &Mutex<Box<dyn ExifDao>>,
path: &str,
library: Option<&str>,
) -> Result<SourceFile, HttpResponse> {
let library_id = match libraries::resolve_library_param_state(app_state, library) {
Ok(l) => l.map(|l| l.id),
Err(msg) => return Err(bad(&msg)),
};
let rel = normalize_path(path);
if !safe_rel(&rel) {
return Err(bad("invalid path"));
}
let mut rows = exif_dao
.lock()
.expect("Unable to lock ExifDao")
.get_exif_batch(library_id, std::slice::from_ref(&rel))
.map_err(|_| error_response(&EnhanceError::Internal("lookup failed".into())))?;
rows.sort_by_key(|r| r.id);
let Some(row) = rows.iter().find(|r| r.content_hash.is_some()) else {
return Err(error_response(&EnhanceError::NotFound));
};
let Some(lib) = app_state.library_by_id(row.library_id) else {
return Err(error_response(&EnhanceError::NotFound));
};
let abs = Path::new(&lib.root_path).join(&rel);
if !abs.is_file() {
return Err(error_response(&EnhanceError::NotFound));
}
Ok(SourceFile {
library_id: row.library_id,
rel_path: rel,
abs,
hash: row.content_hash.clone().unwrap_or_default(),
})
}
/// Library and relative path of an absolute file ComfyUI wrote. The job's own
/// library wins when roots overlap.
fn place_in_libraries(libs: &[Library], prefer: i32, abs: &Path) -> Option<(i32, String)> {
let mut ordered: Vec<&Library> = libs.iter().collect();
ordered.sort_by_key(|l| l.id != prefer);
for lib in ordered {
if let Ok(rest) = abs.strip_prefix(&lib.root_path) {
let rel = rest.to_str()?.replace('\\', "/");
if safe_rel(&rel) {
return Some((lib.id, rel));
}
}
}
None
}
// ---- templates ----------------------------------------------------------
#[derive(Serialize)]
struct TemplateView {
name: String,
label: String,
description: String,
accepts_prompt: bool,
prompt_mode: PromptMode,
/// Replace-mode templates need non-empty text (empty is a 400).
prompt_required: bool,
accepts_reference: bool,
default: bool,
/// Approximate output size; clients warn when the original is larger.
output_megapixels: f64,
warn_above_megapixels: f64,
/// `image` (a PNG version) or `video` (an MP4 clip, e.g. Animate).
output: super::template::OutputKind,
}
#[get("/enhance/templates")]
pub async fn get_templates(_: Claims, svc: web::Data<EnhanceService>) -> impl Responder {
let available = svc.available().await;
let reason = if !svc.is_configured() {
Some(super::service::MSG_DISABLED)
} else if !available {
Some(super::service::MSG_UNAVAILABLE)
} else {
None
};
let default = svc.default_template().map(str::to_string);
let templates: Vec<TemplateView> = svc
.templates()
.iter()
.map(|t| TemplateView {
name: t.cfg.name.clone(),
label: t.cfg.label.clone(),
description: t.cfg.description.clone(),
accepts_prompt: t.accepts_prompt(),
prompt_mode: t.cfg.prompt_mode,
prompt_required: t.prompt_required(),
accepts_reference: t.accepts_reference(),
default: default.as_deref() == Some(t.cfg.name.as_str()),
output_megapixels: t.cfg.output_megapixels,
warn_above_megapixels: svc.config.warn_above_megapixels,
output: t.output,
})
.collect();
let mut body = json!({
"available": available,
"default": default,
"warn_above_megapixels": svc.config.warn_above_megapixels,
"templates": templates,
});
if let Some(r) = reason {
body["reason"] = json!(r);
}
HttpResponse::Ok().json(body)
}
// ---- jobs ---------------------------------------------------------------
#[derive(Debug, Deserialize)]
pub struct LibraryQuery {
pub library: Option<String>,
}
#[derive(Debug, Deserialize)]
pub struct SubmitBody {
pub path: String,
pub template: String,
#[serde(default)]
pub prompt: Option<String>,
#[serde(default)]
pub reference_path: Option<String>,
#[serde(default)]
pub library: Option<String>,
}
#[post("/enhance")]
#[tracing::instrument(name = "enhance_submit", skip_all)]
pub async fn submit(
_: Claims,
query: web::Query<LibraryQuery>,
body: web::Json<SubmitBody>,
svc: web::Data<EnhanceService>,
app_state: web::Data<AppState>,
exif_dao: ExifData,
) -> impl Responder {
if !svc.is_configured() {
return error_response(&EnhanceError::Disabled);
}
let library = body.library.as_deref().or(query.library.as_deref());
let photo = match resolve_file(&app_state, &exif_dao, &body.path, library) {
Ok(f) => f,
Err(r) => return r,
};
let reference = match body
.reference_path
.as_deref()
.filter(|p| !p.trim().is_empty())
{
Some(p) => {
// The reference lives in the photo's own library.
let lib = app_state
.library_by_id(photo.library_id)
.map(|l| l.name.clone());
match resolve_file(&app_state, &exif_dao, p, lib.as_deref()) {
Ok(f) => Some(f),
Err(r) => return r,
}
}
None => None,
};
match svc
.submit(
&photo,
&body.template,
body.prompt.as_deref(),
reference.as_ref(),
)
.await
{
Ok(job) => HttpResponse::Ok().json(job_view(&job)),
Err(e) => error_response(&e),
}
}
#[get("/enhance/jobs/{id}")]
pub async fn get_job(
_: Claims,
id: web::Path<String>,
svc: web::Data<EnhanceService>,
) -> impl Responder {
match svc.get_job(&id) {
Ok(job) => HttpResponse::Ok().json(job_view(&job)),
Err(e) => error_response(&e),
}
}
#[derive(Debug, Deserialize)]
pub struct JobsQuery {
/// A photo (original or version): that photo's jobs. Omitted: every
/// photo's open jobs, for the jobs manager.
pub path: Option<String>,
pub library: Option<String>,
/// Comma-separated statuses to keep (e.g. `queued,running,ready`);
/// omitted = all. Applies to both forms.
pub status: Option<String>,
}
/// With `path`: a photo's recent jobs, newest first, so the viewer can pick up
/// an enhance in progress after reopening. Without: the jobs manager's list,
/// every photo's queued / running / unavailable jobs plus ready / failed /
/// expired ones from the last few days. Discarded and approved-and-done jobs
/// are not listed. DB only.
#[get("/enhance/jobs")]
pub async fn list_jobs(
_: Claims,
q: web::Query<JobsQuery>,
svc: web::Data<EnhanceService>,
app_state: web::Data<AppState>,
exif_dao: ExifData,
) -> impl Responder {
let jobs = match q.path.as_deref() {
Some(path) => {
let file = match resolve_file(&app_state, &exif_dao, path, q.library.as_deref()) {
Ok(f) => f,
Err(r) => return r,
};
svc.jobs_for_hash(&file.hash)
}
None => svc.open_jobs(),
};
let wanted: Option<Vec<&str>> = q.status.as_deref().map(|s| {
s.split(',')
.map(str::trim)
.filter(|s| !s.is_empty())
.collect()
});
match jobs {
Ok(jobs) => {
let views: Vec<JobView> = jobs
.iter()
.filter(|j| {
wanted
.as_ref()
.is_none_or(|w| w.contains(&j.status.as_str()))
})
.map(job_view)
.collect();
HttpResponse::Ok().json(json!({ "jobs": views }))
}
Err(e) => error_response(&e),
}
}
#[derive(Debug, Deserialize)]
pub struct PromptsQuery {
/// Comma-separated template names; omitted = all templates.
pub templates: Option<String>,
pub limit: Option<i64>,
}
#[derive(Debug, Serialize)]
struct PromptView {
prompt: String,
template: String,
last_used: i64,
uses: i64,
}
/// Prompts used before, newest first, so the app can offer them again for a
/// series of similar photos. DB only: not library-scoped, and answered even
/// when ComfyUI is down or Enhance is off (then simply from old rows).
#[get("/enhance/prompts")]
pub async fn recent_prompts(
_: Claims,
q: web::Query<PromptsQuery>,
svc: web::Data<EnhanceService>,
) -> impl Responder {
let names: Option<Vec<String>> = q.templates.as_deref().map(|t| {
t.split(',')
.map(str::trim)
.filter(|n| !n.is_empty())
.map(str::to_string)
.collect()
});
// `templates=` with nothing in it means "no filter", like omitting it.
let names = names.filter(|n| !n.is_empty());
let limit = q.limit.unwrap_or(10).clamp(1, 50);
match svc.recent_prompts(names.as_deref(), limit) {
Ok(rows) => {
let prompts: Vec<PromptView> = rows
.into_iter()
.map(|r| PromptView {
prompt: r.prompt,
template: r.template,
last_used: r.last_used,
uses: r.uses,
})
.collect();
HttpResponse::Ok().json(json!({ "prompts": prompts }))
}
Err(e) => error_response(&e),
}
}
#[get("/enhance/jobs/{id}/preview/{n}")]
pub async fn preview(
_: Claims,
path: web::Path<(String, usize)>,
svc: web::Data<EnhanceService>,
) -> impl Responder {
let (id, n) = path.into_inner();
let job = match svc.get_job(&id) {
Ok(j) => j,
Err(e) => return error_response(&e),
};
match svc.preview(&job, n).await {
Ok((content_type, bytes)) => HttpResponse::Ok()
.content_type(content_type)
.insert_header(("Cache-Control", "private, max-age=3600"))
.body(bytes),
Err(e) => error_response(&e),
}
}
#[post("/enhance/jobs/{id}/discard")]
pub async fn discard(
_: Claims,
id: web::Path<String>,
svc: web::Data<EnhanceService>,
) -> impl Responder {
match svc.discard(&id) {
Ok(job) => HttpResponse::Ok().json(job_view(&job)),
Err(e) => error_response(&e),
}
}
#[derive(Debug, Deserialize)]
pub struct ApproveBody {
#[serde(default)]
pub n: usize,
}
/// Ask ComfyUI to write result `n` into the library, ingest that file now,
/// and answer with the version (an `/image/versions` `versions[]` entry).
#[post("/enhance/jobs/{id}/approve")]
#[tracing::instrument(name = "enhance_approve", skip_all)]
pub async fn approve(
_: Claims,
id: web::Path<String>,
body: Option<web::Json<ApproveBody>>,
svc: web::Data<EnhanceService>,
app_state: web::Data<AppState>,
exif_dao: ExifData,
) -> impl Responder {
let n = body.map(|b| b.n).unwrap_or(0);
let job = match svc.get_job(&id) {
Ok(j) => j,
Err(e) => return error_response(&e),
};
let libs = app_state.libraries.clone();
let prefer = job.library_id;
let place = move |abs: &Path| place_in_libraries(&libs, prefer, abs);
let (job, saved) = match svc.approve(&id, n, &place).await {
Ok(v) => v,
Err(e) => return error_response(&e),
};
let (Some(library_id), Some(rel)) = (saved.library_id, saved.rel_path.clone()) else {
return error_response(&EnhanceError::Unplaceable(
"the saved file isn't in a library".into(),
));
};
let Some(library) = app_state.library_by_id(library_id).cloned() else {
return error_response(&EnhanceError::Unplaceable("unknown library".into()));
};
let abs = PathBuf::from(&library.root_path).join(&rel);
// The file was written by another machine over a share; give a lagging
// mount a moment to show it.
for _ in 0..25 {
if abs.is_file() {
break;
}
tokio::time::sleep(Duration::from_millis(200)).await;
}
if !abs.is_file() {
return error_response(&EnhanceError::Unplaceable(format!(
"ComfyUI saved {rel} but it isn't visible to ImageApi yet; it will appear after the next scan"
)));
}
let version_dao = app_state.version_dao.clone();
let exif_arc: Arc<Mutex<Box<dyn ExifDao>>> = exif_dao.clone().into_inner();
let (lib2, rel2, abs2) = (library.clone(), rel.clone(), abs.clone());
let ingested =
web::block(move || ingest_saved(&lib2, &rel2, &abs2, &exif_arc, &version_dao)).await;
let hash = match ingested {
Ok(Some(h)) => h,
_ => {
warn!("enhance: ingest of {rel} failed");
return HttpResponse::Accepted().json(json!({
"linked": false,
"path": rel,
"library_id": library_id,
"job": job_view(&job),
}));
}
};
// Which version row did that become?
let entry = {
let mut dao = app_state
.version_dao
.lock()
.expect("Unable to lock VersionDao");
let original = match dao.get_link(&hash) {
Ok(Some(link)) => link.original_hash,
_ => job.original_hash.clone(),
};
build_response(dao.as_mut(), &original, job.library_id)
.ok()
// An approved clip (Animate) is listed under `clips`.
.and_then(|r| {
r.versions
.into_iter()
.chain(r.clips)
.find(|v| v.hash == hash)
})
};
match entry {
Some(v) => HttpResponse::Ok().json(v),
None => HttpResponse::Accepted().json(json!({
"linked": false,
"path": rel,
"library_id": library_id,
"job": job_view(&job),
})),
}
}
/// Register the saved file (if the watcher hasn't yet) and link it as a
/// version. Idempotent. Returns the file's content hash.
pub fn ingest_saved(
library: &Library,
rel: &str,
abs: &Path,
exif_dao: &Arc<Mutex<Box<dyn ExifDao>>>,
version_dao: &Arc<Mutex<Box<dyn crate::database::VersionDao>>>,
) -> Option<String> {
let rel_s = rel.to_string();
let existing = exif_dao
.lock()
.expect("Unable to lock ExifDao")
.get_exif_batch(Some(library.id), std::slice::from_ref(&rel_s))
.ok()?;
let file = vec![(abs.to_path_buf(), rel_s.clone())];
if existing.is_empty() {
crate::watcher::register_new_files(library, file.clone(), exif_dao);
}
let linked = crate::versions::link_new_files(library, &file, exif_dao, version_dao)
+ crate::versions::sweep_name_links(library, exif_dao, version_dao);
if linked > 0 {
crate::duplicates::invalidate_caches();
}
exif_dao
.lock()
.expect("Unable to lock ExifDao")
.get_exif_batch(Some(library.id), std::slice::from_ref(&rel_s))
.ok()?
.into_iter()
.find_map(|r| r.content_hash)
}