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
This commit is contained in:
vicrodh committed 2026-07-14 17:47:33 -06:00
1 parent 05e2216acc
commit 196f167873
11 files changed
+2502 -4

No files matched your search

+3
View File
@@ -6002,6 +6002,7 @@ dependencies = [
"clap", "clap",
"clap_complete", "clap_complete",
"dirs 6.0.0", "dirs 6.0.0",
"futures-util",
"libc", "libc",
"log", "log",
"open", "open",
@@ -6019,6 +6020,7 @@ dependencies = [
"qconnect-transport-ws", "qconnect-transport-ws",
"rand 0.10.1", "rand 0.10.1",
"reqwest", "reqwest",
"rusqlite",
"serde", "serde",
"serde_json", "serde_json",
"tempfile", "tempfile",
@@ -6026,6 +6028,7 @@ dependencies = [
"tokio", "tokio",
"toml", "toml",
"urlencoding", "urlencoding",
"uuid 1.22.0",
] ]
[[package]] [[package]]
+4
View File
@@ -32,6 +32,10 @@ dirs = "6"
rand = { workspace = true } rand = { workspace = true }
log = { workspace = true } log = { workspace = true }
async-trait = { 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) # new direct deps (versions already in lock where noted)
tiny_http = "0.12" # in lock via qbz-cast tiny_http = "0.12" # in lock via qbz-cast
toml = "0.9" # in lock transitively toml = "0.9" # in lock transitively
+17 -4
View File
@@ -113,15 +113,28 @@ pub async fn run(roots: ProfileRoots, cfg: QbzdConfig, warns: Vec<String>) -> Re
); );
log::info!("control API listening on {bind_addr}"); 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<AppRuntime>`, 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 // 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 // fns are verified no-ops from a fresh process and re-adding them is the
// documented skeptic-correction #1 trap (§8.1). // documented skeptic-correction #1 trap (§8.1).
wait_for_signal().await; wait_for_signal().await;
// ── Shutdown (§8.2, ordered). Stop the playback driver FIRST: it holds an // ── Shutdown (§8.2, ordered). Step 1: disconnect the QConnect session (and
// Arc<AppRuntime> clone, so its task must finish (dropping that Arc) // stop its auto-connect watcher) BEFORE playback is stopped, then drop the
// before `drop(booted)` can release the audio device ahead of the #521 // handle so its Arc<AppRuntime> clone is released ahead of `drop(booted)`.
// pair. Signal, then join. qconnect.shutdown().await;
drop(qconnect);
// Step 2: stop the playback driver. It holds an Arc<AppRuntime> 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); let _ = shutdown_tx.send(true);
if let Err(e) = driver.await { if let Err(e) = driver.await {
log::warn!("driver task join failed: {e:?}"); log::warn!("driver task join failed: {e:?}");
+1
View File
@@ -8,6 +8,7 @@ mod daemon;
mod lock; mod lock;
mod login; mod login;
mod paths; mod paths;
mod qconnect;
mod state; mod state;
pub const API_VERSION: u32 = 1; // 02-cli-and-api.md §1.6 pub const API_VERSION: u32 = 1; // 02-cli-and-api.md §1.6
+198
View File
@@ -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<AppRuntime<DaemonAdapter>>,
}
impl DaemonRendererEngine {
pub fn new(runtime: Arc<AppRuntime<DaemonAdapter>>) -> Self {
Self { runtime }
}
fn core(&self) -> &Arc<QbzCore<DaemonAdapter>> {
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<QueueTrack>, Option<usize>) {
self.core().get_all_queue_tracks().await
}
async fn set_queue(&self, tracks: Vec<QueueTrack>, start_index: Option<usize>) {
self.core().set_queue(tracks, start_index).await
}
async fn set_queue_with_order(
&self,
tracks: Vec<QueueTrack>,
start_index: Option<usize>,
shuffle_enabled: bool,
shuffle_order: Option<Vec<usize>>,
) {
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<QueueTrack> {
self.core().play_index(index).await
}
// ---- catalog (async) ----
async fn get_track(&self, track_id: u64) -> Result<Track, String> {
self.core()
.get_track(track_id)
.await
.map_err(|err| err.to_string())
}
async fn get_tracks_batch(&self, track_ids: &[u64]) -> Result<Vec<Track>, 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<Vec<u8>, 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())
}
+410
View File
@@ -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<AppRuntime<DaemonAdapter>>;
type SharedState = Arc<std::sync::Mutex<DaemonShared>>;
/// 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<DaemonQconnectApp>,
/// 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<Mutex<QconnectRemoteSyncState>>,
}
/// 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<DaemonQconnectRuntime>,
/// Latched connect/loop error; surfaced by `qbzd status` QConnect block (T11).
#[allow(dead_code)]
pub last_error: Option<String>,
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<Mutex<DaemonQconnectInner>>,
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<Mutex<DaemonQconnectInner>>,
runtime: Runtime,
shared: SharedState,
#[allow(dead_code)] // T11 (settings reload) re-reads the KV through this path.
settings_db: PathBuf,
custom_device_name: Arc<tokio::sync::RwLock<Option<String>>>,
}
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<dyn SessionLoopHost> = 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<AppRuntime>` 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<Self>) {
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<DaemonQconnectService>,
watcher: Option<JoinHandle<()>>,
}
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<AppRuntime>` 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 }
}
+289
View File
@@ -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<RemoteStreamInfo, String> {
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::<u64>().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(())
}
+208
View File
@@ -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<DaemonQconnectApp>,
sync_state: &Arc<Mutex<QconnectRemoteSyncState>>,
runtime: &Arc<AppRuntime<DaemonAdapter>>,
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<QconnectFileAudioQualitySnapshot> {
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<DaemonQconnectApp>,
sync_state: &Arc<Mutex<QconnectRemoteSyncState>>,
track_id: u64,
) -> (Option<u64>, Option<u64>) {
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);
}
}
+342
View File
@@ -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<AppRuntime<DaemonAdapter>>;
/// 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<DaemonQconnectApp>,
pub sync_state: Arc<Mutex<QconnectRemoteSyncState>>,
pub inner: Arc<Mutex<DaemonQconnectInner>>,
pub sink: Arc<DaemonEventSink>,
pub runtime: Runtime,
pub shared: Arc<std::sync::Mutex<DaemonShared>>,
}
#[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<DaemonQconnectApp>,
custom_device_name: Option<String>,
) -> 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<DaemonQconnectApp>,
sync_state: &Arc<Mutex<QconnectRemoteSyncState>>,
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());
}
}
+341
View File
@@ -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<NativeWsTransport, DaemonEventSink>;
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<Mutex<QconnectRemoteSyncState>>,
/// 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<OnceLock<Weak<DaemonQconnectApp>>>,
/// 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<Mutex<QconnectRemoteSyncState>>,
) -> 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<DaemonQconnectApp>) {
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}]");
}
_ => {}
}
}
}
+689
View File
@@ -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<RwLock<Option<QobuzClient>>>`
//! (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<AppRuntime<DaemonAdapter>>;
/// 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<PathBuf> = 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
/// `<roots.data>/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<PathBuf> {
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/<version>` 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::<i32>().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<i32>,
pub max_audio_quality: Option<i32>,
pub volume_remote_control: Option<i32>,
}
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub struct QconnectDeviceInfoPayload {
pub device_uuid: Option<String>,
pub friendly_name: Option<String>,
pub brand: Option<String>,
pub model: Option<String>,
pub serial_number: Option<String>,
pub device_type: Option<i32>,
pub capabilities: Option<QconnectDeviceCapabilitiesPayload>,
pub software_version: Option<String>,
}
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub struct QconnectJoinSessionRequest {
pub session_uuid: Option<String>,
pub device_info: Option<QconnectDeviceInfoPayload>,
}
// ---------------------------------------------------------------------------
// 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<QconnectQueueVersionPayload>,
#[serde(skip_serializing_if = "Option::is_none")]
pub id: Option<i32>,
}
// `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<i32>,
#[serde(skip_serializing_if = "Option::is_none")]
pub current_position: Option<i32>,
#[serde(skip_serializing_if = "Option::is_none")]
pub current_queue_item: Option<QconnectSetPlayerStateQueueItemPayload>,
}
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub struct QconnectSetVolumeRequest {
pub renderer_id: Option<i32>,
pub volume: Option<i32>,
pub volume_delta: Option<i32>,
}
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub struct QconnectMuteVolumeRequest {
pub renderer_id: Option<i32>,
pub value: bool,
}
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub struct QconnectSetActiveRendererRequest {
pub renderer_id: Option<i32>,
}
/// 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<u64>,
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<rusqlite::Connection> {
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<String> {
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<bool> {
let conn = open_qconnect_settings_conn_at(path)?;
let value: Option<String> = 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<String> {
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<WsTransportConfig, String> {
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<String> = 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<String>, Option<String>), 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(|_| "<unserializable>".to_string())
.chars()
.take(300)
.collect::<String>();
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<String>) -> Option<String> {
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<String>) -> Result<Vec<Vec<u8>>, String> {
items.into_iter().map(|item| decode_hex_channel(&item)).collect()
}
pub fn decode_hex_channel(raw: &str) -> Result<Vec<u8>, 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<char> = value.chars().collect();
for idx in (0..chars.len()).step_by(2) {
let pair = [chars[idx], chars[idx + 1]];
let hex = pair.iter().collect::<String>();
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);
}
}