Apps/ImageApi
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

libraries: PATCH /libraries/{id} with live-apply

Browse Source 
Adds an HTTP mutation surface for `libraries.enabled` and
`libraries.excluded_dirs`, replacing the SQL-only workflow noted in
CLAUDE.md. Apollo's Settings panel calls this from the LIBRARIES
section so the operator no longer has to ssh + sqlite3 to flip a
library off or edit its excludes.

Live-apply (no restart) via a new `live_libraries: Arc<RwLock<Vec<
Library>>>` field on AppState. The existing immutable `libraries`
Vec stays for hot-path handlers that only need stable id → root_path
lookups, avoiding a 19-call-site refactor. The watcher and
cleanup_orphaned_playlists now take the lock instead of a Vec
snapshot and re-read at the top of each tick, so `enabled` /
`excluded_dirs` changes are picked up within one
WATCH_QUICK_INTERVAL_SECONDS. The GET /libraries handler also reads
through the live view.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Cameron Cordes
2026-05-13 08:47:35 -04:00
parent 74bf693878
commit b3124437ec
4 changed files with 187 additions and 32 deletions
Download Patch File Download Diff File Expand all files Collapse all files

120
src/libraries.rs
View File

@@ -1,8 +1,9 @@
use actix_web::{HttpResponse, Responder, get, web::Data};
use actix_web::{HttpResponse, Responder, get, patch, web, web::Data};
use chrono::Utc;
use diesel::prelude::*;
use diesel::sqlite::SqliteConnection;
use log::{info, warn};
use serde::Deserialize;
use std::collections::HashMap;
use std::path::{Path, PathBuf};
use std::sync::{Arc, RwLock};
@@ -338,12 +339,19 @@ pub struct LibrariesResponse {
#[get("/libraries")]
pub async fn list_libraries(_claims: Claims, app_state: Data<AppState>) -> impl Responder {
// Read from the live view so a recent PATCH /libraries/{id} that
// flipped `enabled` or rewrote `excluded_dirs` surfaces immediately
// — the immutable `app_state.libraries` snapshot is stale once the
// first mutation lands.
let live_guard = app_state
.live_libraries
.read()
.unwrap_or_else(|e| e.into_inner());
let health_guard = app_state
.library_health
.read()
.unwrap_or_else(|e| e.into_inner());
let libraries = app_state
.libraries
let libraries = live_guard
.iter()
.map(|lib| LibraryStatus {
library: lib.clone(),
@@ -356,6 +364,112 @@ pub async fn list_libraries(_claims: Claims, app_state: Data<AppState>) -> impl
HttpResponse::Ok().json(LibrariesResponse { libraries })
}
/// Body for PATCH /libraries/{id}. Both fields are optional — omitting
/// one leaves it untouched. `excluded_dirs` is the same comma-separated
/// shape as the DB column; an empty string clears (writes NULL).
#[derive(Deserialize, Debug)]
pub struct PatchLibraryBody {
pub enabled: Option<bool>,
pub excluded_dirs: Option<String>,
}
/// Mutate one library row. The watcher reads `app_state.live_libraries`
/// at the top of each tick, so a successful PATCH is picked up within
/// one WATCH_QUICK_INTERVAL_SECONDS without restart — no separate
/// `apply_now` signal. Returns the updated `Library` so the caller can
/// render the new state without a follow-up GET.
///
/// Despite CLAUDE.md noting "Toggle via SQL; there is intentionally no
/// HTTP endpoint for library mutation", we now expose this for Apollo's
/// Settings panel. The single-user trust model hasn't changed; the
/// endpoint just removes the SSH-and-sqlite3 step.
#[patch("/libraries/{id}")]
pub async fn patch_library(
_claims: Claims,
path: web::Path<i32>,
body: web::Json<PatchLibraryBody>,
app_state: Data<AppState>,
) -> impl Responder {
let lib_id = path.into_inner();
let body = body.into_inner();
if body.enabled.is_none() && body.excluded_dirs.is_none() {
return HttpResponse::UnprocessableEntity().body("empty patch body");
}
let mut conn = crate::database::connect();
// Build the SET clause. Diesel's set() takes a tuple of assignments;
// we apply each field independently so an absent field doesn't get
// forced to NULL / its default.
let mut affected = 0usize;
if let Some(enabled) = body.enabled {
match diesel::update(libraries::table.filter(libraries::id.eq(lib_id)))
.set(libraries::enabled.eq(enabled))
.execute(&mut conn)
{
Ok(n) => affected = affected.max(n),
Err(e) => {
warn!("PATCH /libraries/{}: enabled update failed: {:?}", lib_id, e);
return HttpResponse::InternalServerError().body(format!("{}", e));
}
}
}
if let Some(raw) = body.excluded_dirs.as_deref() {
let trimmed = raw.trim();
// Empty / whitespace-only → NULL so the column reads back the
// same way a never-set library does (parse_excluded_dirs_column
// returns Vec::new() for NULL).
let stored: Option<&str> = if trimmed.is_empty() {
None
} else {
Some(trimmed)
};
match diesel::update(libraries::table.filter(libraries::id.eq(lib_id)))
.set(libraries::excluded_dirs.eq(stored))
.execute(&mut conn)
{
Ok(n) => affected = affected.max(n),
Err(e) => {
warn!(
"PATCH /libraries/{}: excluded_dirs update failed: {:?}",
lib_id, e
);
return HttpResponse::InternalServerError().body(format!("{}", e));
}
}
}
if affected == 0 {
return HttpResponse::NotFound().body(format!("library id {} not found", lib_id));
}
// Refresh the live view from the canonical DB state. Reloading the
// whole table (rather than mutating one entry in place) is cheap
// (handful of rows) and keeps the in-memory and DB views trivially
// consistent.
let fresh = load_all(&mut conn);
let updated = fresh.iter().find(|l| l.id == lib_id).cloned();
{
let mut live = app_state
.live_libraries
.write()
.unwrap_or_else(|e| e.into_inner());
*live = fresh;
}
match updated {
Some(lib) => {
info!(
"PATCH /libraries/{}: enabled={:?} excluded_dirs={:?} → applied",
lib_id, body.enabled, body.excluded_dirs
);
HttpResponse::Ok().json(lib)
}
None => HttpResponse::NotFound().body(format!("library id {} not found after update", lib_id)),
}
}
#[cfg(test)]
mod tests {
use super::*;

9
src/main.rs
View File

@@ -128,8 +128,12 @@ fn main() -> std::io::Result<()> {
// Start file watcher with playlist manager and preview generator
let playlist_mgr_for_watcher = app_state.playlist_manager.as_ref().clone();
let preview_gen_for_watcher = app_state.preview_clip_generator.as_ref().clone();
// Both background jobs read from the shared `live_libraries` lock
// so a PATCH /libraries/{id} that flips `enabled` or edits
// `excluded_dirs` takes effect on the next watcher tick / cleanup
// tick without an ImageApi restart.
watcher::watch_files(
app_state.libraries.clone(),
app_state.live_libraries.clone(),
playlist_mgr_for_watcher,
preview_gen_for_watcher,
app_state.face_client.clone(),
@@ -142,7 +146,7 @@ fn main() -> std::io::Result<()> {
// skips the whole cycle while any library is stale (a missing
// source is indistinguishable from a transiently-unmounted share).
watcher::cleanup_orphaned_playlists(
app_state.libraries.clone(),
app_state.live_libraries.clone(),
app_state.excluded_dirs.clone(),
app_state.library_health.clone(),
);
@@ -280,6 +284,7 @@ fn main() -> std::io::Result<()> {
.service(ai::rate_insight_handler)
.service(ai::export_training_data_handler)
.service(libraries::list_libraries)
.service(libraries::patch_library)
.add_feature(add_tag_services::<_, SqliteTagDao>)
.add_feature(knowledge::add_knowledge_services::<_, SqliteKnowledgeDao>)
.add_feature(personas::add_persona_services)

16
src/state.rs
View File

@@ -18,15 +18,25 @@ use crate::video::actors::{
};
use actix::{Actor, Addr};
use std::env;
use std::sync::{Arc, Mutex};
use std::sync::{Arc, Mutex, RwLock};
pub struct AppState {
pub stream_manager: Arc<Addr<StreamActor>>,
pub playlist_manager: Arc<Addr<VideoPlaylistManager>>,
pub preview_clip_generator: Arc<Addr<PreviewClipGenerator>>,
/// All configured media libraries. Ordered by `id` ascending; the first
/// entry is the primary library.
/// entry is the primary library. Frozen at startup — handlers that
/// only need stable lookup (id → name / root_path) read this. Mutable
/// flags (`enabled`, `excluded_dirs`) reflect their startup values;
/// for live state see [`AppState::live_libraries`].
pub libraries: Vec<Library>,
/// Live view of the libraries table, shared mutably between the
/// watcher (which reads it at the top of each tick to honour the
/// latest `enabled` / `excluded_dirs`) and the PATCH /libraries/{id}
/// handler (which writes it on a successful mutation). The split
/// from [`AppState::libraries`] is deliberate: handlers that only
/// look up by id don't need to take a lock per request.
pub live_libraries: Arc<RwLock<Vec<Library>>>,
/// Per-library availability snapshot. Updated by the file watcher at
/// the top of each tick via `libraries::refresh_health`. HTTP handlers
/// read it (e.g. `/libraries` surfacing). See "Library availability
@@ -112,11 +122,13 @@ impl AppState {
);
let library_health = libraries::new_health_map(&libraries_vec);
let live_libraries = Arc::new(RwLock::new(libraries_vec.clone()));
Self {
stream_manager,
playlist_manager: Arc::new(video_playlist_manager.start()),
preview_clip_generator: Arc::new(preview_clip_generator.start()),
libraries: libraries_vec,
live_libraries,
library_health,
base_path,
thumbnail_path,

74
src/watcher.rs
View File

@@ -16,7 +16,7 @@
use std::collections::{HashMap, HashSet};
use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex};
use std::sync::{Arc, Mutex, RwLock};
use std::time::{Duration, SystemTime};
use actix::Addr;
@@ -42,9 +42,13 @@ use crate::thumbnails;
use crate::video;
use crate::video::actors::{GeneratePreviewClipMessage, QueueVideosMessage, VideoPlaylistManager};
/// Clean up orphaned HLS playlists and segments whose source videos no longer exist
/// Clean up orphaned HLS playlists and segments whose source videos no longer exist.
///
/// `libs_lock` is the shared live view of the libraries table — read at the
/// top of each cleanup pass so a PATCH /libraries/{id} that disables or
/// re-mounts a library is picked up without a restart.
pub fn cleanup_orphaned_playlists(
libs: Vec<libraries::Library>,
libs_lock: Arc<RwLock<Vec<libraries::Library>>>,
excluded_dirs: Vec<String>,
library_health: libraries::LibraryHealthMap,
) {
@@ -60,16 +64,25 @@ pub fn cleanup_orphaned_playlists(
info!("Starting orphaned playlist cleanup job");
info!(" Cleanup interval: {} seconds", cleanup_interval_secs);
info!(" Playlist directory: {}", video_path);
for lib in &libs {
info!(
" Checking sources under '{}' at {}",
lib.name, lib.root_path
);
{
let libs = libs_lock.read().unwrap_or_else(|e| e.into_inner());
for lib in libs.iter() {
info!(
" Checking sources under '{}' at {}",
lib.name, lib.root_path
);
}
}
loop {
std::thread::sleep(Duration::from_secs(cleanup_interval_secs));
// Fresh snapshot per tick so a PATCH /libraries/{id} that
// disabled a library (or rewrote its excluded_dirs) is
// honoured immediately.
let libs: Vec<libraries::Library> =
libs_lock.read().unwrap_or_else(|e| e.into_inner()).clone();
// Safety gate: skip the cleanup cycle if any library is
// stale. A missing source video on a stale library is
// indistinguishable from a transient unmount, and the
@@ -211,7 +224,7 @@ pub fn cleanup_orphaned_playlists(
}
pub fn watch_files(
libs: Vec<libraries::Library>,
libs_lock: Arc<RwLock<Vec<libraries::Library>>>,
playlist_manager: Addr<VideoPlaylistManager>,
preview_generator: Addr<video::actors::PreviewClipGenerator>,
face_client: crate::ai::face_client::FaceClient,
@@ -247,11 +260,14 @@ pub fn watch_files(
or APOLLO_API_BASE_URL to enable)"
);
}
for lib in &libs {
info!(
" Watching library '{}' (id={}) at {}",
lib.name, lib.id, lib.root_path
);
{
let libs = libs_lock.read().unwrap_or_else(|e| e.into_inner());
for lib in libs.iter() {
info!(
" Watching library '{}' (id={}) at {}",
lib.name, lib.id, lib.root_path
);
}
}
// Create DAOs for tracking processed files
@@ -306,18 +322,21 @@ pub fn watch_files(
// below; no ingest runs here, just the health update + log.
// Disabled libraries skip the probe entirely — they should
// never enter the health map (treated as out-of-scope).
for lib in &libs {
if !lib.enabled {
continue;
{
let libs = libs_lock.read().unwrap_or_else(|e| e.into_inner());
for lib in libs.iter() {
if !lib.enabled {
continue;
}
let context = opentelemetry::Context::new();
let had_data = exif_dao
.lock()
.expect("exif_dao poisoned")
.count_for_library(&context, lib.id)
.map(|n| n > 0)
.unwrap_or(false);
libraries::refresh_health(&library_health, lib, had_data);
}
let context = opentelemetry::Context::new();
let had_data = exif_dao
.lock()
.expect("exif_dao poisoned")
.count_for_library(&context, lib.id)
.map(|n| n > 0)
.unwrap_or(false);
libraries::refresh_health(&library_health, lib, had_data);
}
loop {
@@ -330,6 +349,11 @@ pub fn watch_files(
let is_full_scan = since_last_full.as_secs() >= full_interval_secs;
// Fresh snapshot per tick — picks up PATCH /libraries/{id}
// mutations to `enabled` / `excluded_dirs` without restart.
let libs: Vec<libraries::Library> =
libs_lock.read().unwrap_or_else(|e| e.into_inner()).clone();
for lib in &libs {
// Operator kill switch: a disabled library is invisible
// to the watcher entirely. No probe, no ingest, no