From cdf46c2e8739e5ac4a1de97f6a8978a3a7dc5d9a Mon Sep 17 00:00:00 2001 From: phantomptr Date: Fri, 16 Oct 2026 06:45:12 -0700 Subject: [PATCH 1/4] Engine: /api/activity reports running jobs and recent endings (#457) A process-wide hub fed from set_job's terminal write and the unified install's finish path: running transfers and installs with their progress, and a ring of the last 64 endings with a monotonic seq so the app's background watcher sees each one once. POST /api/activity/label lets the client name a job. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01K3cPwhp3UrzayBfyy9PDcN --- .../crates/ps5upload-engine/src/activity.rs | 448 ++++++++++++++++++ .../ps5upload-engine/src/install/mod.rs | 30 +- engine/crates/ps5upload-engine/src/lib.rs | 42 ++ .../crates/ps5upload-engine/src/telemetry.rs | 1 + 4 files changed, 517 insertions(+), 4 deletions(-) create mode 100644 engine/crates/ps5upload-engine/src/activity.rs diff --git a/engine/crates/ps5upload-engine/src/activity.rs b/engine/crates/ps5upload-engine/src/activity.rs new file mode 100644 index 00000000..9ec26948 --- /dev/null +++ b/engine/crates/ps5upload-engine/src/activity.rs @@ -0,0 +1,448 @@ +//! What the engine is working on and what just ended, for the app's background watcher (#457). +//! +//! The renderer used to be the only thing that noticed a job finish, and a hidden or frozen +//! WebView notices nothing. The Tauri host polls `GET /api/activity` instead: running jobs +//! (transfers and installs) with their progress, plus a ring of recently finished ones with a +//! monotonic `seq` so a poller sees each ending once. Labels ("Astro Bot", a folder name) come +//! from the client through `POST /api/activity/label`; the engine only knows ids and paths. + +use std::collections::{HashMap, VecDeque}; +use std::sync::{Mutex, OnceLock}; + +use serde::{Deserialize, Serialize}; + +/// Finished entries kept for pollers. A poller one second behind needs a handful; 64 covers a +/// watcher that was paused for a while without growing with the session. +pub const RING_CAP: usize = 64; +/// Labels for jobs that never reported an ending (a label posted for an unknown id). +const LABELS_CAP: usize = 256; + +#[derive(Debug, Clone, Serialize, PartialEq)] +pub struct Active { + pub id: String, + pub kind: String, + pub label: String, + pub bytes_done: u64, + pub bytes_total: u64, + pub started_at_ms: u64, +} + +#[derive(Debug, Clone, Serialize, PartialEq)] +pub struct Finished { + pub seq: u64, + pub id: String, + pub kind: String, + pub label: String, + pub ok: bool, + /// The user stopped it: nothing to tell them. + pub cancelled: bool, + pub error: Option, + pub bytes: u64, + pub elapsed_ms: u64, +} + +#[derive(Debug, Clone, Default)] +struct Tag { + kind: Option, + label: Option, + /// Insertion order, for dropping the oldest past `LABELS_CAP`. + at: u64, +} + +#[derive(Default)] +struct Hub { + seq: u64, + tags: HashMap, + tag_counter: u64, + ring: VecDeque, +} + +impl Hub { + fn tag_mut(&mut self, id: &str) -> &mut Tag { + if !self.tags.contains_key(id) && self.tags.len() >= LABELS_CAP { + if let Some(oldest) = self + .tags + .iter() + .min_by_key(|(_, t)| t.at) + .map(|(k, _)| k.clone()) + { + self.tags.remove(&oldest); + } + } + self.tag_counter += 1; + let at = self.tag_counter; + self.tags.entry(id.to_string()).or_insert(Tag { + at, + ..Tag::default() + }) + } + + fn kind_label(&self, id: &str, default_kind: &str) -> (String, String) { + let t = self.tags.get(id); + ( + t.and_then(|t| t.kind.clone()) + .unwrap_or_else(|| default_kind.to_string()), + t.and_then(|t| t.label.clone()).unwrap_or_default(), + ) + } + + fn set_kind(&mut self, id: &str, kind: &str) { + let t = self.tag_mut(id); + if t.kind.is_none() { + t.kind = Some(kind.to_string()); + } + } + + fn set_label(&mut self, id: &str, kind: Option<&str>, label: &str) { + // The job may have ended before the label arrived (a tiny upload): name its entry. + if let Some(e) = self.ring.iter_mut().find(|e| e.id == id) { + e.label = label.to_string(); + return; + } + let t = self.tag_mut(id); + t.label = Some(label.to_string()); + if let Some(k) = kind.filter(|k| !k.is_empty()) { + t.kind = Some(k.to_string()); + } + } + + fn snapshot(&self, since: u64, active: Vec) -> serde_json::Value { + let finished: Vec<&Finished> = self.ring.iter().filter(|e| e.seq > since).collect(); + serde_json::json!({ "seq": self.seq, "active": active, "finished": finished }) + } + + fn finish(&mut self, mut f: Finished) { + // A job can be written terminal twice (a late ticker, a retry of the final write); + // a poller must see its ending once. + if self.ring.iter().any(|e| e.id == f.id) { + return; + } + let (kind, label) = self.kind_label(&f.id, &f.kind); + self.tags.remove(&f.id); + self.seq += 1; + f.seq = self.seq; + f.kind = kind; + // The client's name beats the engine's fallback (an install's title id). + if !label.is_empty() { + f.label = label; + } + if self.ring.len() >= RING_CAP { + self.ring.pop_front(); + } + self.ring.push_back(f); + } +} + +fn hub() -> &'static Mutex { + static H: OnceLock> = OnceLock::new(); + H.get_or_init(|| Mutex::new(Hub::default())) +} + +fn lock() -> std::sync::MutexGuard<'static, Hub> { + hub().lock().unwrap_or_else(|e| e.into_inner()) +} + +/// The kind a transfer's telemetry shape (`file`, `dir`, `download`, …) is shown as. +pub fn kind_of_shape(shape: &str) -> &'static str { + match shape { + "download" | "link_download" => "download", + _ => "upload", + } +} + +/// Records what kind of job `id` is, unless the client already said. +pub(crate) fn set_kind(id: &str, kind: &str) { + lock().set_kind(id, kind); +} + +/// A readable name for a job, from the client. `kind` overrides the engine's guess. +pub fn set_label(id: &str, kind: Option<&str>, label: &str) { + lock().set_label(id, kind, label); +} + +/// A transfer (or any `JobState` job) reached `done` / `failed`; `state` is its snapshot. +pub(crate) fn transfer_ended(id: &str, state: &serde_json::Value) { + if let Some(f) = transfer_finished(id, state) { + lock().finish(f); + } +} + +/// An install job ended (the unified install's finish path). +pub(crate) fn install_ended(st: &crate::install::status::InstallStatus) { + lock().finish(install_finished(st)); +} + +/// The ending a terminal job snapshot describes; `None` for any other state. +fn transfer_finished(id: &str, state: &serde_json::Value) -> Option { + let status = state.get("status").and_then(|v| v.as_str()).unwrap_or(""); + let ok = status == "done"; + if !ok && status != "failed" { + return None; + } + let error = state + .get("error") + .and_then(|v| v.as_str()) + .map(str::to_string); + let reason = state + .get("error_reason") + .and_then(|v| v.as_str()) + .unwrap_or(""); + let err = error.as_deref().unwrap_or(""); + // Same rule as the job summaries (telemetry::result_of). + let cancelled = !ok + && (reason.contains("cancel") || err.contains("transfer_cancelled") || err == "cancelled"); + let num = |k: &str| state.get(k).and_then(|v| v.as_u64()).unwrap_or(0); + Some(Finished { + seq: 0, + id: id.to_string(), + kind: "job".into(), + label: String::new(), + ok, + cancelled, + error: if ok { None } else { error }, + bytes: num("bytes_sent"), + elapsed_ms: num("elapsed_ms"), + }) +} + +fn install_finished(st: &crate::install::status::InstallStatus) -> Finished { + use crate::install::status::{FailReason, Phase, Verdict}; + let ok = st.phase == Phase::Done && st.verdict != Some(Verdict::Failed); + Finished { + seq: 0, + id: st.job.clone(), + kind: "install".into(), + label: st.title_id.clone().unwrap_or_else(|| st.content_id.clone()), + ok, + cancelled: st.reason == Some(FailReason::Cancelled), + error: if ok { + None + } else { + Some(st.hint.clone().unwrap_or_else(|| "install failed".into())) + }, + bytes: st.metrics.served_bytes, + elapsed_ms: st.updated_at.saturating_sub(st.started_at) * 1000, + } +} + +/// The running install `st`, as an activity entry. +pub(crate) fn install_active(st: &crate::install::status::InstallStatus) -> Active { + let (kind, label) = lock().kind_label(&st.job, "install"); + Active { + id: st.job.clone(), + kind, + label: if label.is_empty() { + st.title_id.clone().unwrap_or_else(|| st.content_id.clone()) + } else { + label + }, + bytes_done: st.metrics.served_bytes, + bytes_total: st.metrics.total_bytes, + started_at_ms: st.started_at * 1000, + } +} + +/// The running transfer `id` with its progress, as an activity entry. +pub(crate) fn transfer_active(id: &str, done: u64, total: u64, started_at_ms: u64) -> Active { + let (kind, label) = lock().kind_label(id, "job"); + Active { + id: id.to_string(), + kind, + label, + bytes_done: done, + bytes_total: total, + started_at_ms, + } +} + +/// `{seq, active, finished}`: entries finished after `since`, oldest first. +pub fn snapshot(since: u64, active: Vec) -> serde_json::Value { + lock().snapshot(since, active) +} + +#[derive(Debug, Deserialize)] +pub struct SinceQuery { + #[serde(default)] + pub since: u64, +} + +#[derive(Debug, Deserialize)] +pub struct LabelReq { + pub job_id: String, + #[serde(default)] + pub kind: Option, + pub label: String, +} + +/// `POST /api/activity/label {job_id, kind?, label}`. +pub async fn label_handler(axum::Json(req): axum::Json) -> axum::response::Response { + use axum::response::IntoResponse; + if req.job_id.is_empty() { + return ( + axum::http::StatusCode::BAD_REQUEST, + axum::Json(serde_json::json!({"ok": false, "error": "job_id is required"})), + ) + .into_response(); + } + // A name for a notification, not a document: keep it short. + let label: String = req.label.trim().chars().take(120).collect(); + set_label(&req.job_id, req.kind.as_deref(), &label); + axum::Json(serde_json::json!({"ok": true})).into_response() +} + +#[cfg(test)] +mod tests { + use super::*; + use serde_json::json; + + fn end(h: &mut Hub, id: &str, state: serde_json::Value) { + if let Some(f) = transfer_finished(id, &state) { + h.finish(f); + } + } + + #[test] + fn a_finished_job_is_listed_once_with_its_label_and_kind() { + let mut h = Hub::default(); + h.set_kind("a", "upload"); + h.set_label("a", None, "Astro Bot"); + end( + &mut h, + "a", + json!({"status":"done","bytes_sent":10,"elapsed_ms":5}), + ); + end( + &mut h, + "a", + json!({"status":"done","bytes_sent":10,"elapsed_ms":5}), + ); + let v = h.snapshot(0, vec![]); + assert_eq!(v["seq"], 1); + let f = v["finished"].as_array().unwrap(); + assert_eq!(f.len(), 1); + assert_eq!(f[0]["label"], "Astro Bot"); + assert_eq!(f[0]["kind"], "upload"); + assert_eq!(f[0]["ok"], true); + assert_eq!(f[0]["bytes"], 10); + assert!(h.tags.is_empty(), "the label is dropped with the ending"); + // Nothing new after the seq a poller already has. + assert!(h.snapshot(1, vec![])["finished"] + .as_array() + .unwrap() + .is_empty()); + } + + #[test] + fn the_client_kind_wins_and_a_late_label_names_the_entry() { + let mut h = Hub::default(); + h.set_label("b", Some("download"), "Saves"); + h.set_kind("b", "upload"); + end( + &mut h, + "b", + json!({"status":"failed","error":"transfer_cancelled"}), + ); + end(&mut h, "c", json!({"status":"failed","error":"boom"})); + h.set_label("c", None, "Late"); + let v = h.snapshot(0, vec![]); + let f = v["finished"].as_array().unwrap(); + assert_eq!(f[0]["kind"], "download"); + assert_eq!(f[0]["cancelled"], true); + assert_eq!(f[1]["kind"], "job"); + assert_eq!(f[1]["label"], "Late"); + assert_eq!(f[1]["ok"], false); + assert_eq!(f[1]["cancelled"], false); + assert_eq!(f[1]["error"], "boom"); + } + + #[test] + fn running_states_are_not_endings() { + assert!(transfer_finished("r", &json!({"status":"running"})).is_none()); + } + + #[test] + fn the_ring_keeps_the_newest_and_seq_keeps_climbing() { + let mut h = Hub::default(); + for i in 0..(RING_CAP + 10) { + end(&mut h, &format!("j{i}"), json!({"status":"done"})); + } + let v = h.snapshot(0, vec![]); + let f = v["finished"].as_array().unwrap(); + assert_eq!(f.len(), RING_CAP); + assert_eq!(v["seq"], (RING_CAP + 10) as u64); + assert_eq!(f[0]["seq"], 11); + assert_eq!(f[RING_CAP - 1]["id"], format!("j{}", RING_CAP + 9)); + } + + #[test] + fn labels_for_jobs_that_never_end_are_bounded() { + let mut h = Hub::default(); + for i in 0..(LABELS_CAP + 5) { + h.set_label(&format!("x{i}"), None, "n"); + } + assert_eq!(h.tags.len(), LABELS_CAP); + assert!(h.tags.contains_key(&format!("x{}", LABELS_CAP + 4))); + assert!(!h.tags.contains_key("x0")); + } + + #[test] + fn an_install_ending_carries_its_verdict_and_the_clients_name() { + use crate::install::status::{FailReason, InstallStatus, Phase, Verdict}; + let mut h = Hub::default(); + let mut st = InstallStatus::new("1-1", "ps5", "UP0000-PPSA00000_00"); + st.title_id = Some("PPSA00000".into()); + st.phase = Phase::Done; + st.verdict = Some(Verdict::MayNotLaunch); + st.metrics.served_bytes = 7; + h.finish(install_finished(&st)); + let mut st2 = InstallStatus::new("1-2", "ps5", "c"); + st2.phase = Phase::Failed; + st2.verdict = Some(Verdict::Failed); + st2.reason = Some(FailReason::Cancelled); + h.set_label("1-2", None, "Jak X"); + h.finish(install_finished(&st2)); + let v = h.snapshot(0, vec![]); + let f = v["finished"].as_array().unwrap(); + assert_eq!(f[0]["kind"], "install"); + assert_eq!(f[0]["ok"], true); + assert_eq!(f[0]["label"], "PPSA00000"); + assert_eq!(f[0]["bytes"], 7); + assert_eq!(f[1]["ok"], false); + assert_eq!(f[1]["cancelled"], true); + assert_eq!(f[1]["label"], "Jak X"); + } + + // The process-wide hub through the label route. Other tests end jobs concurrently, so this + // only looks for its own id. + #[tokio::test] + async fn the_label_route_names_a_job_the_activity_snapshot_reports() { + let id = format!("route-{}", uuid::Uuid::new_v4()); + let r = label_handler(axum::Json(LabelReq { + job_id: id.clone(), + kind: Some("upload".into()), + label: format!(" {} ", "x".repeat(200)), + })) + .await; + assert_eq!(r.status(), axum::http::StatusCode::OK); + let active = transfer_active(&id, 5, 10, 1); + assert_eq!(active.kind, "upload"); + assert_eq!(active.label.chars().count(), 120); + transfer_ended(&id, &json!({"status":"done","bytes_sent":10})); + let v = snapshot(0, vec![]); + let mine: Vec<_> = v["finished"] + .as_array() + .unwrap() + .iter() + .filter(|e| e["id"] == id.as_str()) + .collect(); + assert_eq!(mine.len(), 1); + assert_eq!(mine[0]["kind"], "upload"); + let bad = label_handler(axum::Json(LabelReq { + job_id: String::new(), + kind: None, + label: "x".into(), + })) + .await; + assert_eq!(bad.status(), axum::http::StatusCode::BAD_REQUEST); + } +} diff --git a/engine/crates/ps5upload-engine/src/install/mod.rs b/engine/crates/ps5upload-engine/src/install/mod.rs index d2f1d811..8ce4ebe7 100644 --- a/engine/crates/ps5upload-engine/src/install/mod.rs +++ b/engine/crates/ps5upload-engine/src/install/mod.rs @@ -83,22 +83,33 @@ impl JobStore { // Taken before the jobs lock: `begin` locks the slots first, then the jobs. let holding: std::collections::HashSet = lock(&self.active).values().cloned().collect(); - let cid = { + // Read before the prune, which may drop this very status. + let ended = { let mut jobs = lock(&self.jobs); - let cid = jobs.get(job).map(|st| slot_key(&st.ps5_addr)); + let ended = jobs.get(job).cloned(); prune_finished(&mut jobs, KEEP_FINISHED, &holding); - cid + ended }; lock(&self.cancelled).remove(job); - if let Some(cid) = cid { + if let Some(st) = ended { + let cid = slot_key(&st.ps5_addr); let mut active = lock(&self.active); // only clear if this job still owns the slot if active.get(&cid).map(|j| j.as_str()) == Some(job) { active.remove(&cid); + // Once per job: a second finish finds the slot already released. + crate::activity::install_ended(&st); } } } + /// Statuses of the installs still holding a console's slot. + pub fn running(&self) -> Vec { + let ids: Vec = lock(&self.active).values().cloned().collect(); + let jobs = lock(&self.jobs); + ids.iter().filter_map(|j| jobs.get(j).cloned()).collect() + } + /// The install task ended without finishing the job (it panicked): a job still /// marked running is marked failed so the client stops waiting, and the slot is /// released. A job that finished normally was already released; nothing changes. @@ -1936,6 +1947,17 @@ mod tests { } /// Finishing forgets the stop request and keeps only the newest finished statuses. + #[test] + fn running_lists_only_installs_holding_a_slot() { + let jobs = JobStore::new(); + let a = jobs.begin("10.1.0.1").unwrap(); + let b = jobs.begin("10.1.0.2").unwrap(); + jobs.update(&b, |s| s.phase = Phase::Done); + jobs.finish(&b); + let running: Vec = jobs.running().into_iter().map(|s| s.job).collect(); + assert_eq!(running, vec![a]); + } + #[test] fn finish_prunes_old_statuses_and_stop_requests() { let jobs = JobStore::new(); diff --git a/engine/crates/ps5upload-engine/src/lib.rs b/engine/crates/ps5upload-engine/src/lib.rs index 8f02ca85..cb8b43bb 100644 --- a/engine/crates/ps5upload-engine/src/lib.rs +++ b/engine/crates/ps5upload-engine/src/lib.rs @@ -31,6 +31,7 @@ //! GET /api/ps5/volumes → list storage volumes detected by the payload //! GET /api/ps5/list-dir?path=... → list immediate children of a directory on PS5 +mod activity; mod ampr_trace; mod ava1_api; mod bundled_payload; @@ -1528,6 +1529,7 @@ pub(crate) fn set_job( } } if terminal { + activity::transfer_ended(&job_id.to_string(), &json); telemetry::on_state(job_id, &json); if let Some(e) = job_event(&job_id.to_string(), &json) { ps5upload_core::events::emit_event(e); @@ -9616,6 +9618,36 @@ async fn engine_version() -> impl IntoResponse { ) } +/// GET /api/activity?since= — running jobs and the endings after `since`, for the app's +/// background watcher (see `activity`). +async fn activity_handler( + State(state): State, + installs: pkg_install::PkgInstallStateHandle, + q: activity::SinceQuery, +) -> impl IntoResponse { + let mut active: Vec = { + let jobs = state.jobs.lock().unwrap_or_else(|e| e.into_inner()); + jobs.iter() + .filter_map(|(id, s)| match s { + JobState::Running { + started_at_ms, + bytes_sent, + total_bytes, + .. + } => Some(activity::transfer_active( + &id.to_string(), + *bytes_sent, + *total_bytes, + *started_at_ms, + )), + _ => None, + }) + .collect() + }; + active.extend(installs.jobs.running().iter().map(activity::install_active)); + Json(activity::snapshot(q.since, active)) +} + /// GET /api/jobs async fn list_jobs(State(state): State) -> impl IntoResponse { let jobs = state.jobs.lock().unwrap_or_else(|e| e.into_inner()); @@ -10026,8 +10058,18 @@ async fn run(cfg: EngineConfig) -> anyhow::Result<()> { })); } + let activity_installs = pkg_state.clone(); let app = Router::new() .route("/", get(ui_handler)) + .route( + "/api/activity", + get( + move |st: State, Query(q): Query| { + activity_handler(st, activity_installs.clone(), q) + }, + ), + ) + .route("/api/activity/label", post(activity::label_handler)) .route("/api/ps5/status", get(ps5_status)) .route("/api/ps5/port-check", get(ps5_port_check)) .route("/api/ps5/readiness", get(ps5_readiness)) diff --git a/engine/crates/ps5upload-engine/src/telemetry.rs b/engine/crates/ps5upload-engine/src/telemetry.rs index ae615892..c35017e3 100644 --- a/engine/crates/ps5upload-engine/src/telemetry.rs +++ b/engine/crates/ps5upload-engine/src/telemetry.rs @@ -55,6 +55,7 @@ fn held() -> &'static Mutex> { /// Marks a job as one that gets a record when it ends. `kind` is the transfer's shape /// (`file`, `dir`, `zip`, `7z`, `rar`, `file_list`, `download`, `relay`, …). pub(crate) fn tag(job_id: Uuid, kind: &'static str) { + crate::activity::set_kind(&job_id.to_string(), crate::activity::kind_of_shape(kind)); held() .lock() .unwrap_or_else(|e| e.into_inner()) From 413703be89df925262b359cc20b653075080a218 Mon Sep 17 00:00:00 2001 From: phantomptr Date: Fri, 16 Oct 2026 07:04:16 -0700 Subject: [PATCH 2/4] Host: background watcher and Android transfer service (#457) bg_activity.rs polls the engine's /api/activity every second from Rust, so a hidden or frozen WebView no longer decides whether work is running or has ended. A pure step() turns each snapshot into start/update/finished/stop actions: the first poll only records seq, an engine restart resets it, a renderer hold refreshed within 5 minutes counts as work, and nothing active for 30 s stops. Finished jobs are emitted to the page (bg-job-finished, plus a bg-tick heartbeat while work runs) and announced through the OS when the app is in the background, in the language the page sends. Android: a connectedDevice foreground service (dataSync as fallback) with a partial wake lock and a Wi-Fi lock keeps the process and its in-process engine alive, with an ongoing progress notification and a separate result channel. The Kotlin class, manifest entries and R8 keep rules are written by android-postinit-patch.sh. JNI_OnLoad and the context helpers move to android_jni.rs so both callers share them; app classes load through the context's class loader. The mobile engine start is now once per process, as Activity recreation can run setup again. Desktop: notifications through the notification plugin's Rust API when the main window is hidden, minimised or unfocused; backgroundThrottling disabled on the window (macOS 14+). Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01K3cPwhp3UrzayBfyy9PDcN --- client/src-tauri/src/android_bg.rs | 138 ++++ client/src-tauri/src/android_jni.rs | 88 +++ client/src-tauri/src/bg_activity.rs | 856 ++++++++++++++++++++++ client/src-tauri/src/commands/local_fs.rs | 56 +- client/src-tauri/src/engine_mobile.rs | 7 + client/src-tauri/src/lib.rs | 16 + client/src-tauri/tauri.conf.json | 3 +- scripts/release/android-postinit-patch.sh | 271 +++++++ 8 files changed, 1380 insertions(+), 55 deletions(-) create mode 100644 client/src-tauri/src/android_bg.rs create mode 100644 client/src-tauri/src/android_jni.rs create mode 100644 client/src-tauri/src/bg_activity.rs diff --git a/client/src-tauri/src/android_bg.rs b/client/src-tauri/src/android_bg.rs new file mode 100644 index 00000000..69e647b4 --- /dev/null +++ b/client/src-tauri/src/android_bg.rs @@ -0,0 +1,138 @@ +//! Android side of the background watcher (`bg_activity`): drives +//! `com.phantomptr.ps5upload.TransferService`, the foreground service that keeps the process +//! (and with it the in-process engine) alive while a transfer runs with the app in the +//! background, and posts the complete / failed notifications. The Kotlin class is written into +//! `gen/android` by `scripts/release/android-postinit-patch.sh`. +//! +//! Every call is best-effort: a failure is logged and the transfer carries on without the +//! service, as it did before it existed. + +use jni::objects::{GlobalRef, JClass, JObject, JValue}; +use jni::JNIEnv; +use std::sync::OnceLock; + +use crate::android_jni::{app_class, app_context, vm}; + +const SERVICE_CLASS: &str = "com.phantomptr.ps5upload.TransferService"; + +/// The service class, looked up once through the app's class loader. +static CLASS: OnceLock = OnceLock::new(); + +fn with_service( + what: &str, + f: impl for<'a> FnOnce(&mut JNIEnv<'a>, &JClass<'a>, &JObject<'a>) -> jni::errors::Result<()>, +) { + let result = (|| -> Result<(), String> { + let mut env = vm()? + .attach_current_thread() + .map_err(|e| format!("attach: {e}"))?; + let out = (|| { + let context = app_context(&mut env)?; + let global = match CLASS.get() { + Some(g) => g, + None => { + let class = app_class(&mut env, &context, SERVICE_CLASS)?; + let g = env + .new_global_ref(class) + .map_err(|e| format!("global ref: {e}"))?; + CLASS.get_or_init(|| g) + } + }; + let local = env + .new_local_ref(global.as_obj()) + .map_err(|e| format!("local ref: {e}"))?; + let class = JClass::from(local); + f(&mut env, &class, &context).map_err(|e| e.to_string()) + })(); + // Never hand the thread back with a pending exception. + if env.exception_check().unwrap_or(false) { + let _ = env.exception_describe(); + let _ = env.exception_clear(); + } + out + })(); + if let Err(e) = result { + eprintln!("[bg] TransferService.{what}: {e}"); + } +} + +/// Start the foreground service with its first progress line. `percent` < 0 = indeterminate. +pub fn start(title: &str, text: &str, percent: i32, channel: &str) { + with_service("start", |env, class, ctx| { + let (t, x, c) = ( + env.new_string(title)?, + env.new_string(text)?, + env.new_string(channel)?, + ); + env.call_static_method( + class, + "start", + "(Landroid/content/Context;Ljava/lang/String;Ljava/lang/String;ILjava/lang/String;)V", + &[ + JValue::Object(ctx), + JValue::Object(&t), + JValue::Object(&x), + JValue::Int(percent), + JValue::Object(&c), + ], + )?; + Ok(()) + }); +} + +/// Refresh the ongoing progress notification. +pub fn update(title: &str, text: &str, percent: i32) { + with_service("update", |env, class, ctx| { + let (t, x) = (env.new_string(title)?, env.new_string(text)?); + env.call_static_method( + class, + "update", + "(Landroid/content/Context;Ljava/lang/String;Ljava/lang/String;I)V", + &[ + JValue::Object(ctx), + JValue::Object(&t), + JValue::Object(&x), + JValue::Int(percent), + ], + )?; + Ok(()) + }); +} + +/// Post a complete / failed notification on the results channel. `id` keeps one per job. +pub fn finished(id: i32, title: &str, text: &str, ok: bool, channel: &str) { + with_service("finished", |env, class, ctx| { + let (t, x, c) = ( + env.new_string(title)?, + env.new_string(text)?, + env.new_string(channel)?, + ); + env.call_static_method( + class, + "finished", + "(Landroid/content/Context;ILjava/lang/String;Ljava/lang/String;ZLjava/lang/String;)V", + &[ + JValue::Object(ctx), + JValue::Int(id), + JValue::Object(&t), + JValue::Object(&x), + JValue::Bool(ok as u8), + JValue::Object(&c), + ], + )?; + Ok(()) + }); +} + +/// Stop the service and remove its progress notification. +pub fn stop() { + with_service("stop", |env, class, ctx| { + env.call_static_method( + class, + "stop", + "(Landroid/content/Context;)V", + &[JValue::Object(ctx)], + )?; + Ok(()) + }); +} diff --git a/client/src-tauri/src/android_jni.rs b/client/src-tauri/src/android_jni.rs new file mode 100644 index 00000000..bffc4f8d --- /dev/null +++ b/client/src-tauri/src/android_jni.rs @@ -0,0 +1,88 @@ +//! The JavaVM and Application context every Android JNI caller in this crate shares +//! (`commands::local_fs`, `android_bg`). `JNI_OnLoad` may be defined once per library, so it +//! lives here rather than with any one caller. + +use jni::objects::{JClass, JObject, JValue}; +use jni::{JNIEnv, JavaVM}; +use std::ffi::c_void; +use std::sync::OnceLock; + +/// JavaVM captured at native-library load. +/// +/// Tauri 2's Android backend (a Kotlin WryActivity that `loadLibrary`s +/// this `.so`) does NOT initialize the `ndk_context` crate's global — +/// tao carries its own internal glue and never calls +/// `ndk_context::initialize_android_context`. So the old +/// `ndk_context::android_context()` here always hit an uninitialized +/// global and aborted the whole app on first use ("android context was +/// not initialized"). We capture the VM ourselves in `JNI_OnLoad` +/// instead — the runtime calls it once when the library loads, before +/// any Tauri command can run, and neither wry nor tao defines it. +static JVM: OnceLock = OnceLock::new(); + +/// Android runtime entry point, invoked on `System.loadLibrary`. +#[no_mangle] +pub extern "system" fn JNI_OnLoad( + vm: *mut jni::sys::JavaVM, + _reserved: *mut c_void, +) -> jni::sys::jint { + if let Ok(vm) = unsafe { JavaVM::from_raw(vm) } { + let _ = JVM.set(vm); + } + jni::sys::JNI_VERSION_1_6 +} + +pub(crate) fn vm() -> Result<&'static JavaVM, String> { + JVM.get() + .ok_or_else(|| "Android JavaVM unavailable (JNI_OnLoad not run)".to_string()) +} + +/// The app's Application context — a `Context` we can use for +/// `getSystemService` and `startActivity(FLAG_ACTIVITY_NEW_TASK)`. +/// Fetched via `ActivityThread.currentApplication()` so we don't need a +/// captured Activity (or the unavailable `ndk_context`). +pub(crate) fn app_context<'local>(env: &mut JNIEnv<'local>) -> Result, String> { + let app = env + .call_static_method( + "android/app/ActivityThread", + "currentApplication", + "()Landroid/app/Application;", + &[], + ) + .map_err(|e| format!("currentApplication: {e}"))? + .l() + .map_err(|e| format!("application obj: {e}"))?; + if app.is_null() { + return Err("currentApplication() returned null".into()); + } + Ok(app) +} + +/// One of the app's own classes (`com.phantomptr.ps5upload.…`). `FindClass` on a thread Rust +/// attached searches the system class loader, which does not have them; the context's loader +/// does. `name` is dotted (`com.example.Foo`). +pub(crate) fn app_class<'local>( + env: &mut JNIEnv<'local>, + context: &JObject, + name: &str, +) -> Result, String> { + let loader = env + .call_method(context, "getClassLoader", "()Ljava/lang/ClassLoader;", &[]) + .map_err(|e| format!("getClassLoader: {e}"))? + .l() + .map_err(|e| format!("loader obj: {e}"))?; + let jname = env + .new_string(name) + .map_err(|e| format!("class name: {e}"))?; + let class = env + .call_method( + &loader, + "loadClass", + "(Ljava/lang/String;)Ljava/lang/Class;", + &[JValue::Object(&jname)], + ) + .map_err(|e| format!("loadClass {name}: {e}"))? + .l() + .map_err(|e| format!("class obj: {e}"))?; + Ok(JClass::from(class)) +} diff --git a/client/src-tauri/src/bg_activity.rs b/client/src-tauri/src/bg_activity.rs new file mode 100644 index 00000000..fe50500c --- /dev/null +++ b/client/src-tauri/src/bg_activity.rs @@ -0,0 +1,856 @@ +//! Background watcher (#457): notices from the host, not the WebView, that work is running or +//! has just ended. +//! +//! A hidden window throttles the renderer's timers (WebView2, WebKitGTK, a hidden WKWebView can +//! be suspended) and Android freezes a backgrounded app's process outright, so the renderer is +//! the wrong place to decide "a job finished, tell the person and start the next one". This +//! task polls the engine's `GET /api/activity` every second instead and: +//! +//! - on Android, runs a foreground service (`android_bg`) while anything is active, which keeps +//! the process and the in-process engine alive, with an ongoing progress notification; +//! - posts a complete / failed notification when the app is in the background; +//! - emits `bg-job-finished` (and a `bg-tick` heartbeat while work runs) so the renderer's queue +//! advances without waiting on a throttled timer. +//! +//! "Active" is an engine job running, or a renderer hold (queued items waiting their turn) +//! refreshed within the last 5 minutes; the service stops 30 s after nothing is active. +//! The decisions live in [`step`], which is pure and unit-tested. + +use std::collections::HashMap; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::{Mutex, OnceLock, RwLock}; +use std::time::Duration; + +use serde::{Deserialize, Serialize}; +use tauri::{AppHandle, Emitter, Runtime}; + +const POLL: Duration = Duration::from_secs(1); +/// A renderer hold counts this long after its last refresh (the renderer refreshes every 60 s): +/// a frozen renderer cannot keep the service up forever. +const HOLD_TTL_MS: u64 = 5 * 60 * 1000; +/// Nothing active for this long stops the service. Covers the gap between queue items. +const IDLE_GRACE_MS: u64 = 30 * 1000; +/// Weight of the newest sample in the speed average. +const EMA_ALPHA: f64 = 0.3; + +// ── Engine snapshot ────────────────────────────────────────────────────────── + +#[derive(Debug, Clone, Default, Deserialize, PartialEq)] +pub struct ActiveJob { + pub id: String, + #[serde(default)] + pub kind: String, + #[serde(default)] + pub label: String, + #[serde(default)] + pub bytes_done: u64, + #[serde(default)] + pub bytes_total: u64, +} + +#[derive(Debug, Clone, Default, Deserialize, Serialize, PartialEq)] +pub struct FinishedJob { + pub seq: u64, + pub id: String, + #[serde(default)] + pub kind: String, + #[serde(default)] + pub label: String, + #[serde(default)] + pub ok: bool, + #[serde(default)] + pub cancelled: bool, + #[serde(default)] + pub error: Option, + #[serde(default)] + pub bytes: u64, + #[serde(default)] + pub elapsed_ms: u64, +} + +#[derive(Debug, Clone, Default, Deserialize, PartialEq)] +pub struct Snapshot { + pub seq: u64, + #[serde(default)] + pub active: Vec, + #[serde(default)] + pub finished: Vec, +} + +// ── Strings (the renderer sends its language's; English until it does) ────── + +#[derive(Debug, Clone, Deserialize, PartialEq)] +#[serde(default)] +pub struct BgStrings { + pub uploading: String, + pub downloading: String, + pub installing: String, + pub working: String, + pub jobs_running: String, + pub waiting: String, + pub progress_of: String, + pub done_upload: String, + pub done_download: String, + pub done_install: String, + pub done_job: String, + pub failed_upload: String, + pub failed_download: String, + pub failed_install: String, + pub failed_job: String, + pub channel_progress: String, + pub channel_results: String, +} + +impl Default for BgStrings { + fn default() -> Self { + let s = |v: &str| v.to_string(); + Self { + uploading: s("Uploading {name}"), + downloading: s("Downloading {name}"), + installing: s("Installing {name}"), + working: s("Working on {name}"), + jobs_running: s("{n} jobs running"), + waiting: s("Waiting for the next item"), + progress_of: s("{done} of {total}"), + done_upload: s("Upload complete"), + done_download: s("Download complete"), + done_install: s("Installed"), + done_job: s("Finished"), + failed_upload: s("Upload failed"), + failed_download: s("Download failed"), + failed_install: s("Install failed"), + failed_job: s("Failed"), + channel_progress: s("Transfers in progress"), + channel_results: s("Finished transfers"), + } + } +} + +// ── The pure core ──────────────────────────────────────────────────────────── + +#[derive(Debug, Clone, PartialEq)] +pub struct Progress { + pub title: String, + pub text: String, + /// `None` when the total is unknown (indeterminate bar). + pub percent: Option, +} + +#[derive(Debug, Clone, PartialEq)] +pub enum Action { + StartFg(Progress), + Update(Progress), + Finished(FinishedJob), + StopFg, +} + +#[derive(Debug, Clone, Copy, Default, PartialEq)] +pub struct Rate { + bytes: u64, + at_ms: u64, + bps: f64, +} + +#[derive(Debug, Default)] +pub struct State { + /// The engine seq already seen; `None` until the first poll. + pub seq: Option, + /// The service (and progress notification) is up. + pub fg: bool, + idle_since_ms: Option, + last: Option, + rates: HashMap, + pub strings: BgStrings, +} + +/// One poll's decisions. The first poll only records `seq`, so a launch never re-announces +/// endings from before it; a `seq` lower than the one seen means the engine restarted, and +/// counting starts again from its value. +pub fn step(state: &mut State, snap: &Snapshot, hold: bool, now_ms: u64) -> Vec { + let mut out = Vec::new(); + match state.seq { + Some(seen) if snap.seq >= seen => { + for f in snap.finished.iter().filter(|f| f.seq > seen) { + out.push(Action::Finished(f.clone())); + } + } + _ => {} + } + state.seq = Some(snap.seq); + + update_rates(&mut state.rates, &snap.active, now_ms); + let active = !snap.active.is_empty() || hold; + if active { + state.idle_since_ms = None; + let p = format_progress(&snap.active, &state.rates, &state.strings); + if !state.fg { + state.fg = true; + out.push(Action::StartFg(p.clone())); + state.last = Some(p); + } else if state.last.as_ref() != Some(&p) { + out.push(Action::Update(p.clone())); + state.last = Some(p); + } + } else if state.fg { + match state.idle_since_ms { + None => { + state.idle_since_ms = Some(now_ms); + // The last job's 99% would otherwise sit there through the grace period. + let p = format_progress(&[], &state.rates, &state.strings); + if state.last.as_ref() != Some(&p) { + out.push(Action::Update(p.clone())); + state.last = Some(p); + } + } + Some(t) if now_ms.saturating_sub(t) >= IDLE_GRACE_MS => { + state.fg = false; + state.idle_since_ms = None; + state.last = None; + out.push(Action::StopFg); + } + Some(_) => {} + } + } + out +} + +/// Per-job speed, an exponential moving average of bytes over time between polls. Jobs that +/// are no longer running are forgotten. +fn update_rates(rates: &mut HashMap, active: &[ActiveJob], now_ms: u64) { + rates.retain(|id, _| active.iter().any(|j| &j.id == id)); + for j in active { + match rates.get_mut(&j.id) { + None => { + rates.insert( + j.id.clone(), + Rate { + bytes: j.bytes_done, + at_ms: now_ms, + bps: 0.0, + }, + ); + } + Some(r) => { + let dt = now_ms.saturating_sub(r.at_ms); + if dt == 0 { + continue; + } + let sample = j.bytes_done.saturating_sub(r.bytes) as f64 * 1000.0 / dt as f64; + r.bps = if r.bps == 0.0 { + sample + } else { + EMA_ALPHA * sample + (1.0 - EMA_ALPHA) * r.bps + }; + r.bytes = j.bytes_done; + r.at_ms = now_ms; + } + } + } +} + +/// `1.2 GB`, `38 MB`: SI units like the engine's MB/s, one decimal below 10. +pub fn fmt_bytes(n: f64) -> String { + const UNITS: [&str; 5] = ["B", "kB", "MB", "GB", "TB"]; + let mut v = n.max(0.0); + let mut u = 0; + while v >= 1000.0 && u < UNITS.len() - 1 { + v /= 1000.0; + u += 1; + } + if u == 0 || v >= 10.0 { + format!("{v:.0} {}", UNITS[u]) + } else { + format!("{v:.1} {}", UNITS[u]) + } +} + +/// The progress notification: "Uploading