From 196f167873898fe2d8c11473c5f798774257f4d4 Mon Sep 17 00:00:00 2001 From: vicrodh <103399283+vicrodh@users.noreply.github.com> Date: Tue, 14 Jul 2026 17:47:33 -0600 Subject: [PATCH] feat(qbzd): copy QConnect glue with TODO(converge) markers and daemon trait impls Real-volume join report and reconnect credential re-resolution are daemon-copy fixes; desktop call sites untouched (OD2). cc --- crates/Cargo.lock | 3 + crates/qbzd/Cargo.toml | 4 + crates/qbzd/src/daemon.rs | 21 +- crates/qbzd/src/main.rs | 1 + crates/qbzd/src/qconnect/engine.rs | 198 +++++++ crates/qbzd/src/qconnect/mod.rs | 410 +++++++++++++ crates/qbzd/src/qconnect/remote_stream.rs | 289 +++++++++ crates/qbzd/src/qconnect/report.rs | 208 +++++++ crates/qbzd/src/qconnect/session.rs | 342 +++++++++++ crates/qbzd/src/qconnect/sink.rs | 341 +++++++++++ crates/qbzd/src/qconnect/transport.rs | 689 ++++++++++++++++++++++ 11 files changed, 2502 insertions(+), 4 deletions(-) create mode 100644 crates/qbzd/src/qconnect/engine.rs create mode 100644 crates/qbzd/src/qconnect/mod.rs create mode 100644 crates/qbzd/src/qconnect/remote_stream.rs create mode 100644 crates/qbzd/src/qconnect/report.rs create mode 100644 crates/qbzd/src/qconnect/session.rs create mode 100644 crates/qbzd/src/qconnect/sink.rs create mode 100644 crates/qbzd/src/qconnect/transport.rs diff --git a/crates/Cargo.lock b/crates/Cargo.lock index f201eb143..860591d2d 100644 --- a/crates/Cargo.lock +++ b/crates/Cargo.lock @@ -6002,6 +6002,7 @@ dependencies = [ "clap", "clap_complete", "dirs 6.0.0", + "futures-util", "libc", "log", "open", @@ -6019,6 +6020,7 @@ dependencies = [ "qconnect-transport-ws", "rand 0.10.1", "reqwest", + "rusqlite", "serde", "serde_json", "tempfile", @@ -6026,6 +6028,7 @@ dependencies = [ "tokio", "toml", "urlencoding", + "uuid 1.22.0", ] [[package]] diff --git a/crates/qbzd/Cargo.toml b/crates/qbzd/Cargo.toml index 509b27157..5745aece1 100644 --- a/crates/qbzd/Cargo.toml +++ b/crates/qbzd/Cargo.toml @@ -32,6 +32,10 @@ dirs = "6" rand = { workspace = true } log = { workspace = true } async-trait = { workspace = true } +# QConnect glue copies (T9) — mirror the desktop copies' direct imports +uuid = { version = "1", features = ["v4"] } # renderer-report action uuids +rusqlite = { version = "0.31", features = ["bundled"] } # daemon-root qconnect_settings.db KV +futures-util = { workspace = true } # remote_stream.rs feeder (byte-exact) # new direct deps (versions already in lock where noted) tiny_http = "0.12" # in lock via qbz-cast toml = "0.9" # in lock transitively diff --git a/crates/qbzd/src/daemon.rs b/crates/qbzd/src/daemon.rs index 55ba9ff3e..f085675e0 100644 --- a/crates/qbzd/src/daemon.rs +++ b/crates/qbzd/src/daemon.rs @@ -113,15 +113,28 @@ pub async fn run(roots: ProfileRoots, cfg: QbzdConfig, warns: Vec) -> Re ); log::info!("control API listening on {bind_addr}"); + // 12. QConnect (T9): mint the daemon's OWN device identity in the daemon-root + // KV, decide auto-connect from the persisted startup mode (cli_override = + // None so the KV that `qbzd qconnect enable|disable` writes is never + // shadowed), and — when enabled — connect-on-Ready with the bounded retry + // schedule. Reads NOTHING from qbzd.toml. Held to shut the session down + // ahead of playback (§8.2-1); it also clones `Arc`, so it must + // drop before `drop(booted)` (the #521 ordering). + let mut qconnect = crate::qconnect::start(booted.runtime.clone(), booted.shared.clone(), &roots); + // 13. park on SIGTERM/SIGINT. NO startup audio "hygiene": both candidate // fns are verified no-ops from a fresh process and re-adding them is the // documented skeptic-correction #1 trap (§8.1). wait_for_signal().await; - // ── Shutdown (§8.2, ordered). Stop the playback driver FIRST: it holds an - // Arc clone, so its task must finish (dropping that Arc) - // before `drop(booted)` can release the audio device ahead of the #521 - // pair. Signal, then join. + // ── Shutdown (§8.2, ordered). Step 1: disconnect the QConnect session (and + // stop its auto-connect watcher) BEFORE playback is stopped, then drop the + // handle so its Arc clone is released ahead of `drop(booted)`. + qconnect.shutdown().await; + drop(qconnect); + // Step 2: stop the playback driver. It holds an Arc clone, so its + // task must finish (dropping that Arc) before `drop(booted)` can release + // the audio device ahead of the #521 pair. Signal, then join. let _ = shutdown_tx.send(true); if let Err(e) = driver.await { log::warn!("driver task join failed: {e:?}"); diff --git a/crates/qbzd/src/main.rs b/crates/qbzd/src/main.rs index 12f3260c0..11342bf98 100644 --- a/crates/qbzd/src/main.rs +++ b/crates/qbzd/src/main.rs @@ -8,6 +8,7 @@ mod daemon; mod lock; mod login; mod paths; +mod qconnect; mod state; pub const API_VERSION: u32 = 1; // 02-cli-and-api.md §1.6 diff --git a/crates/qbzd/src/qconnect/engine.rs b/crates/qbzd/src/qconnect/engine.rs new file mode 100644 index 000000000..adda07bd2 --- /dev/null +++ b/crates/qbzd/src/qconnect/engine.rs @@ -0,0 +1,198 @@ +// TODO(converge: qconnect-glue) — copied from crates/qbz/src/qconnect_engine.rs @ c8ef2a1b; +// do not fix bugs here without fixing the source, and vice versa. +// +//! Qobuz Connect renderer engine for the qbzd daemon. +//! +//! Implements [`qconnect_app::QconnectRendererEngine`] over the daemon +//! `AppRuntime`'s `QbzCore` + `Player`, so qbzd becomes a QConnect renderer +//! that inherits the shared echo/cursor/materialize/shuffle orchestration in +//! `qconnect_app::renderer` instead of re-deriving it. +//! +//! The protected bit-perfect seams (`play_streaming_dynamic` / `play_data`) and +//! the HTTP feeder live here, impl-side, exactly as the Tauri `CoreBridge` impl +//! does; the probe-derived sample_rate/channels/bit_depth flow STRAIGHT into +//! `play_streaming_dynamic` (never defaulted, or hi-res remote playback silently +//! resamples). The feeder body is a near-verbatim port of the Tauri +//! `track_loading.rs` feeder, with `bridge.player()` -> `self.core().player()`; +//! the only deviation is the TLS backend — the crates workspace `reqwest` ships +//! `rustls-tls` (not `native-tls`), so the `.use_native_tls()` calls are dropped. +//! TLS is transport encryption only; the decoded audio bytes are identical, so +//! bit-perfect is unaffected. (If the Qobuz streaming CDN ever presents a cert +//! rustls rejects, add `native-tls` to qbzd's reqwest features.) +#![allow(dead_code)] + +use std::sync::Arc; + +use async_trait::async_trait; +use qbz_app::shell::AppRuntime; +use qbz_core::QbzCore; +use qbz_models::{Quality, QueueTrack, RepeatMode, Track}; +use qbz_player::PlaybackState; +use qconnect_app::QconnectRendererEngine; + +use crate::adapter::DaemonAdapter; + +/// QConnect renderer engine backed by the daemon `AppRuntime`. Holds the shared +/// runtime and forwards every trait method through `runtime.core()`; the async +/// feeder spawns on the ambient tokio runtime (`start_track_stream` is always +/// awaited from a runtime task). +pub struct DaemonRendererEngine { + runtime: Arc>, +} + +impl DaemonRendererEngine { + pub fn new(runtime: Arc>) -> Self { + Self { runtime } + } + + fn core(&self) -> &Arc> { + self.runtime.core() + } +} + +#[async_trait] +impl QconnectRendererEngine for DaemonRendererEngine { + // ---- transport (sync) ---- + fn resume(&self) -> Result<(), String> { + self.core().resume().map_err(|err| err.to_string()) + } + fn pause(&self) -> Result<(), String> { + self.core().pause().map_err(|err| err.to_string()) + } + fn stop(&self) -> Result<(), String> { + self.core().stop().map_err(|err| err.to_string()) + } + fn seek(&self, position_secs: u64) -> Result<(), String> { + self.core().seek(position_secs).map_err(|err| err.to_string()) + } + fn set_volume(&self, fraction: f32) -> Result<(), String> { + self.core().set_volume(fraction).map_err(|err| err.to_string()) + } + fn get_playback_state(&self) -> PlaybackState { + self.core().get_playback_state() + } + fn has_loaded_audio(&self) -> bool { + self.core().player().has_loaded_audio() + } + + // ---- queue / mode (async) ---- + async fn set_repeat_mode(&self, mode: RepeatMode) { + self.core().set_repeat_mode(mode).await + } + async fn set_shuffle(&self, enabled: bool) { + self.core().set_shuffle(enabled).await + } + async fn set_shuffle_flag(&self, enabled: bool) { + self.core().set_shuffle_with_order(enabled, None).await + } + async fn get_all_queue_tracks(&self) -> (Vec, Option) { + self.core().get_all_queue_tracks().await + } + async fn set_queue(&self, tracks: Vec, start_index: Option) { + self.core().set_queue(tracks, start_index).await + } + async fn set_queue_with_order( + &self, + tracks: Vec, + start_index: Option, + shuffle_enabled: bool, + shuffle_order: Option>, + ) { + self.core() + .set_queue_with_order(tracks, start_index, shuffle_enabled, shuffle_order) + .await + } + async fn clear_queue(&self, keep_current: bool) { + self.core().clear_queue(keep_current).await + } + async fn play_index(&self, index: usize) -> Option { + self.core().play_index(index).await + } + + // ---- catalog (async) ---- + async fn get_track(&self, track_id: u64) -> Result { + self.core() + .get_track(track_id) + .await + .map_err(|err| err.to_string()) + } + async fn get_tracks_batch(&self, track_ids: &[u64]) -> Result, String> { + self.core() + .get_tracks_batch(track_ids) + .await + .map_err(|err| err.to_string()) + } + + // ---- protected audio seam (the only protected touch) ---- + async fn start_track_stream( + &self, + track_id: u64, + quality: Quality, + duration_secs: u64, + start_position_secs: u64, + ) -> Result<(), String> { + let stream_url = self + .core() + .get_stream_url(track_id, quality) + .await + .map_err(|err| format!("resolve stream url for remote track {track_id}: {err}"))?; + + let player = self.core().player(); + match super::remote_stream::stream_remote_track_into_player( + &player, + track_id, + duration_secs, + start_position_secs, + &stream_url.url, + "QConnect", + ) + .await + { + Ok(()) => Ok(()), + Err(stream_err) => { + log::warn!( + "[QConnect] Streaming handoff unavailable for track {}: {}. Falling back to full download.", + track_id, + stream_err + ); + let audio_data = download_remote_audio(&stream_url.url).await?; + self.core() + .player() + .play_data(audio_data, track_id) + .map_err(|err| format!("play remote track {track_id}: {err}"))?; + Ok(()) + } + } + } + + fn current_output_format(&self) -> Option<(u32, u32)> { + let player = self.core().player(); + Some(( + player.state.get_sample_rate(), + player.state.get_bit_depth(), + )) + } +} + + +async fn download_remote_audio(url: &str) -> Result, String> { + let response = reqwest::Client::new() + .get(url) + .header("User-Agent", "Mozilla/5.0") + .send() + .await + .map_err(|err| format!("download remote audio request failed: {err}"))?; + + if !response.status().is_success() { + return Err(format!( + "download remote audio failed with status {}", + response.status() + )); + } + + let bytes = response + .bytes() + .await + .map_err(|err| format!("read remote audio bytes failed: {err}"))?; + Ok(bytes.to_vec()) +} diff --git a/crates/qbzd/src/qconnect/mod.rs b/crates/qbzd/src/qconnect/mod.rs new file mode 100644 index 000000000..aab0eed61 --- /dev/null +++ b/crates/qbzd/src/qconnect/mod.rs @@ -0,0 +1,410 @@ +// TODO(converge: qconnect-glue) — derived from crates/qbz/src/qconnect_service.rs @ 5d50158e +// (the connect/disconnect facade + startup auto-connect, UI stripped); +// do not fix bugs here without fixing the source, and vice versa. +// +//! Daemon QConnect service facade + boot-step-12 entry point. +//! +//! Composes the copied glue (engine / sink / session / report / transport / +//! remote_stream) into a headless connect flow that reproduces the desktop +//! `SlintQconnectService::connect` recipe (build transport -> one shared +//! sync-state Mutex -> sink -> `QconnectApp::new` -> `set_app` -> `connect` -> +//! subscribe transport events BEFORE the spawn -> spawn `run_session_loop` -> +//! `bootstrap_remote_presence`), minus every UI surface. Lifecycle transitions +//! latch into `DaemonShared.qconnect` so `/api/status` stays diagnosable. +//! +//! `start()` mints the daemon's OWN device identity in the daemon-root KV, reads +//! the effective startup mode (cli_override = None — never shadow the KV that +//! T11/T13 write), and, when auto-connect is on, spawns a connect-on-Ready task +//! with the bounded [2s, 5s, 15s, 30s] retry schedule. QConnect reads NOTHING +//! from qbzd.toml — only the daemon-root `qconnect_settings.db`. + +pub mod engine; +pub mod remote_stream; +pub mod report; +pub mod session; +pub mod sink; +pub mod transport; + +use std::path::PathBuf; +use std::sync::Arc; +use std::time::Duration; + +use qbz_app::shell::AppRuntime; +use qconnect_app::{ + compute_effective_startup, QconnectApp, QconnectAppEvent, QconnectEventSink, + QconnectLifecycleState, QconnectRemoteSyncState, QconnectSessionState, SessionLoopHost, +}; +use qconnect_transport_ws::{NativeWsTransport, WsTransportConfig}; +use tokio::sync::Mutex; +use tokio::task::JoinHandle; + +use crate::adapter::DaemonAdapter; +use crate::paths::ProfileRoots; +use crate::state::{AuthState, DaemonShared}; + +use self::engine::DaemonRendererEngine; +use self::session::{bootstrap_remote_presence, DaemonSessionLoopHost}; +use self::sink::{DaemonEventSink, DaemonQconnectApp}; + +type Runtime = Arc>; +type SharedState = Arc>; + +/// The live QConnect runtime for one connected session (app + its config + the +/// spawned event loop + the shared sync accumulator). +pub(crate) struct DaemonQconnectRuntime { + pub app: Arc, + /// Re-latched by `bootstrap_after_reconnect` on a credential re-resolve; + /// consumed by the full-reconnect path (T10) + status/endpoint reporting. + #[allow(dead_code)] + pub config: WsTransportConfig, + pub event_loop: JoinHandle<()>, + pub sync_state: Arc>, +} + +/// Connect-flow state, mirrored on the desktop `SlintQconnectInner`. `pub(crate)` +/// fields so `session::DaemonSessionLoopHost` can gate lifecycle + re-latch the +/// config + drop the runtime on reconnect-exhausted. +#[derive(Default)] +pub(crate) struct DaemonQconnectInner { + pub runtime: Option, + /// Latched connect/loop error; surfaced by `qbzd status` QConnect block (T11). + #[allow(dead_code)] + pub last_error: Option, + pub lifecycle_state: QconnectLifecycleState, +} + +/// Map a lifecycle state to the `/api/status` `qconnect.state` label + the +/// session-active flag, and latch it into `DaemonShared`. +fn latch_lifecycle_into_shared(shared: &SharedState, state: QconnectLifecycleState) { + let (label, active) = match state { + QconnectLifecycleState::Off => ("off", false), + QconnectLifecycleState::Connecting => ("connecting", false), + QconnectLifecycleState::Connected => ("connected", true), + QconnectLifecycleState::Reconnecting => ("retrying", false), + QconnectLifecycleState::Exhausted => ("exhausted", false), + }; + if let Ok(mut s) = shared.lock() { + s.qconnect.state = label.to_string(); + s.qconnect.session_active = active; + if matches!(state, QconnectLifecycleState::Reconnecting) { + s.qconnect.last_transport_reconnect = Some(unix_seconds_string()); + } + } +} + +fn unix_seconds_string() -> String { + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_secs().to_string()) + .unwrap_or_else(|_| "0".to_string()) +} + +/// Dedup + gate a lifecycle transition: only emit while a runtime is alive and +/// the state actually changes. Mirrors the desktop +/// `update_lifecycle_state_if_running`, plus the `DaemonShared` latch. +pub(crate) async fn update_lifecycle_state_if_running( + inner: &Arc>, + sink: &DaemonEventSink, + shared: &SharedState, + next: QconnectLifecycleState, +) { + let mut guard = inner.lock().await; + if guard.runtime.is_none() { + return; + } + if guard.lifecycle_state == next { + return; + } + guard.lifecycle_state = next; + drop(guard); + latch_lifecycle_into_shared(shared, next); + sink.on_event(QconnectAppEvent::LifecycleChanged { state: next }) + .await; +} + +/// The headless QConnect connect service. +pub struct DaemonQconnectService { + inner: Arc>, + runtime: Runtime, + shared: SharedState, + #[allow(dead_code)] // T11 (settings reload) re-reads the KV through this path. + settings_db: PathBuf, + custom_device_name: Arc>>, +} + +impl DaemonQconnectService { + /// Establish the QConnect session. Gated on an initialized API client (the + /// qws/createToken discovery needs it). Idempotent: a second call while a + /// runtime is alive (or a connect is in flight) is a no-op. + pub async fn connect(&self) -> Result<(), String> { + if !self.runtime.core().is_api_initialized().await { + return Err("Qobuz API is not initialized; cannot start Qobuz Connect".to_string()); + } + + // Claim the connect slot ATOMICALLY before the transport-config await, so + // two concurrent connect()s can't both build a runtime. A live runtime OR + // an in-flight `Connecting` both short-circuit to a no-op. + { + let mut guard = self.inner.lock().await; + if guard.runtime.is_some() + || guard.lifecycle_state == QconnectLifecycleState::Connecting + { + log::info!( + "[QConnect] connect() called while already {:?}; no-op", + guard.lifecycle_state + ); + return Ok(()); + } + guard.lifecycle_state = QconnectLifecycleState::Connecting; + guard.last_error = None; + } + latch_lifecycle_into_shared(&self.shared, QconnectLifecycleState::Connecting); + + let config = match transport::resolve_transport_config(&self.runtime).await { + Ok(config) => config, + Err(err) => { + let mut guard = self.inner.lock().await; + if guard.runtime.is_none() { + guard.lifecycle_state = QconnectLifecycleState::Off; + } + drop(guard); + latch_lifecycle_into_shared(&self.shared, QconnectLifecycleState::Off); + return Err(err); + } + }; + + let transport = Arc::new(NativeWsTransport::new()); + let sync_state = Arc::new(Mutex::new(QconnectRemoteSyncState::default())); + let engine = DaemonRendererEngine::new(Arc::clone(&self.runtime)); + let sink = Arc::new(DaemonEventSink::new(engine, Arc::clone(&sync_state))); + let app = Arc::new(QconnectApp::new( + Arc::clone(&transport), + Arc::clone(&sink), + Arc::clone(&sync_state), + )); + // Wire the owning app into the sink so it can emit reports + drive + // session-apply. + sink.set_app(&app); + + if let Err(err) = app.connect(config.clone()).await { + let mut guard = self.inner.lock().await; + guard.lifecycle_state = QconnectLifecycleState::Off; + let msg = format!("qconnect transport connect failed: {err}"); + guard.last_error = Some(msg.clone()); + drop(guard); + latch_lifecycle_into_shared(&self.shared, QconnectLifecycleState::Off); + return Err(msg); + } + + // Subscribe to transport events SYNCHRONOUSLY here — after connect() + // returns and BEFORE the spawn / any further await — so the receiver is + // live before the WS handshake emits Connected / Subscribed / + // SessionEstablished / SESSION_STATE. tokio broadcast has no replay; a + // receiver created inside the spawned loop would race + drop those. + let transport_rx = app.subscribe_transport_events(); + let idle_retry_active = config.reconnect_idle_retry_ms > 0; + let host: Arc = Arc::new(DaemonSessionLoopHost { + app: Arc::clone(&app), + sync_state: Arc::clone(&sync_state), + inner: Arc::clone(&self.inner), + sink: Arc::clone(&sink), + runtime: Arc::clone(&self.runtime), + shared: Arc::clone(&self.shared), + }); + let app_for_loop = Arc::clone(&app); + let event_loop = tokio::spawn(async move { + app_for_loop + .run_session_loop(host, transport_rx, idle_retry_active) + .await; + }); + + let runtime_app = Arc::clone(&app); + { + let mut guard = self.inner.lock().await; + guard.last_error = None; + guard.runtime = Some(DaemonQconnectRuntime { + app, + config, + event_loop, + sync_state, + }); + } + + let custom_name = self.custom_device_name.read().await.clone(); + if let Err(err) = bootstrap_remote_presence(&runtime_app, custom_name.clone()).await { + let _ = self.disconnect().await; + let mut guard = self.inner.lock().await; + guard.last_error = Some(format!("qconnect bootstrap failed: {err}")); + return Err(format!("qconnect bootstrap failed: {err}")); + } + + // Reflect the resolved device name in `/api/status`. + let effective_name = transport::resolve_qconnect_friendly_name(custom_name.as_deref()); + if let Ok(mut s) = self.shared.lock() { + s.qconnect.device_name = effective_name; + } + + Ok(()) + } + + /// Tear the QConnect session down. Always forces Off. Aborts AND joins the + /// event loop so its `Arc` clone drops before the daemon's + /// shutdown releases the audio device (§8.2 / #521 ordering). + pub async fn disconnect(&self) -> Result<(), String> { + let runtime = { + let mut guard = self.inner.lock().await; + guard.lifecycle_state = QconnectLifecycleState::Off; + guard.runtime.take() + }; + + if let Some(runtime) = runtime { + // Disarm any in-flight liveness watchdog + clear the session topology + // BEFORE aborting the loop, so a late event can't resurrect a stale + // active-renderer state. + { + let mut state = runtime.sync_state.lock().await; + state.watchdog_generation = state.watchdog_generation.wrapping_add(1); + state.session = QconnectSessionState::default(); + state.session_renderer_states.clear(); + } + if let Err(err) = runtime.app.disconnect().await { + let mut guard = self.inner.lock().await; + guard.last_error = Some(format!("qconnect disconnect failed: {err}")); + } + runtime.event_loop.abort(); + let _ = runtime.event_loop.await; + } + + if let Ok(mut s) = self.shared.lock() { + s.qconnect.state = "off".to_string(); + s.qconnect.session_active = false; + } + Ok(()) + } + + /// Wait until the daemon is Ready (logged in + API initialized), then attempt + /// `connect()` with the bounded [2s, 5s, 15s, 30s] retry schedule. Each + /// `connect()` re-resolves the transport config internally, so a transient + /// credential/network failure can clear on a later attempt. + async fn connect_on_ready(self: Arc) { + loop { + let logged_in = self + .shared + .lock() + .map(|s| s.auth == AuthState::LoggedIn) + .unwrap_or(false); + if logged_in && self.runtime.core().is_api_initialized().await { + break; + } + tokio::time::sleep(Duration::from_millis(500)).await; + } + + let schedule: [u64; 4] = [2_000, 5_000, 15_000, 30_000]; + for attempt in 0..=schedule.len() { + match self.connect().await { + Ok(()) => { + log::info!("[QConnect] auto-connect succeeded"); + return; + } + Err(err) => { + log::warn!("[QConnect] auto-connect attempt {} failed: {err}", attempt + 1); + } + } + match schedule.get(attempt) { + Some(delay_ms) => tokio::time::sleep(Duration::from_millis(*delay_ms)).await, + None => { + log::warn!( + "[QConnect] auto-connect gave up for this session after {} attempts", + attempt + 1 + ); + return; + } + } + } + } +} + +/// Owner handle held by the daemon boot for the process lifetime. Drives runtime +/// enable/disable (T11) and the ordered shutdown (§8.2-1). +pub struct QconnectHandle { + service: Arc, + watcher: Option>, +} + +impl QconnectHandle { + /// Connect on demand (T11 `qbzd qconnect enable`). + #[allow(dead_code)] + pub async fn connect(&self) -> Result<(), String> { + self.service.connect().await + } + + /// Disconnect on demand (T11 `qbzd qconnect disable`). + #[allow(dead_code)] + pub async fn disconnect(&self) -> Result<(), String> { + self.service.disconnect().await + } + + /// §8.2-1: stop the auto-connect watcher and disconnect the session BEFORE + /// the daemon stops playback. Aborts + joins the watcher and the event loop so + /// every `Arc` clone this handle owns drops ahead of + /// `drop(booted)` (the #521 clock-release ordering). + pub async fn shutdown(&mut self) { + if let Some(watcher) = self.watcher.take() { + watcher.abort(); + let _ = watcher.await; + } + let _ = self.service.disconnect().await; + } +} + +/// Boot step 12: wire QConnect. Mints the daemon's OWN device identity in the +/// daemon-root KV, decides auto-connect from the persisted startup mode +/// (`cli_override = None`, `last_known = None` — P0), latches the initial status, +/// and, when enabled, spawns the connect-on-Ready retry task. +pub fn start(runtime: Runtime, shared: SharedState, roots: &ProfileRoots) -> QconnectHandle { + let settings_db = roots.data.join("qconnect_settings.db"); + // Re-point device identity + KV at the daemon root (NEVER the desktop global). + transport::init_settings_db_path(settings_db.clone()); + + // Effective startup decision (Ready-state only). `cli_override` stays None: a + // `Some` would permanently shadow the KV store that `qbzd qconnect + // enable|disable` (T11) + the TUI (T13) write, making both dead controls. + // `last_known` is None in P0 (RememberLast resolves to off). + let mode = transport::load_startup_mode_at(&settings_db); + let should_auto_connect = compute_effective_startup(mode, None, None); + let custom_name = transport::load_device_name_at(&settings_db); + let effective_name = transport::resolve_qconnect_friendly_name(custom_name.as_deref()); + + // Latch the initial status so `/api/status` reflects the config before Ready. + if let Ok(mut s) = shared.lock() { + s.qconnect.enabled = should_auto_connect; + s.qconnect.device_name = effective_name; + s.qconnect.state = "off".to_string(); + s.qconnect.session_active = false; + } + + let service = Arc::new(DaemonQconnectService { + inner: Arc::new(Mutex::new(DaemonQconnectInner::default())), + runtime, + shared, + settings_db, + custom_device_name: Arc::new(tokio::sync::RwLock::new(custom_name)), + }); + + let watcher = if should_auto_connect { + log::info!( + "[QConnect] auto-connect enabled (startup mode = {}); waiting for Ready", + mode.as_str() + ); + let svc = Arc::clone(&service); + Some(tokio::spawn(async move { svc.connect_on_ready().await })) + } else { + log::info!( + "[QConnect] auto-connect disabled (startup mode = {})", + mode.as_str() + ); + None + }; + + QconnectHandle { service, watcher } +} diff --git a/crates/qbzd/src/qconnect/remote_stream.rs b/crates/qbzd/src/qconnect/remote_stream.rs new file mode 100644 index 000000000..d67c583f4 --- /dev/null +++ b/crates/qbzd/src/qconnect/remote_stream.rs @@ -0,0 +1,289 @@ +// TODO(converge: qconnect-glue) — copied from crates/qbz/src/remote_stream.rs @ b4450965; +// do not fix bugs here without fixing the source, and vice versa. +//! Shared HTTP streaming feeder. +//! +//! Ports the Tauri `track_loading.rs` progressive feeder verbatim: probe a +//! remote audio URL for size + FLAC format, open the player's progressive +//! streaming sink (`Player::play_streaming_dynamic`), then push the body to the +//! returned `BufferWriter` chunk-by-chunk as it arrives. Playback starts as soon +//! as the initial buffer fills — not after the whole file lands. +//! +//! `reqwest + BufferWriter` bound only, so it stays frontend-side and never +//! crosses the qconnect-app boundary. Used by BOTH the QConnect renderer +//! (`qconnect_engine.rs`) and the Plex playback path (`playback.rs`), so there +//! is exactly one feeder. +//! +//! BIT-PERFECT: `play_streaming_dynamic` decodes the same original bytes and +//! drives the PROTECTED device init from the decoded stream. The +//! `sample_rate`/`bit_depth` parsed here are only the streaming-config hints; +//! the audio backend (`pipewire_backend.rs`, `init_device`, `audio_settings.rs`) +//! is untouched. + +use std::time::Duration; + +use qbz_player::{BufferWriter, Player}; + +/// Format/size facts sniffed from a remote audio URL before streaming. +pub struct RemoteStreamInfo { + pub content_length: u64, + pub sample_rate: u32, + pub channels: u16, + pub bit_depth: u32, + pub speed_mbps: f64, +} + +/// Probe + open the progressive sink + spawn the background feeder. +/// +/// On success the player has begun buffering and `play_streaming_dynamic` will +/// start audio once the initial buffer fills; the body download runs in a +/// spawned task. Errors here mean the caller should fall back to a full +/// download (the probe or the sink open failed). +pub async fn stream_remote_track_into_player( + player: &Player, + track_id: u64, + duration_secs: u64, + start_position_secs: u64, + url: &str, + log_tag: &str, +) -> Result<(), String> { + let stream_info = probe_remote_stream_info(url).await?; + log::info!( + "[{}/STREAMING] Track {} - {:.2} MB, {}Hz, {} ch, {}-bit, {:.1} MB/s", + log_tag, + track_id, + stream_info.content_length as f64 / (1024.0 * 1024.0), + stream_info.sample_rate, + stream_info.channels, + stream_info.bit_depth, + stream_info.speed_mbps + ); + + let writer = player + .play_streaming_dynamic( + track_id, + stream_info.sample_rate, + stream_info.channels, + stream_info.bit_depth, + stream_info.content_length, + stream_info.speed_mbps, + duration_secs, + start_position_secs, + ) + .map_err(|err| format!("start streaming remote track {track_id}: {err}"))?; + + let url = url.to_string(); + let content_length = stream_info.content_length; + let log_tag = log_tag.to_string(); + tokio::spawn(async move { + if let Err(err) = + download_and_stream_remote_track(&url, writer, track_id, content_length, &log_tag).await + { + log::error!( + "[{}/STREAMING] Track {} failed while streaming: {}", + log_tag, + track_id, + err + ); + } + }); + + Ok(()) +} + +/// HEAD for content-length, then a small `Range: bytes=0-65535` GET to (a) +/// measure throughput and (b) parse the FLAC `STREAMINFO` block for the real +/// sample rate / channels / bit depth. Never defaults silently for FLAC (a +/// wrong sample rate would silently resample hi-res). +pub async fn probe_remote_stream_info(url: &str) -> Result { + use std::time::Instant; + + let client = reqwest::Client::builder() + .timeout(Duration::from_secs(30)) + .connect_timeout(Duration::from_secs(10)) + .build() + .map_err(|err| format!("create stream probe client: {err}"))?; + + let head_response = client + .head(url) + .header("User-Agent", "Mozilla/5.0") + .send() + .await + .map_err(|err| format!("probe HEAD request failed: {err}"))?; + + if !head_response.status().is_success() { + return Err(format!( + "probe HEAD request failed with status {}", + head_response.status() + )); + } + + let content_length = head_response + .headers() + .get("content-length") + .and_then(|value| value.to_str().ok()) + .and_then(|value| value.parse::().ok()) + .ok_or_else(|| "probe missing content-length header".to_string())?; + + let start_time = Instant::now(); + let range_response = client + .get(url) + .header("User-Agent", "Mozilla/5.0") + .header("Range", "bytes=0-65535") + .send() + .await + .map_err(|err| format!("probe range request failed: {err}"))?; + + if !range_response.status().is_success() { + return Err(format!( + "probe range request failed with status {}", + range_response.status() + )); + } + + let initial_bytes = range_response + .bytes() + .await + .map_err(|err| format!("read probe bytes failed: {err}"))?; + + let elapsed = start_time.elapsed(); + let speed_mbps = if elapsed.as_secs_f64() > 0.0 { + (initial_bytes.len() as f64 / elapsed.as_secs_f64()) / (1024.0 * 1024.0) + } else { + 10.0 + }; + + let (sample_rate, channels, bit_depth) = + if initial_bytes.len() >= 26 && initial_bytes.starts_with(b"fLaC") { + let sample_rate = ((initial_bytes[18] as u32) << 12) + | ((initial_bytes[19] as u32) << 4) + | ((initial_bytes[20] as u32) >> 4); + let channels = ((initial_bytes[20] >> 1) & 0x07) + 1; + let bit_depth = ((initial_bytes[20] & 0x01) << 4) | ((initial_bytes[21] >> 4) & 0x0F); + (sample_rate, channels as u16, (bit_depth + 1) as u32) + } else { + log::warn!("[remote-stream] Non-FLAC probe for remote handoff, using defaults"); + (44_100, 2, 16) + }; + + Ok(RemoteStreamInfo { + content_length, + sample_rate, + channels, + bit_depth, + speed_mbps, + }) +} + +/// Plain full-body GET → `bytes_stream()` loop → `writer.push_chunk` → +/// `writer.complete()`. No HTTP Range on the main GET (the `BufferedMediaSource` +/// buffers every pushed byte and serves seeks from the growing buffer). +pub async fn download_and_stream_remote_track( + url: &str, + writer: BufferWriter, + track_id: u64, + content_length: u64, + log_tag: &str, +) -> Result<(), String> { + use futures_util::StreamExt; + use std::time::Instant; + + struct FailGuard { + writer: BufferWriter, + armed: bool, + } + impl Drop for FailGuard { + fn drop(&mut self) { + if self.armed { + let _ = self + .writer + .error("remote stream aborted before completion".into()); + } + } + } + let mut guard = FailGuard { + writer, + armed: true, + }; + let writer = &guard.writer; + + let client = reqwest::Client::builder() + .connect_timeout(Duration::from_secs(10)) + .timeout(Duration::from_secs(300)) + .build() + .map_err(|err| format!("create remote streaming client: {err}"))?; + + let response = client + .get(url) + .header("User-Agent", "Mozilla/5.0") + .send() + .await + .map_err(|err| format!("start remote streaming request failed: {err}"))?; + + if !response.status().is_success() { + return Err(format!( + "remote streaming request failed with status {}", + response.status() + )); + } + + let mut bytes_received = 0u64; + let mut stream = response.bytes_stream(); + let start_time = Instant::now(); + let mut last_log_time = Instant::now(); + + while let Some(chunk_result) = stream.next().await { + let chunk = chunk_result.map_err(|err| format!("remote streaming chunk failed: {err}"))?; + bytes_received += chunk.len() as u64; + + if let Err(err) = writer.push_chunk(&chunk) { + log::error!( + "[{}/STREAMING] Failed to push chunk for track {}: {}", + log_tag, + track_id, + err + ); + guard.armed = false; + let _ = writer.error(format!("push_chunk failed: {err}")); + return Err(format!("push_chunk failed: {err}")); + } + + let now = Instant::now(); + if now.duration_since(last_log_time) >= Duration::from_secs(2) && content_length > 0 { + let progress = (bytes_received as f64 / content_length as f64) * 100.0; + let avg_speed = + (bytes_received as f64 / start_time.elapsed().as_secs_f64()) / (1024.0 * 1024.0); + log::info!( + "[{}/STREAMING] Track {} {:.1}% ({:.2}/{:.2} MB) @ {:.2} MB/s", + log_tag, + track_id, + progress, + bytes_received as f64 / (1024.0 * 1024.0), + content_length as f64 / (1024.0 * 1024.0), + avg_speed + ); + last_log_time = now; + } + } + + guard.armed = false; + if let Err(err) = writer.complete() { + log::error!( + "[{}/STREAMING] Failed to mark stream complete for track {}: {}", + log_tag, + track_id, + err + ); + let _ = writer.error(format!("complete failed: {err}")); + return Err(format!("complete failed: {err}")); + } + + log::info!( + "[{}/STREAMING] Track {} complete: {:.2} MB in {:.1}s", + log_tag, + track_id, + bytes_received as f64 / (1024.0 * 1024.0), + start_time.elapsed().as_secs_f64() + ); + + Ok(()) +} diff --git a/crates/qbzd/src/qconnect/report.rs b/crates/qbzd/src/qconnect/report.rs new file mode 100644 index 000000000..e7486a0b0 --- /dev/null +++ b/crates/qbzd/src/qconnect/report.rs @@ -0,0 +1,208 @@ +// TODO(converge: qconnect-glue) — copied from crates/qbz/src/qconnect_service.rs @ 5d50158e; +// do not fix bugs here without fixing the source, and vice versa. +// +//! Renderer playback-state report (the UI-free body of the desktop +//! `report_playback_state`, qconnect_service.rs:592). +//! +//! Daemon adaptation vs. the Slint copy (§1.4): the desktop `report_playback_state` +//! is a method on `SlintQconnectService` driven by the Slint playback POLL LOOP; +//! here it is a free function the T10 report tick calls on a tokio interval. No +//! behavior change — it still self-gates on `is_local_renderer_active`, resolves +//! current/next queue_item_id from the playing track, sends a +//! `RndrSrvrStateUpdated`, keeps the app's renderer position in sync, and reports +//! the live output format for the controller's quality badge. `position_ms` / +//! `duration_ms` are MILLISECONDS (the QConnect protocol unit). +#![allow(dead_code)] + +use std::sync::Arc; + +use qbz_app::shell::AppRuntime; +use qconnect_app::{ + is_local_renderer_active, QconnectFileAudioQualitySnapshot, QconnectRemoteSyncState, + RendererReport, RendererReportType, +}; +use serde_json::json; +use tokio::sync::Mutex; +use uuid::Uuid; + +use crate::adapter::DaemonAdapter; +use super::sink::DaemonQconnectApp; +use super::transport::BUFFER_STATE_OK; + +pub const QCONNECT_RENDERER_CHANNELS: i32 = 2; +const AUDIO_QUALITY_UNKNOWN: i32 = 0; +const AUDIO_QUALITY_MP3: i32 = 1; +const AUDIO_QUALITY_CD: i32 = 2; +const AUDIO_QUALITY_HIRES_L1: i32 = 3; +const AUDIO_QUALITY_HIRES_L2: i32 = 4; +const AUDIO_QUALITY_HIRES_L3: i32 = 5; + +/// Report this device's playback state to the cloud while the daemon is the +/// ACTIVE LOCAL renderer. Self-gates on `is_local_renderer_active` (no-op when a +/// PEER owns playback), resolves the current/next queue_item_id from the playing +/// track, sends a `RndrSrvrStateUpdated`, and keeps the app's renderer position +/// in sync. +pub async fn report_playback_state( + app: &Arc, + sync_state: &Arc>, + runtime: &Arc>, + playing_state: i32, + position_ms: i64, + duration_ms: i64, + track_id: u64, +) { + // Only report when WE are the active renderer. When a peer renderer owns + // playback (the daemon is acting as a controller) the renderer reports come + // from the peer, not us. + { + let state = sync_state.lock().await; + if !is_local_renderer_active(&state.session) { + return; + } + } + + let (current_qid, next_qid) = + resolve_queue_item_ids_by_track_id(app, sync_state, track_id).await; + let queue_version = app.queue_state_snapshot().await.version; + + let report = RendererReport::new( + RendererReportType::RndrSrvrStateUpdated, + Uuid::new_v4().to_string(), + queue_version, + json!({ + "playing_state": playing_state, + "buffer_state": BUFFER_STATE_OK, + "current_position": position_ms, + "duration": duration_ms, + "current_queue_item_id": current_qid, + "next_queue_item_id": next_qid, + "queue_version": { + "major": queue_version.major, + "minor": queue_version.minor + } + }), + ); + if let Err(err) = app.send_renderer_report_command(report).await { + log::warn!("[QConnect] Failed to report playback state: {err}"); + } + + if position_ms >= 0 { + app.update_renderer_position(position_ms as u64).await; + } + + // Report the live output format so the controller shows the correct quality + // badge (CD / Hi-Res). Reads the player's current output (sample_rate/ + // bit_depth); channels default to stereo. Both reports dedup internally in + // qconnect-app, so calling them every report tick is cheap. + let player = runtime.core().player(); + let sample_rate = player.state.get_sample_rate(); + let bit_depth = player.state.get_bit_depth(); + if let Some(snapshot) = + build_file_audio_quality_snapshot(sample_rate, bit_depth, QCONNECT_RENDERER_CHANNELS) + { + if let Err(err) = app + .report_file_audio_quality_if_changed(queue_version, snapshot) + .await + { + log::warn!("[QConnect] Failed to report file audio quality: {err}"); + } + if let Err(err) = app + .report_device_audio_quality_if_changed( + queue_version, + snapshot.sampling_rate, + snapshot.bit_depth, + snapshot.nb_channels, + ) + .await + { + log::warn!("[QConnect] Failed to report device audio quality: {err}"); + } + } +} + +/// Classify a (sample_rate, bit_depth) output into the QConnect AudioQuality +/// level. Pure mirror of the Tauri `classify_qconnect_audio_quality`. +fn classify_audio_quality(sample_rate: u32, bit_depth: u32) -> i32 { + if sample_rate == 0 || bit_depth == 0 { + AUDIO_QUALITY_UNKNOWN + } else if sample_rate >= 384_000 { + AUDIO_QUALITY_HIRES_L3 + } else if sample_rate >= 192_000 { + AUDIO_QUALITY_HIRES_L2 + } else if bit_depth > 16 || sample_rate > 48_000 { + AUDIO_QUALITY_HIRES_L1 + } else if sample_rate >= 44_100 { + AUDIO_QUALITY_CD + } else { + AUDIO_QUALITY_MP3 + } +} + +/// Build a file-audio-quality snapshot from the live output format, or None when +/// the format isn't known yet. Pure mirror of the Tauri +/// `build_qconnect_file_audio_quality_snapshot`. +fn build_file_audio_quality_snapshot( + sample_rate: u32, + bit_depth: u32, + nb_channels: i32, +) -> Option { + if sample_rate == 0 || bit_depth == 0 { + return None; + } + Some(QconnectFileAudioQualitySnapshot { + sampling_rate: sample_rate as i32, + bit_depth: bit_depth as i32, + nb_channels, + audio_quality: classify_audio_quality(sample_rate, bit_depth), + }) +} + +/// Resolve the current + next `queue_item_id` for a playing `track_id` from the +/// cloud queue snapshot, caching the result into the sync accumulator. Mirrors +/// the Tauri `resolve_queue_item_ids_by_track_id`. +async fn resolve_queue_item_ids_by_track_id( + app: &Arc, + sync_state: &Arc>, + track_id: u64, +) -> (Option, Option) { + let queue = app.queue_state_snapshot().await; + let (current_qid, next_qid, next_track_id) = + qconnect_app::queue_resolution::resolve_queue_item_ids_from_queue_state(&queue, track_id); + + if let Some(current_qid) = current_qid { + let mut state = sync_state.lock().await; + state.last_renderer_queue_item_id = Some(current_qid); + state.last_renderer_next_queue_item_id = next_qid; + state.last_renderer_track_id = Some(track_id); + state.last_renderer_next_track_id = next_track_id; + (Some(current_qid), next_qid) + } else { + (None, None) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn classify_audio_quality_matches_the_desktop_thresholds() { + assert_eq!(classify_audio_quality(0, 0), AUDIO_QUALITY_UNKNOWN); + assert_eq!(classify_audio_quality(44_100, 16), AUDIO_QUALITY_CD); + assert_eq!(classify_audio_quality(48_000, 16), AUDIO_QUALITY_CD); + assert_eq!(classify_audio_quality(96_000, 24), AUDIO_QUALITY_HIRES_L1); + assert_eq!(classify_audio_quality(192_000, 24), AUDIO_QUALITY_HIRES_L2); + assert_eq!(classify_audio_quality(384_000, 24), AUDIO_QUALITY_HIRES_L3); + assert_eq!(classify_audio_quality(22_050, 16), AUDIO_QUALITY_MP3); + } + + #[test] + fn snapshot_is_none_until_format_known() { + assert!(build_file_audio_quality_snapshot(0, 0, 2).is_none()); + let snap = build_file_audio_quality_snapshot(96_000, 24, 2).expect("known format"); + assert_eq!(snap.sampling_rate, 96_000); + assert_eq!(snap.bit_depth, 24); + assert_eq!(snap.nb_channels, 2); + assert_eq!(snap.audio_quality, AUDIO_QUALITY_HIRES_L1); + } +} diff --git a/crates/qbzd/src/qconnect/session.rs b/crates/qbzd/src/qconnect/session.rs new file mode 100644 index 000000000..66827dd55 --- /dev/null +++ b/crates/qbzd/src/qconnect/session.rs @@ -0,0 +1,342 @@ +// TODO(converge: qconnect-glue) — copied from crates/qbz/src/qconnect_service.rs @ 5d50158e; +// do not fix bugs here without fixing the source, and vice versa. +// +//! Daemon session-loop host + controller bootstrap + deferred renderer-join. +//! +//! [`DaemonSessionLoopHost`] implements the frontend-agnostic +//! `qconnect_app::SessionLoopHost` so the shared `run_session_loop` drives +//! lifecycle, reconnect bootstrap/resync, deferred renderer-join, and +//! reconnect-exhausted teardown through this adapter — exactly as +//! `SlintSessionLoopHost` does. +//! +//! Two DELIBERATE daemon-copy behavior fixes vs. the desktop (OD2 — desktop call +//! sites untouched; owner may fold these back into the shared path later): +//! (a) `deferred_renderer_join` reports the player's REAL volume instead of the +//! hardcoded `json!({"volume": 100})` (qconnect_service.rs:2711-2715; §7.4). +//! (b) `bootstrap_after_reconnect` RE-RESOLVES transport credentials via +//! `resolve_transport_config` so a reconnect after JWT expiry recovers +//! (IV3 — the WS transport reuses `last_config` and never re-resolves +//! `/qws/createToken`; design-input/qconnect-headless.md §7.6.7). +//! +//! The desktop's D5 offline-mode parks are dropped: qbzd has no offline MODE. + +use std::sync::Arc; + +use qbz_app::shell::AppRuntime; +use qconnect_app::renderer::PLAYING_STATE_STOPPED; +use qconnect_app::{ + QconnectLifecycleState, QconnectRemoteSyncState, QueueCommandType, RendererReport, + RendererReportType, SessionLoopHost, JOIN_SESSION_REASON_RECONNECTION, +}; +use serde_json::json; +use tokio::sync::Mutex; +use uuid::Uuid; + +use crate::adapter::DaemonAdapter; +use crate::state::DaemonShared; +use super::sink::{DaemonEventSink, DaemonQconnectApp}; +use super::transport::{ + default_qconnect_device_info, default_qconnect_device_info_with_name, resolve_transport_config, + QconnectJoinSessionRequest, AUDIO_QUALITY_HIRES_LEVEL2, BUFFER_STATE_OK, +}; +use super::{update_lifecycle_state_if_running, DaemonQconnectInner}; + +type Runtime = Arc>; + +/// Daemon-side implementation of the shared session-loop seams. Holds the handles +/// the loop reaches back into: the app (renderer join + state reads), the shared +/// sync accumulator, the service inner (lifecycle gating + teardown + config +/// re-latch), the sink (lifecycle emit), the runtime (track-duration + volume +/// reads), and the daemon shared state (status latching for `/api/status`). +pub struct DaemonSessionLoopHost { + pub app: Arc, + pub sync_state: Arc>, + pub inner: Arc>, + pub sink: Arc, + pub runtime: Runtime, + pub shared: Arc>, +} + +#[async_trait::async_trait] +impl SessionLoopHost for DaemonSessionLoopHost { + async fn update_lifecycle(&self, state: QconnectLifecycleState) { + update_lifecycle_state_if_running(&self.inner, &self.sink, &self.shared, state).await; + } + + async fn bootstrap_after_reconnect(&self) { + // FIX (b) (IV3, daemon-copy only): the WS transport reuses `last_config` + // on in-loop reconnects and never re-resolves `/qws/createToken`, so a + // reconnect after the JWT expires keeps failing. Re-resolve the transport + // credentials (fresh `/qws/createToken`) and latch the new config into the + // runtime so a subsequent full reconnect uses valid credentials. The + // desktop copy does NOT do this. + match resolve_transport_config(&self.runtime).await { + Ok(fresh) => { + let mut guard = self.inner.lock().await; + if let Some(rt) = guard.runtime.as_mut() { + rt.config = fresh; + } + log::info!("[QConnect] reconnect: re-resolved transport credentials (fresh JWT)"); + } + Err(err) => { + log::warn!("[QConnect] reconnect: credential re-resolve failed: {err}"); + } + } + if let Err(err) = bootstrap_remote_presence(&self.app, None).await { + log::error!("[QConnect] Re-bootstrap after reconnect failed: {err}"); + } + } + + async fn deferred_renderer_join(&self, session_uuid: String, reason: i32) { + deferred_renderer_join( + &self.app, + &self.sync_state, + &self.runtime, + &session_uuid, + reason, + ) + .await; + } + + async fn on_reconnect_exhausted( + &self, + attempts: u32, + last_reason: String, + idle_retry_active: bool, + ) -> bool { + { + let mut guard = self.inner.lock().await; + guard.lifecycle_state = QconnectLifecycleState::Exhausted; + guard.last_error = Some(format!( + "Reconnect attempts exhausted ({attempts}): {last_reason}" + )); + if !idle_retry_active { + // Legacy terminate path: drop the runtime so a fresh connect() + // succeeds. Dropping it detaches this task's own JoinHandle (fine + // — the loop breaks right after). + guard.runtime = None; + } + } + if let Ok(mut s) = self.shared.lock() { + s.qconnect.state = "exhausted".to_string(); + s.qconnect.session_active = false; + } + log::warn!("[QConnect] Reconnect exhausted ({attempts}): {last_reason}"); + // Daemon has no offline MODE, so the desktop's idle-retry offline park is + // dropped: keep idling (the transport's internal 60s idle rearm keeps + // trying) unless idle-retry is off, in which case break + drop. + !idle_retry_active + } + + async fn on_loop_error(&self, message: String) { + // Slint copy surfaced a toast here — daemon logs it. + log::error!("[QConnect] session loop error: {message}"); + } +} + +/// Controller-side bootstrap: JoinSession (works without a session_uuid) then ask +/// for the current queue state. The renderer-side join is deferred until the +/// server sends SESSION_STATE with a session_uuid (handled in the session loop). +/// Mirrors the Tauri `bootstrap_remote_presence`. +pub async fn bootstrap_remote_presence( + app: &Arc, + custom_device_name: Option, +) -> Result<(), String> { + let device_info = default_qconnect_device_info_with_name(custom_device_name.as_deref()); + + let join_payload = serde_json::to_value(QconnectJoinSessionRequest { + session_uuid: None, + device_info: Some(device_info), + }) + .map_err(|err| format!("serialize join_session bootstrap payload: {err}"))?; + + let join_command = app + .build_queue_command(QueueCommandType::CtrlSrvrJoinSession, join_payload) + .await; + let join_action_uuid = app + .send_queue_command(join_command) + .await + .map_err(|err| format!("send bootstrap ctrl_srvr_join_session failed: {err}"))?; + // JoinSession responds with session/renderer controller events not part of + // queue reducer correlation. Drop the pending slot so queue ops aren't blocked. + app.clear_pending_if_matches(&join_action_uuid).await; + + let ask_queue_command = app + .build_queue_command(QueueCommandType::CtrlSrvrAskForQueueState, json!({})) + .await; + let ask_action_uuid = app + .send_queue_command(ask_queue_command) + .await + .map_err(|err| format!("send bootstrap ask_for_queue_state failed: {err}"))?; + app.clear_pending_if_matches(&ask_action_uuid).await; + + log::info!( + "[QConnect] Bootstrap complete: controller joined, queue state requested. Renderer join deferred until session_uuid received." + ); + Ok(()) +} + +/// Deferred renderer join: called from the session loop when SESSION_STATE with a +/// session_uuid arrives. Idempotent per uuid (P1-8). Mirrors the Tauri +/// `deferred_renderer_join`, reading the current track duration via +/// `runtime.core().get_track` and the current volume via +/// `runtime.core().get_playback_state()`. +pub async fn deferred_renderer_join( + app: &Arc, + sync_state: &Arc>, + runtime: &Runtime, + session_uuid: &str, + join_reason: i32, +) { + let already_joined = { + let st = sync_state.lock().await; + st.last_joined_session_uuid.as_deref() == Some(session_uuid) + }; + if already_joined { + log::info!( + "[QConnect] Deferred join skipped (already joined session_uuid={session_uuid}); re-asking renderer state" + ); + if let Err(err) = app.ask_for_active_renderer_state().await { + log::warn!("[QConnect] Idempotent-join AskForRendererState failed: {err}"); + } + return; + } + + let device_info = default_qconnect_device_info(); + let queue_version_ref = app.queue_state_snapshot().await.version; + + log::info!("[QConnect] Deferred renderer join with session_uuid={session_uuid}"); + + // 1. Renderer JoinSession with session_uuid. + // Do NOT auto-steal the render on a fresh connect: join as an AVAILABLE + // renderer (is_active=false), not the active one. Only a post-drop + // RECONNECTION rejoins as active, so a network blip mid-render does not lose + // the render. + let join_as_active = join_reason == JOIN_SESSION_REASON_RECONNECTION; + let renderer_join_payload = json!({ + "session_uuid": session_uuid, + "device_info": serde_json::to_value(&device_info).unwrap_or_default(), + "is_active": join_as_active, + "reason": join_reason, + "initial_state": { + "playing_state": PLAYING_STATE_STOPPED, + "buffer_state": BUFFER_STATE_OK, + "current_position": 0, + "duration": 0, + "queue_version": { + "major": queue_version_ref.major, + "minor": queue_version_ref.minor + } + } + }); + let renderer_join_report = RendererReport::new( + RendererReportType::RndrSrvrJoinSession, + Uuid::new_v4().to_string(), + queue_version_ref, + renderer_join_payload, + ); + if let Err(err) = app.send_renderer_report_command(renderer_join_report).await { + log::error!("[QConnect] Deferred renderer join failed: {err}"); + return; + } + + // 2. Initial StateUpdated report. At join time (e.g. reconnect mid-playback) + // we may already have a current track, so resolve the real duration + current/ + // next queue_item_ids instead of hardcoding nulls. + let renderer = app.renderer_state_snapshot().await; + let queue = app.queue_state_snapshot().await; + let current_track_id = renderer.current_track.as_ref().map(|item| item.track_id); + let (current_qid, next_qid, _) = current_track_id + .map(|tid| { + qconnect_app::queue_resolution::resolve_queue_item_ids_from_queue_state(&queue, tid) + }) + .unwrap_or((None, None, None)); + let duration_secs = match current_track_id { + Some(track_id) => runtime + .core() + .get_track(track_id) + .await + .map(|track| u64::from(track.duration)) + .unwrap_or(0), + None => 0, + }; + let mut state_report_payload = json!({ + "playing_state": PLAYING_STATE_STOPPED, + "buffer_state": BUFFER_STATE_OK, + "current_position": 0, + "duration": duration_secs, + "queue_version": { + "major": queue_version_ref.major, + "minor": queue_version_ref.minor + } + }); + if let Some(qid) = current_qid { + state_report_payload["current_queue_item_id"] = json!(qid); + } + if let Some(qid) = next_qid { + state_report_payload["next_queue_item_id"] = json!(qid); + } + let state_report = RendererReport::new( + RendererReportType::RndrSrvrStateUpdated, + Uuid::new_v4().to_string(), + queue_version_ref, + state_report_payload, + ); + if let Err(err) = app.send_renderer_report_command(state_report).await { + log::error!("[QConnect] Deferred renderer state report failed: {err}"); + } + + // 3. Report volume and max audio quality. + // FIX (a) (§7.4, daemon-copy only): report the player's REAL volume, not the + // desktop's hardcoded 100. `get_playback_state().volume` is a 0.0-1.0 + // fraction; the wire wants a 0-100 percent. + let volume_pct = { + let fraction = runtime.core().get_playback_state().volume; + (fraction.clamp(0.0, 1.0) * 100.0).round() as i32 + }; + let volume_report = RendererReport::new( + RendererReportType::RndrSrvrVolumeChanged, + Uuid::new_v4().to_string(), + queue_version_ref, + json!({ "volume": volume_pct }), + ); + if let Err(err) = app.send_renderer_report_command(volume_report).await { + log::error!("[QConnect] Deferred renderer volume report failed: {err}"); + } + + let max_quality_report = RendererReport::new( + RendererReportType::RndrSrvrMaxAudioQualityChanged, + Uuid::new_v4().to_string(), + queue_version_ref, + json!({ "max_audio_quality": AUDIO_QUALITY_HIRES_LEVEL2 }), + ); + if let Err(err) = app.send_renderer_report_command(max_quality_report).await { + log::error!("[QConnect] Deferred renderer max quality report failed: {err}"); + } + + log::info!("[QConnect] Deferred renderer join complete"); + + // Re-request session state so the server sends an updated renderer list + // (including ourselves). + tokio::time::sleep(std::time::Duration::from_millis(500)).await; + let refresh_command = app + .build_queue_command(QueueCommandType::CtrlSrvrAskForQueueState, json!({})) + .await; + if let Ok(action_uuid) = app.send_queue_command(refresh_command).await { + app.clear_pending_if_matches(&action_uuid).await; + log::info!("[QConnect] Re-requested session state after renderer join"); + } + + // Resync the active renderer's full state too, so a reconnect rejoin restores + // renderer state. + if let Err(err) = app.ask_for_active_renderer_state().await { + log::warn!("[QConnect] Post-join AskForRendererState failed: {err}"); + } + + // Record this session_uuid so a subsequent SESSION_STATE with the same uuid + // takes the idempotent fast-path above. + { + let mut st = sync_state.lock().await; + st.last_joined_session_uuid = Some(session_uuid.to_string()); + } +} diff --git a/crates/qbzd/src/qconnect/sink.rs b/crates/qbzd/src/qconnect/sink.rs new file mode 100644 index 000000000..acbd3a4cd --- /dev/null +++ b/crates/qbzd/src/qconnect/sink.rs @@ -0,0 +1,341 @@ +// TODO(converge: qconnect-glue) — copied from crates/qbz/src/qconnect_event_sink.rs @ c8ef2a1b; +// do not fix bugs here without fixing the source, and vice versa. +// +//! Daemon `QconnectEventSink`. +//! +//! Receives `QconnectAppEvent`s from the qconnect-app crate and dispatches the +//! renderer-critical arms into the [`DaemonRendererEngine`] (the renderer seam), +//! the shared `QconnectRemoteSyncState` accumulator, and the frontend-agnostic +//! `qconnect_app::renderer` orchestration (materialize / apply / loop-mode / +//! cursor-align). The session-management critical section is delegated to +//! `QconnectApp::apply_session_management_event`; only the post-lock work the +//! returned `SessionApplyOutcome` asks for runs here. +//! +//! Daemon adaptation vs. the Slint copy (§1.4): the FOUR renderer-critical arms +//! (QueueUpdated / RendererCommandApplied / SessionManagementEvent / +//! RendererUpdated) are wired so the daemon functions AS a renderer; every UI arm +//! (device picker, now-playing card, DEV modal, toasts) is dropped, and the three +//! desktop `toast::error_weak` surfaces become `log::warn!`. + +use std::sync::{Arc, OnceLock, Weak}; + +use async_trait::async_trait; +use qconnect_app::{ + build_session_renderer_snapshot, cache_renderer_snapshot, is_peer_renderer_active, QconnectApp, + QconnectAppEvent, QconnectEventSink, QconnectRemoteSyncState, QconnectRendererEngine, + RendererCommand, RendererReport, RendererReportType, +}; +use qconnect_transport_ws::NativeWsTransport; +use serde_json::Value; +use tokio::sync::Mutex; +use uuid::Uuid; + +use super::engine::DaemonRendererEngine; +use super::transport::{resolve_local_identity, BUFFER_STATE_OK}; + +/// Concrete `QconnectApp` type used by the daemon adapter. +pub type DaemonQconnectApp = QconnectApp; + +pub struct DaemonEventSink { + /// Renderer seam — forwards the `qconnect_app::renderer` orchestration onto + /// `runtime.core()` + the protected player. + engine: DaemonRendererEngine, + /// THE shared remote-sync accumulator (one Mutex, shared with `QconnectApp`). + sync_state: Arc>, + /// Late-bound weak handle to the owning app, wired via `set_app` after the + /// app is built FROM this sink. Used to emit renderer reports (e.g. + /// is_active=true after SetActive(true)) and to drive the session-apply + + /// freeze/watchdog without an ownership cycle. + app: Arc>>, + /// FIX #13: previous "a peer is the active renderer" state, tracked across + /// `apply_session_management_event` calls. On a false->true transition (the + /// daemon becomes a CONTROLLER) we fire one `ask_for_active_renderer_state` to + /// fetch the peer's full state (incl. `current_queue_item_id`) so the queue + /// cursor resolves the peer's CURRENT track immediately instead of staying + /// stale until the peer changes track. Edge-detected to avoid spamming on + /// every periodic state-update frame. + last_peer_active: std::sync::atomic::AtomicBool, +} + +impl DaemonEventSink { + pub fn new( + engine: DaemonRendererEngine, + sync_state: Arc>, + ) -> Self { + Self { + engine, + sync_state, + app: Arc::new(OnceLock::new()), + last_peer_active: std::sync::atomic::AtomicBool::new(false), + } + } + + /// Wire the owning app after construction. Idempotent (OnceLock). + pub fn set_app(&self, app: &Arc) { + let _ = self.app.set(Arc::downgrade(app)); + } + + /// Emit a StateUpdated report announcing this renderer is now active. Sent + /// after SetActive(true) is applied so the controller learns we are ready. + async fn report_active_renderer_ready(&self) { + let Some(app) = self.app.get().and_then(Weak::upgrade) else { + return; + }; + let queue_version = app.queue_state_snapshot().await.version; + let report = RendererReport::new( + RendererReportType::RndrSrvrStateUpdated, + Uuid::new_v4().to_string(), + queue_version, + serde_json::json!({ + "is_active": true, + "buffer_state": BUFFER_STATE_OK, + "queue_version": { + "major": queue_version.major, + "minor": queue_version.minor + } + }), + ); + if let Err(err) = app.send_renderer_report_command(report).await { + log::warn!("[QConnect] Failed to report active-renderer-ready: {err}"); + } + } + + /// Apply a server session-management event by delegating the locked critical + /// section to qconnect-app, then running the post-lock renderer-engine work + /// the returned `SessionApplyOutcome` asks for. Mirrors the Tauri + /// `apply_session_management_event`; the post-lock ordering (loop mode -> + /// local-playback handoff -> projection -> freeze -> watchdog) is identical. + async fn apply_session_management_event(&self, message_type: &str, payload: &Value) { + let Some(app) = self.app.get().and_then(Weak::upgrade) else { + return; + }; + let identity = resolve_local_identity(); + let outcome = app + .apply_session_management_event(message_type, payload, &identity) + .await; + + if let Some(loop_mode) = outcome.apply_loop_mode { + if let Err(err) = qconnect_app::renderer::apply_remote_loop_mode(&self.engine, loop_mode) + .await + { + log::warn!("[QConnect] Failed to apply remote loop mode: {err}"); + } + } + + if outcome.sync_local_playback { + self.sync_local_playback_for_renderer_ownership().await; + } + + if let Some(renderer_id) = outcome.remote_projection_renderer_id { + self.sync_active_renderer_projection(renderer_id).await; + } + + if let Some(renderer_id) = outcome.disconnected_renderer_id { + app.freeze_active_renderer_projection( + renderer_id, + QconnectAppEvent::RendererDisconnected { renderer_id }, + ) + .await; + } + + if let Some((renderer_id, generation)) = outcome.watchdog_arm { + app.arm_renderer_watchdog(renderer_id, generation); + } + + // FIX #13: when the daemon transitions INTO controller mode (a PEER + // becomes the active renderer), the peer's periodic state-update frames + // carry `current_queue_item_id: null` (position-only), so on the + // transition the cursor/projection can't resolve the peer's CURRENT track. + // Fetch the peer's FULL state once on the false->true edge so the existing + // align + projection resolve the real current track now. + let peer_active_now = { + let state = self.sync_state.lock().await; + is_peer_renderer_active(&state.session) + }; + let was_peer_active = self + .last_peer_active + .swap(peer_active_now, std::sync::atomic::Ordering::Relaxed); + if peer_active_now && !was_peer_active { + if let Err(err) = app.ask_for_active_renderer_state().await { + log::warn!( + "[QConnect] controller entry: ask_for_active_renderer_state failed: {err}" + ); + } + } + } + + /// When an active PEER renderer now owns playback, stop our local playback so + /// the two don't double-play. Mirrors the Tauri helper, with the engine seam + /// in place of `CoreBridge`. + async fn sync_local_playback_for_renderer_ownership(&self) { + let peer_renderer_active = { + let state = self.sync_state.lock().await; + is_peer_renderer_active(&state.session) + }; + if !peer_renderer_active { + return; + } + + let playback_state = self.engine.get_playback_state(); + if playback_state.track_id == 0 { + return; + } + + log::info!( + "[QConnect] Stopping local playback because active renderer is a peer (track_id={})", + playback_state.track_id + ); + if let Err(err) = self.engine.stop() { + log::warn!("[QConnect] Failed to stop local playback after renderer handoff: {err}"); + } + } + + /// Refresh the cached projection for the active renderer and, when a peer owns + /// playback, align the local queue cursor to the peer's current track (so a + /// later takeover lands on the right track). Mirrors the Tauri helper. + async fn sync_active_renderer_projection(&self, renderer_id: i32) { + let (queue_state, renderer_state, session_loop_mode, should_align_engine) = { + let state = self.sync_state.lock().await; + let Some(active_renderer_id) = state.session.active_renderer_id else { + return; + }; + if active_renderer_id != renderer_id { + return; + } + + ( + state.last_remote_queue_state.clone(), + state + .session_renderer_states + .get(&active_renderer_id) + .cloned(), + state.session_loop_mode, + state.session.local_renderer_id != Some(active_renderer_id), + ) + }; + + let (Some(queue_state), Some(renderer_state)) = (queue_state, renderer_state) else { + return; + }; + + let renderer_snapshot = + build_session_renderer_snapshot(&queue_state, Some(&renderer_state), session_loop_mode); + { + let mut state = self.sync_state.lock().await; + cache_renderer_snapshot(&mut state, &renderer_snapshot); + } + + if !should_align_engine { + return; + } + + let Some(current_track) = renderer_snapshot.current_track.as_ref() else { + return; + }; + + if let Err(err) = + qconnect_app::renderer::align_queue_cursor(&self.engine, current_track.track_id).await + { + log::warn!("[QConnect] Failed to sync peer renderer cursor into engine: {err}"); + } + } +} + +#[async_trait] +impl QconnectEventSink for DaemonEventSink { + async fn on_event(&self, event: QconnectAppEvent) { + match &event { + QconnectAppEvent::SessionManagementEvent { + message_type, + payload, + } => { + log::info!( + "[QConnect] Session management: {} payload={}", + message_type, + serde_json::to_string(payload).unwrap_or_else(|_| "?".to_string()) + ); + self.apply_session_management_event(message_type, payload) + .await; + } + QconnectAppEvent::RendererUpdated(renderer_state) => { + log::info!( + "[QConnect] Renderer updated: playing_state={:?} volume={:?} position={:?}", + renderer_state.playing_state, + renderer_state.volume, + renderer_state.current_position_ms, + ); + let mut sync_state = self.sync_state.lock().await; + cache_renderer_snapshot(&mut sync_state, renderer_state); + } + QconnectAppEvent::QueueUpdated(queue_state) => { + log::debug!( + "[QConnect] QueueUpdated: items={} shuffle_mode={} version={}.{}", + queue_state.queue_items.len(), + queue_state.shuffle_mode, + queue_state.version.major, + queue_state.version.minor, + ); + { + let mut sync_state = self.sync_state.lock().await; + sync_state.last_remote_queue_state = Some(queue_state.clone()); + } + if let Err(err) = qconnect_app::renderer::materialize_remote_queue( + &self.engine, + &self.sync_state, + queue_state, + ) + .await + { + log::warn!("[QConnect] Failed to materialize remote queue: {err}"); + } + } + QconnectAppEvent::RendererCommandApplied { command, state } => { + log::info!("[QConnect] Renderer command applied: {:?}", command); + let became_active = + matches!(command, RendererCommand::SetActive { active: true }); + if let Err(err) = qconnect_app::renderer::apply_renderer_command( + &self.engine, + &self.sync_state, + command, + state, + ) + .await + { + log::warn!("[QConnect] Failed to apply renderer command: {err}"); + } else if became_active { + self.report_active_renderer_ready().await; + } + } + QconnectAppEvent::RendererUnreachable { renderer_id } => { + // Slint copy surfaced a toast here — daemon logs it (§1.4). + log::warn!("[QConnect] Renderer {renderer_id} unreachable"); + } + QconnectAppEvent::RendererDisconnected { renderer_id } => { + // Slint copy surfaced a toast here — daemon logs it (§1.4). + log::warn!("[QConnect] Renderer {renderer_id} disconnected"); + } + QconnectAppEvent::PlaybackError { + queue_item_id, + error_type, + .. + } => { + // Slint copy surfaced a toast here — daemon logs it (§1.4). + log::warn!( + "[QConnect] Playback error on queue_item {queue_item_id}: {error_type:?}" + ); + } + QconnectAppEvent::ResyncComplete => { + log::info!("[QConnect] Post-reconnect resync complete"); + } + QconnectAppEvent::LifecycleChanged { state } => { + log::info!("[QConnect] Lifecycle -> {state:?}"); + } + QconnectAppEvent::Diagnostic { + channel, level, .. + } => { + log::debug!("[QConnect] diagnostic {channel} [{level}]"); + } + _ => {} + } + } +} diff --git a/crates/qbzd/src/qconnect/transport.rs b/crates/qbzd/src/qconnect/transport.rs new file mode 100644 index 000000000..c0ae7be05 --- /dev/null +++ b/crates/qbzd/src/qconnect/transport.rs @@ -0,0 +1,689 @@ +// TODO(converge: qconnect-glue) — copied from crates/qbz/src/qconnect_transport.rs @ 00a44e12; +// do not fix bugs here without fixing the source, and vice versa. +// +//! QConnect transport config + credential discovery + device identity for the +//! qbzd daemon. +//! +//! Ports the desktop `qconnect_transport.rs` config layer. The Qobuz client is +//! reached via `runtime.core().client()` — an `Arc>>` +//! (may be `None` before the API is initialized). Two daemon adaptations vs. the +//! desktop copy (§1.4, §4): (1) the device-uuid + settings-DB path resolve +//! against the DAEMON root's `qconnect_settings.db` — its OWN device_uuid, minted +//! fresh, NEVER the desktop's global KV; (2) the default device name is +//! "QBZ (hostname)" (OD6). The KV load/persist helpers are path-parameterized +//! `*_at` variants so T11 (settings) and T13 (TUI) drive the same daemon-root DB. +#![allow(dead_code)] + +use std::path::{Path, PathBuf}; +use std::sync::{Arc, OnceLock}; + +use qbz_app::shell::AppRuntime; +use qconnect_transport_ws::WsTransportConfig; +use serde::{Deserialize, Serialize}; +use serde_json::Value; + +use crate::adapter::DaemonAdapter; + +const DEFAULT_QCONNECT_DEVICE_BRAND: &str = "QBZ"; +const DEFAULT_QCONNECT_DEVICE_MODEL: &str = "QBZ"; +const DEFAULT_QCONNECT_DEVICE_TYPE: i32 = 5; // computer +const DEFAULT_QCONNECT_SOFTWARE_PREFIX: &str = "qbz"; +const QCONNECT_QWS_TOKEN_KIND: &str = "jwt_qws"; +const QCONNECT_QWS_CREATE_TOKEN_PATH: &str = "/qws/createToken"; + +type Runtime = Arc>; + +/// The daemon-root `qconnect_settings.db` path, set once by `qconnect::start` +/// before any connect. Daemon adaptation: the desktop delegates the path to the +/// GLOBAL `qbz_app::qconnect_identity`; the daemon points every device-identity +/// and KV read at its OWN root (§4 — the daemon never touches the desktop KV). +static DAEMON_QCONNECT_SETTINGS_DB: OnceLock = OnceLock::new(); + +/// Point the QConnect settings/device-identity DB at the daemon root. Idempotent +/// (a second call is ignored). Called from `qconnect::start(roots)` with +/// `/qconnect_settings.db`. +pub fn init_settings_db_path(path: PathBuf) { + let _ = DAEMON_QCONNECT_SETTINGS_DB.set(path); +} + +/// The daemon-root QConnect settings DB path, if `init_settings_db_path` ran. +fn qconnect_settings_db_path() -> Option { + DAEMON_QCONNECT_SETTINGS_DB.get().cloned() +} + +/// Persistent QConnect device identity for the DAEMON. Daemon adaptation vs. the +/// desktop (which delegates to the shared global identity module): the uuid is +/// minted + persisted in the daemon root's `qconnect_settings.db`, so the daemon +/// keeps its OWN identity, distinct from the desktop install. An explicit +/// `QBZ_QCONNECT_DEVICE_UUID` env value still wins. Fail-open: a fresh v4 when no +/// daemon settings path is available yet. +pub fn resolve_qconnect_device_uuid() -> String { + if let Some(explicit) = std::env::var("QBZ_QCONNECT_DEVICE_UUID") + .ok() + .map(|value| value.trim().to_string()) + .filter(|value| !value.is_empty()) + { + return explicit; + } + match qconnect_settings_db_path() { + Some(path) => qbz_app::qconnect_identity::device_uuid_from_db(&path), + None => uuid::Uuid::new_v4().to_string(), + } +} + +/// `qbz/` software-version string for the device-info payload. +pub fn resolve_qconnect_software_version() -> String { + std::env::var("QBZ_QCONNECT_SOFTWARE_VERSION") + .ok() + .filter(|value| !value.trim().is_empty()) + .unwrap_or_else(|| { + format!( + "{DEFAULT_QCONNECT_SOFTWARE_PREFIX}/{}", + env!("CARGO_PKG_VERSION") + ) + }) +} + +pub fn resolve_qconnect_device_brand() -> String { + std::env::var("QBZ_QCONNECT_DEVICE_BRAND") + .ok() + .filter(|value| !value.trim().is_empty()) + .unwrap_or_else(|| DEFAULT_QCONNECT_DEVICE_BRAND.to_string()) +} + +pub fn resolve_qconnect_device_model() -> String { + std::env::var("QBZ_QCONNECT_DEVICE_MODEL") + .ok() + .filter(|value| !value.trim().is_empty()) + .unwrap_or_else(|| DEFAULT_QCONNECT_DEVICE_MODEL.to_string()) +} + +pub fn resolve_qconnect_device_type() -> i32 { + std::env::var("QBZ_QCONNECT_DEVICE_TYPE") + .ok() + .and_then(|value| value.trim().parse::().ok()) + .unwrap_or(DEFAULT_QCONNECT_DEVICE_TYPE) +} + +pub fn resolve_system_hostname() -> String { + if let Ok(h) = std::env::var("HOSTNAME") { + if !h.trim().is_empty() { + return h.trim().to_string(); + } + } + if let Ok(h) = std::fs::read_to_string("/etc/hostname") { + let trimmed = h.trim().to_string(); + if !trimmed.is_empty() { + return trimmed; + } + } + "Desktop".to_string() +} + +/// Returns "QBZ ({hostname})" as the default device name (OD6 — daemon +/// adaptation; the desktop default is "Qbz - {hostname}"). +pub fn resolve_default_qconnect_device_name() -> String { + format!("QBZ ({})", resolve_system_hostname()) +} + +/// Effective friendly name: custom override -> env -> "QBZ ({hostname})". +pub fn resolve_qconnect_friendly_name(custom_name: Option<&str>) -> String { + custom_name + .filter(|s| !s.trim().is_empty()) + .map(|s| s.to_string()) + .or_else(|| { + std::env::var("QBZ_QCONNECT_DEVICE_NAME") + .ok() + .filter(|value| !value.trim().is_empty()) + }) + .unwrap_or_else(resolve_default_qconnect_device_name) +} + +// --------------------------------------------------------------------------- +// Device-info + join payload shapes (piece a). Mirror the Tauri adapter +// `types.rs` `QconnectDeviceInfoPayload` / `QconnectDeviceCapabilitiesPayload` / +// `QconnectJoinSessionRequest` exactly so the wire frames the server sees are +// byte-identical between the frontends + the daemon. Built from the device- +// identity resolvers above; the uuid resolves against the daemon root so the +// daemon keeps ONE identity of its own. +// --------------------------------------------------------------------------- + +// AudioQuality wire levels: 1=mp3, 4=hires_l2(192k). Capabilities advertise the +// device's min/max decode support; VOLUME_REMOTE_CONTROL_ALLOWED(2) means a +// controller may set our volume. +pub const AUDIO_QUALITY_MP3: i32 = 1; +pub const AUDIO_QUALITY_HIRES_LEVEL2: i32 = 4; +const VOLUME_REMOTE_CONTROL_ALLOWED: i32 = 2; +/// Renderer buffer-state wire value for OK/ready (mirrors the Tauri adapter). +pub const BUFFER_STATE_OK: i32 = 2; + +#[derive(Debug, Clone, Default, Serialize, Deserialize)] +pub struct QconnectDeviceCapabilitiesPayload { + pub min_audio_quality: Option, + pub max_audio_quality: Option, + pub volume_remote_control: Option, +} + +#[derive(Debug, Clone, Default, Serialize, Deserialize)] +pub struct QconnectDeviceInfoPayload { + pub device_uuid: Option, + pub friendly_name: Option, + pub brand: Option, + pub model: Option, + pub serial_number: Option, + pub device_type: Option, + pub capabilities: Option, + pub software_version: Option, +} + +#[derive(Debug, Clone, Default, Serialize, Deserialize)] +pub struct QconnectJoinSessionRequest { + pub session_uuid: Option, + pub device_info: Option, +} + +// --------------------------------------------------------------------------- +// Controller-mode transport request DTOs (P1 CONTROLLER port). Mirror the Tauri +// adapter `types.rs` field-for-field so the wire frames the cloud parses are +// byte-identical. Do NOT rename/reorder optionals. Unused in the P0 renderer +// path (dead_code allowed) — kept so the controller port converges cleanly. +// --------------------------------------------------------------------------- + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct QconnectQueueVersionPayload { + pub major: u64, + pub minor: u64, +} + +#[derive(Debug, Clone, Default, Serialize, Deserialize)] +pub struct QconnectSetPlayerStateQueueItemPayload { + #[serde(skip_serializing_if = "Option::is_none")] + pub queue_version: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub id: Option, +} + +// `skip_serializing_if` is LOAD-BEARING here: a `None` field must be OMITTED, not +// serialized as JSON `null`. The protocol mapper rejects a `current_queue_item` +// that is present-but-null ("must be an object") and aborts the whole send, so a +// bare play/pause toggle (which sends only `playing_state`) would never reach the +// wire without these. Each SetPlayerState field is independently optional per the +// protocol, so omitting them is the intended "do one or several" behavior. +#[derive(Debug, Clone, Default, Serialize, Deserialize)] +pub struct QconnectSetPlayerStateRequest { + #[serde(skip_serializing_if = "Option::is_none")] + pub playing_state: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub current_position: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub current_queue_item: Option, +} + +#[derive(Debug, Clone, Default, Serialize, Deserialize)] +pub struct QconnectSetVolumeRequest { + pub renderer_id: Option, + pub volume: Option, + pub volume_delta: Option, +} + +#[derive(Debug, Clone, Default, Serialize, Deserialize)] +pub struct QconnectMuteVolumeRequest { + pub renderer_id: Option, + pub value: bool, +} + +#[derive(Debug, Clone, Default, Serialize, Deserialize)] +pub struct QconnectSetActiveRendererRequest { + pub renderer_id: Option, +} + +/// Build a `CtrlSrvrSetPlayerState` request that seeks the remote renderer to +/// `position_ms` for `current_queue_item_id` under optimistic concurrency. +/// +/// `playing_state` is intentionally `None`: a seek must not toggle play/pause. +/// Pure mirror of the Tauri `build_set_position_player_state_request`. +pub fn build_set_position_player_state_request( + position_ms: i64, + current_queue_item_id: Option, + queue_version: QconnectQueueVersionPayload, +) -> QconnectSetPlayerStateRequest { + let current_position = i32::try_from(position_ms.max(0)).ok(); + let current_queue_item = current_queue_item_id.and_then(|qid| { + i32::try_from(qid) + .ok() + .map(|id| QconnectSetPlayerStateQueueItemPayload { + queue_version: Some(QconnectQueueVersionPayload { + major: queue_version.major, + minor: queue_version.minor, + }), + id: Some(id), + }) + }); + QconnectSetPlayerStateRequest { + playing_state: None, + current_position, + current_queue_item, + } +} + +/// Build the device-info payload with the default friendly name. +pub fn default_qconnect_device_info() -> QconnectDeviceInfoPayload { + default_qconnect_device_info_with_name(None) +} + +/// Build the device-info payload, overriding the friendly name when a custom +/// device name is set. Mirrors the Tauri `default_qconnect_device_info_with_name`. +pub fn default_qconnect_device_info_with_name( + custom_name: Option<&str>, +) -> QconnectDeviceInfoPayload { + QconnectDeviceInfoPayload { + device_uuid: Some(resolve_qconnect_device_uuid()), + friendly_name: Some(resolve_qconnect_friendly_name(custom_name)), + brand: Some(resolve_qconnect_device_brand()), + model: Some(resolve_qconnect_device_model()), + serial_number: None, + device_type: Some(resolve_qconnect_device_type()), + capabilities: Some(QconnectDeviceCapabilitiesPayload { + min_audio_quality: Some(AUDIO_QUALITY_MP3), + max_audio_quality: Some(AUDIO_QUALITY_HIRES_LEVEL2), + volume_remote_control: Some(VOLUME_REMOTE_CONTROL_ALLOWED), + }), + software_version: Some(resolve_qconnect_software_version()), + } +} + +/// Resolve THIS device's identity for injection into the frontend-agnostic +/// session-apply logic in qconnect-app (`apply_session_management_event` takes a +/// `LocalIdentity` so the crate never depends on the persistence layer). Mirrors +/// the Tauri `resolve_local_identity`. +pub fn resolve_local_identity() -> qconnect_app::LocalIdentity { + let info = default_qconnect_device_info(); + qconnect_app::LocalIdentity { + device_uuid: info.device_uuid.unwrap_or_default(), + friendly_name: info.friendly_name, + brand: info.brand, + model: info.model, + device_type: info.device_type, + } +} + +// --------------------------------------------------------------------------- +// Daemon-root KV helpers (path-parameterized `*_at` copies of +// qconnect_transport.rs:281-411). The desktop variants read the GLOBAL settings +// DB; the daemon threads the DB path so T11 (`qbzd qconnect enable|disable`, +// settings show/set) and T13 (TUI QConnect screen) drive the SAME daemon-root DB +// the connect path reads. Every helper is fail-open (returns the default / no-op +// on any error). +// --------------------------------------------------------------------------- + +/// Open the daemon-root QConnect settings DB (key/value table), creating it when +/// missing. Mirrors the Tauri `startup.rs::open_settings_conn`. +fn open_qconnect_settings_conn_at(path: &Path) -> Option { + let conn = rusqlite::Connection::open(path).ok()?; + conn.execute_batch("PRAGMA journal_mode=WAL; PRAGMA synchronous=NORMAL;") + .ok()?; + conn.execute_batch( + "CREATE TABLE IF NOT EXISTS settings ( + key TEXT PRIMARY KEY, + value TEXT NOT NULL + )", + ) + .ok()?; + Some(conn) +} + +/// Load the persisted custom device name (fail-open: None on any error). +pub fn load_device_name_at(path: &Path) -> Option { + let conn = open_qconnect_settings_conn_at(path)?; + conn.query_row( + "SELECT value FROM settings WHERE key = 'device_name'", + [], + |row| row.get::<_, String>(0), + ) + .ok() + .filter(|v| !v.trim().is_empty()) +} + +/// Persist the custom device name (None clears it). +pub fn persist_device_name_at(path: &Path, name: Option<&str>) { + let Some(conn) = open_qconnect_settings_conn_at(path) else { + return; + }; + match name { + Some(n) => { + let _ = conn.execute( + "INSERT OR REPLACE INTO settings (key, value) VALUES ('device_name', ?1)", + rusqlite::params![n], + ); + } + None => { + let _ = conn.execute("DELETE FROM settings WHERE key = 'device_name'", []); + } + } +} + +/// Load the persisted QConnect startup mode (`startup_mode` = "off" | "on" | +/// "remember_last"). Fail-open: returns `Off` (the default) when missing/invalid. +pub fn load_startup_mode_at(path: &Path) -> qconnect_app::QconnectStartupMode { + let Some(conn) = open_qconnect_settings_conn_at(path) else { + return qconnect_app::QconnectStartupMode::default(); + }; + conn.query_row( + "SELECT value FROM settings WHERE key = 'startup_mode'", + [], + |row| row.get::<_, String>(0), + ) + .ok() + .as_deref() + .and_then(qconnect_app::QconnectStartupMode::from_str) + .unwrap_or_default() +} + +/// Persist the QConnect startup mode. +pub fn save_startup_mode_at(path: &Path, mode: qconnect_app::QconnectStartupMode) { + let Some(conn) = open_qconnect_settings_conn_at(path) else { + return; + }; + let _ = conn.execute( + "INSERT OR REPLACE INTO settings (key, value) VALUES ('startup_mode', ?1)", + rusqlite::params![mode.as_str()], + ); +} + +/// Load the last-known QConnect on/off state, if recorded (`last_known_state` = +/// "on" | "off"). +pub fn load_last_known_state_at(path: &Path) -> Option { + let conn = open_qconnect_settings_conn_at(path)?; + let value: Option = conn + .query_row( + "SELECT value FROM settings WHERE key = 'last_known_state'", + [], + |row| row.get::<_, String>(0), + ) + .ok(); + match value.as_deref() { + Some("on") => Some(true), + Some("off") => Some(false), + _ => None, + } +} + +/// Persist the last-known on/off state. +pub fn save_last_known_state_at(path: &Path, state: bool) { + let Some(conn) = open_qconnect_settings_conn_at(path) else { + return; + }; + let value = if state { "on" } else { "off" }; + let _ = conn.execute( + "INSERT OR REPLACE INTO settings (key, value) VALUES ('last_known_state', ?1)", + rusqlite::params![value], + ); +} + +/// Load the daemon volume mode (`volume_mode` = "software" | "locked"; 01 §7.4). +/// NEW daemon key (the desktop never reads it). Fail-open: None when unset. +pub fn load_volume_mode_at(path: &Path) -> Option { + let conn = open_qconnect_settings_conn_at(path)?; + conn.query_row( + "SELECT value FROM settings WHERE key = 'volume_mode'", + [], + |row| row.get::<_, String>(0), + ) + .ok() + .filter(|v| !v.trim().is_empty()) +} + +/// Persist the daemon volume mode ("software" | "locked"). +pub fn save_volume_mode_at(path: &Path, mode: &str) { + let Some(conn) = open_qconnect_settings_conn_at(path) else { + return; + }; + let _ = conn.execute( + "INSERT OR REPLACE INTO settings (key, value) VALUES ('volume_mode', ?1)", + rusqlite::params![mode], + ); +} + +/// Build the WS transport config: env overrides first, then auto-discovery via +/// `qws/createToken`. Mirrors the Tauri `resolve_transport_config` (the per-option +/// knobs are deferred — the connect path uses defaults + env). `require_jwt = +/// true` (hard) and `reconnect_idle_retry_ms = 60s` match the hardened defaults. +pub async fn resolve_transport_config(runtime: &Runtime) -> Result { + let mut endpoint_url = normalize_opt_string(std::env::var("QBZ_QCONNECT_WS_ENDPOINT").ok()); + let mut jwt_qws = normalize_opt_string(std::env::var("QBZ_QCONNECT_JWT_QWS").ok()) + .or_else(|| normalize_opt_string(std::env::var("QBZ_QCONNECT_JWT").ok())); + + if endpoint_url.is_none() || jwt_qws.is_none() { + match fetch_qconnect_transport_credentials(runtime).await { + Ok((discovered_endpoint, discovered_jwt_qws)) => { + endpoint_url = endpoint_url.or(discovered_endpoint); + jwt_qws = jwt_qws.or(discovered_jwt_qws); + } + Err(err) if endpoint_url.is_some() => { + log::warn!( + "[QConnect] qws/createToken auto-discovery failed, using provided endpoint: {err}" + ); + } + Err(err) => { + return Err(format!( + "QConnect endpoint_url is required (or QBZ_QCONNECT_WS_ENDPOINT). Auto-discovery via qws/createToken failed: {err}" + )); + } + } + } + + let endpoint_url = endpoint_url.ok_or_else(|| { + "QConnect endpoint_url is required (or QBZ_QCONNECT_WS_ENDPOINT)".to_string() + })?; + + let subscribe_channels = if let Ok(raw) = std::env::var("QBZ_QCONNECT_SUBSCRIBE_CHANNELS_HEX") { + let channels: Vec = raw + .split(',') + .map(str::trim) + .filter(|item| !item.is_empty()) + .map(ToString::to_string) + .collect(); + parse_subscribe_channels(channels)? + } else { + // Default QConnect channels: connectionId(0x01), backend(0x02), controllers(0x03) + vec![vec![0x01], vec![0x02], vec![0x03]] + }; + + let mut config = WsTransportConfig::default(); + config.endpoint_url = endpoint_url; + config.jwt_qws = jwt_qws; + config.require_jwt = true; + config.reconnect_idle_retry_ms = 60_000; + config.subscribe_channels = subscribe_channels; + Ok(config) +} + +async fn fetch_qconnect_transport_credentials( + runtime: &Runtime, +) -> Result<(Option, Option), String> { + // The client is Option-wrapped (None before API init). + let client = runtime + .core() + .client() + .read() + .await + .clone() + .ok_or_else(|| "qws/createToken requires an initialized API client".to_string())?; + + let app_id = client + .app_id() + .await + .map_err(|err| format!("qws/createToken requires initialized API client: {err}"))?; + let user_auth_token = client + .auth_token() + .await + .map_err(|err| format!("qws/createToken requires authenticated user: {err}"))?; + + let mut headers = reqwest::header::HeaderMap::new(); + headers.insert( + "X-App-Id", + reqwest::header::HeaderValue::from_str(&app_id) + .map_err(|_| "invalid X-App-Id header value".to_string())?, + ); + headers.insert( + "X-User-Auth-Token", + reqwest::header::HeaderValue::from_str(&user_auth_token) + .map_err(|_| "invalid X-User-Auth-Token header value".to_string())?, + ); + + let url = qbz_qobuz::endpoints::build_url(QCONNECT_QWS_CREATE_TOKEN_PATH); + let response = client + .get_http() + .post(&url) + .headers(headers) + .form(&[ + ("jwt", QCONNECT_QWS_TOKEN_KIND), + ("user_auth_token_needed", "true"), + ("strong_auth_needed", "true"), + ]) + .send() + .await + .map_err(|err| format!("qws/createToken HTTP request failed: {err}"))?; + + let status = response.status(); + let payload: Value = response + .json() + .await + .map_err(|err| format!("qws/createToken response decode failed: {err}"))?; + + if !status.is_success() { + let preview = serde_json::to_string(&payload) + .unwrap_or_else(|_| "".to_string()) + .chars() + .take(300) + .collect::(); + return Err(format!("qws/createToken status {status}: {preview}")); + } + + let jwt_qws_payload = payload + .get("jwt_qws") + .ok_or_else(|| "qws/createToken response missing jwt_qws payload".to_string())?; + + let endpoint_url = normalize_opt_string( + jwt_qws_payload + .get("endpoint") + .and_then(Value::as_str) + .map(ToString::to_string), + ); + let jwt_qws = normalize_opt_string( + jwt_qws_payload + .get("jwt") + .and_then(Value::as_str) + .map(ToString::to_string), + ); + + if endpoint_url.is_none() { + return Err("qws/createToken response missing jwt_qws.endpoint".to_string()); + } + + Ok((endpoint_url, jwt_qws)) +} + +fn normalize_opt_string(value: Option) -> Option { + value.and_then(|value| { + let trimmed = value.trim(); + if trimmed.is_empty() { + None + } else { + Some(trimmed.to_string()) + } + }) +} + +pub fn parse_subscribe_channels(items: Vec) -> Result>, String> { + items.into_iter().map(|item| decode_hex_channel(&item)).collect() +} + +pub fn decode_hex_channel(raw: &str) -> Result, String> { + let normalized = raw.trim().trim_start_matches("0x").trim_start_matches("0X"); + if normalized.is_empty() { + return Err("empty subscribe channel hex value".to_string()); + } + + let needs_padding = normalized.len() % 2 != 0; + let value = if needs_padding { + format!("0{normalized}") + } else { + normalized.to_string() + }; + + let mut bytes = Vec::with_capacity(value.len() / 2); + let chars: Vec = value.chars().collect(); + + for idx in (0..chars.len()).step_by(2) { + let pair = [chars[idx], chars[idx + 1]]; + let hex = pair.iter().collect::(); + let byte = u8::from_str_radix(&hex, 16) + .map_err(|_| format!("invalid subscribe channel hex byte '{hex}' in '{raw}'"))?; + bytes.push(byte); + } + + Ok(bytes) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn decode_hex_channel_pads_odd_length() { + assert_eq!(decode_hex_channel("1").unwrap(), vec![0x01]); + assert_eq!(decode_hex_channel("0x03").unwrap(), vec![0x03]); + assert_eq!(decode_hex_channel("0102").unwrap(), vec![0x01, 0x02]); + assert!(decode_hex_channel("").is_err()); + assert!(decode_hex_channel("zz").is_err()); + } + + #[test] + fn default_device_name_is_qbz_hostname_form() { + // OD6: the daemon default is "QBZ (hostname)", never the desktop + // "Qbz - hostname" form. + std::env::set_var("HOSTNAME", "studio-pi"); + let name = resolve_default_qconnect_device_name(); + std::env::remove_var("HOSTNAME"); + assert_eq!(name, "QBZ (studio-pi)"); + } + + #[test] + fn kv_helpers_round_trip_against_a_temp_daemon_db() { + let tmp = std::env::temp_dir().join(format!( + "qbzd_qconnect_kv_test_{}.db", + std::process::id() + )); + let _ = std::fs::remove_file(&tmp); + + // Unset -> defaults / None. + assert_eq!( + load_startup_mode_at(&tmp), + qconnect_app::QconnectStartupMode::Off + ); + assert_eq!(load_device_name_at(&tmp), None); + assert_eq!(load_volume_mode_at(&tmp), None); + assert_eq!(load_last_known_state_at(&tmp), None); + + // Persist + read back. + save_startup_mode_at(&tmp, qconnect_app::QconnectStartupMode::On); + persist_device_name_at(&tmp, Some("Living Room")); + save_volume_mode_at(&tmp, "locked"); + save_last_known_state_at(&tmp, true); + + assert_eq!( + load_startup_mode_at(&tmp), + qconnect_app::QconnectStartupMode::On + ); + assert_eq!(load_device_name_at(&tmp).as_deref(), Some("Living Room")); + assert_eq!(load_volume_mode_at(&tmp).as_deref(), Some("locked")); + assert_eq!(load_last_known_state_at(&tmp), Some(true)); + + // Clearing the device name removes it. + persist_device_name_at(&tmp, None); + assert_eq!(load_device_name_at(&tmp), None); + + let _ = std::fs::remove_file(&tmp); + } +}