luna/catacomb
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

WebSocket job progress + mobile-responsive web UI

Browse source 
WebSocket progress
- New `/ws/progress` route using axum 0.7's `ws` feature. Upgrades to
  a WebSocket that streams the same `ProgressResponse` payload as the
  HTTP polling endpoint, pushed instantly whenever a job moves.
- `WebState.progress_tx` is a `tokio::sync::broadcast::Sender<String>`
  (capacity 16, lossy — Lagged subscribers resync on the next tick).
- A background tokio task ticks every 500 ms while any download is
  running or queued, every 5 s when idle. Skips the JSON encode when
  no subscribers are listening, so a single-user install pays zero
  CPU for the broadcast when the browser tab is closed.
- JS replaces the per-interval HTTP poll with a WebSocket connection.
  The polling path stays as fallback for mobile carriers / reverse
  proxies that strip WS — when the socket closes or fails to open,
  schedulePoll() resumes immediately and a reconnect attempt fires
  after 5 s.
- Initial snapshot is sent right after subscribe so the UI is live
  before the next tick.

Mobile-responsive web UI
- Existing `@media(max-width:640px)` block tightened:
  - Hide the H1, AGPL footer notice, sort dropdown, and library-size
    counter — they don't fit and aren't actionable.
  - Header buttons get min 34×34 hit targets (Apple HIG / Material).
  - `.card-foot` wraps and grows to 34×34 buttons so the five action
    icons (play / watched / favourite / bookmark / waiting) stay
    tappable instead of cramming into thumb-unfriendly 22 px chips.
  - Modals go full-screen (no 10 px page-margin) so the iPhone safe
    area doesn't crop content.
  - Footer (URL bar + download button) wraps onto two rows so neither
    squeezes the other off-screen.
  - Bulk-actions toolbar wraps and gets bigger buttons.
  - Per-job stdout preview hides on small screens.
- New `@media(max-width:380px)` collapses the grid to single column
  for very narrow phones.

55 unit tests still pass.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
Luna 2026-05-27 00:59:48 -07:00
parent de122817af
commit 9e64297da6
4 changed files with 251 additions and 15 deletions
Download patch file Download diff file Expand all files Collapse all files

111
src/web.rs
View file

@ -109,6 +109,12 @@ pub struct WebState {
/// Per-IP failure tracker for [`post_login`]. Each entry is the number of
/// recent failures and the instant the lockout (if any) expires.
pub login_attempts: Mutex<HashMap<std::net::IpAddr, LoginAttempt>>,
/// Push channel for `/ws/progress` subscribers. A background tokio task
/// ticks every 500 ms while jobs are active and broadcasts a fresh
/// [`ProgressResponse`] snapshot here; the WebSocket handler forwards
/// each message to its client. The `/api/progress` HTTP endpoint stays
/// available as a fallback for clients that can't open a socket.
pub progress_tx: tokio::sync::broadcast::Sender<String>,
}
/// Failed-login tracking entry. After [`LOGIN_LOCKOUT_AFTER`] failures from
@ -996,6 +1002,68 @@ async fn get_music(State(state): State<Arc<WebState>>) -> impl IntoResponse {
Json(serde_json::json!({ "tracks": track_infos }))
}
/// `GET /ws/progress` — WebSocket upgrade that streams the same
/// [`ProgressResponse`] payload as the polling HTTP endpoint, pushed
/// from the background broadcast task. Clients should treat the JSON
/// payload identically to `/api/progress`.
///
/// On socket close (auth lost, network drop, server restart) the JS
/// client falls back to polling `/api/progress` so users with broken
/// WebSocket setups (some reverse proxies, certain mobile carriers)
/// still see updates.
async fn ws_progress(
State(state): State<Arc<WebState>>,
ws: axum::extract::ws::WebSocketUpgrade,
) -> axum::response::Response {
ws.on_upgrade(move |socket| ws_progress_handler(socket, state))
}
async fn ws_progress_handler(
mut socket: axum::extract::ws::WebSocket,
state: Arc<WebState>,
) {
use axum::extract::ws::Message;
// Subscribe before sending the initial snapshot so we don't miss a
// broadcast that fires between the snapshot build and the subscribe.
let mut rx = state.progress_tx.subscribe();
// Initial snapshot so the UI shows the current state immediately
// (without waiting up to 5s for the next idle broadcast tick).
let initial = {
let mut dl = state.downloader.lock().unwrap();
dl.poll();
let queued = dl.pending_snapshots().into_iter()
.map(|(label, url)| QueuedSnapshot { label, url })
.collect::<Vec<_>>();
ProgressResponse {
jobs: WebState::job_snapshots(&dl),
queued,
max_concurrent: dl.max_concurrent,
}
};
if let Ok(json) = serde_json::to_string(&initial) {
if socket.send(Message::Text(json)).await.is_err() { return; }
}
loop {
tokio::select! {
msg = rx.recv() => match msg {
Ok(json) => {
if socket.send(Message::Text(json)).await.is_err() { break; }
}
// Subscriber lagged. Resync via the next broadcast tick.
Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => continue,
Err(_) => break,
},
// Drain inbound frames (mostly pongs / client closes) so the
// socket doesn't stall.
frame = socket.recv() => match frame {
Some(Ok(Message::Close(_))) | None => break,
Some(Err(_)) => break,
_ => {}
},
}
}
}
async fn get_progress(State(state): State<Arc<WebState>>) -> impl IntoResponse {
let mut dl = state.downloader.lock().unwrap();
dl.poll();
@ -2021,6 +2089,10 @@ async fn serve(config: Config, shutdown_rx: std::sync::mpsc::Receiver<()>) {
let password_required_initial = db.get_setting("password_hash").ok().flatten().is_some();
let flags = db.get_video_flags().unwrap_or_default();
// Capacity 16 is plenty — only a small number of browser tabs ever
// subscribe and the broadcast is lossy by design (subscribers that lag
// get a Lagged error and just resubscribe).
let (progress_tx, _initial_rx) = tokio::sync::broadcast::channel::<String>(16);
let state = Arc::new(WebState {
library: Mutex::new(library),
downloader: Mutex::new(downloader),
@ -2037,9 +2109,47 @@ async fn serve(config: Config, shutdown_rx: std::sync::mpsc::Receiver<()>) {
last_scheduled_check: Mutex::new(None),
password_required_cache: AtomicBool::new(password_required_initial),
library_version: AtomicU64::new(1),
progress_tx,
login_attempts: Mutex::new(HashMap::new()),
});
// Broadcast progress snapshots to WebSocket subscribers. Ticks fast
// (every 500 ms) while any download is active or queued; slow
// (every 5 s) when idle, just to keep clients refreshed if state
// changed via direct flag/option/folder edits.
let progress_state = state.clone();
tokio::spawn(async move {
loop {
// Pick interval based on whether there's anything to report on.
let interval_ms = {
let dl = progress_state.downloader.lock().unwrap();
if dl.any_running() || dl.pending_count() > 0 { 500 } else { 5_000 }
};
tokio::time::sleep(std::time::Duration::from_millis(interval_ms)).await;
// Skip broadcast when no one's listening — saves the JSON encode.
if progress_state.progress_tx.receiver_count() == 0 {
continue;
}
// Drive the same poll() that /api/progress does so the snapshot
// includes the latest stdout/stderr lines.
let snapshot = {
let mut dl = progress_state.downloader.lock().unwrap();
dl.poll();
let queued = dl.pending_snapshots().into_iter()
.map(|(label, url)| QueuedSnapshot { label, url })
.collect::<Vec<_>>();
ProgressResponse {
jobs: WebState::job_snapshots(&dl),
queued,
max_concurrent: dl.max_concurrent,
}
};
if let Ok(json) = serde_json::to_string(&snapshot) {
let _ = progress_state.progress_tx.send(json);
}
}
});
// Background scheduler — ticks every 60 s; runs channel checks when due.
let sched_state = state.clone();
tokio::spawn(async move {
@ -2081,6 +2191,7 @@ async fn serve(config: Config, shutdown_rx: std::sync::mpsc::Receiver<()>) {
.route("/", get(get_index))
.route("/api/library", get(get_library))
.route("/api/progress", get(get_progress))
.route("/ws/progress", get(ws_progress))
.route("/api/download", post(post_download))
.route("/api/watched/:id", post(post_watched))
.route("/api/videos/:id/flags/:flag", post(post_video_flag))