feat(qconnect-core): add queue domain model, reducer and pending lifecycle

-cx

crates/qconnect-core/src/admission.rs

@@ -0,0 +1,63 @@
+use serde::{Deserialize, Serialize};
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+pub enum TrackOrigin {
+    QobuzOnline,
+    QobuzOfflineCache,
+    LocalLibrary,
+    Plex,
+    ExternalUnknown,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct AdmissionDecision {
+    pub accepted: bool,
+    pub reason: &'static str,
+}
+
+impl AdmissionDecision {
+    pub const fn allow(reason: &'static str) -> Self {
+        Self {
+            accepted: true,
+            reason,
+        }
+    }
+
+    pub const fn block(reason: &'static str) -> Self {
+        Self {
+            accepted: false,
+            reason,
+        }
+    }
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+pub enum HandoffIntent {
+    ContinueLocally,
+    SendToConnect,
+}
+
+pub fn evaluate_remote_queue_admission(origin: TrackOrigin) -> AdmissionDecision {
+    match origin {
+        TrackOrigin::QobuzOnline => AdmissionDecision::allow("qobuz_online_source"),
+        TrackOrigin::QobuzOfflineCache => AdmissionDecision::allow("qobuz_offline_cache_source"),
+        TrackOrigin::LocalLibrary => {
+            AdmissionDecision::block("local_library_tracks_never_enter_remote_qconnect_queue")
+        }
+        TrackOrigin::Plex => {
+            AdmissionDecision::block("plex_tracks_never_enter_remote_qconnect_queue")
+        }
+        TrackOrigin::ExternalUnknown => {
+            AdmissionDecision::block("unknown_origin_blocked_for_remote_qconnect_queue")
+        }
+    }
+}
+
+pub fn resolve_handoff_intent(origin: TrackOrigin) -> HandoffIntent {
+    match origin {
+        TrackOrigin::QobuzOnline | TrackOrigin::QobuzOfflineCache => HandoffIntent::SendToConnect,
+        TrackOrigin::LocalLibrary | TrackOrigin::Plex | TrackOrigin::ExternalUnknown => {
+            HandoffIntent::ContinueLocally
+        }
+    }
+}

crates/qconnect-core/src/lib.rs

@@ -0,0 +1,17 @@
+//! qconnect-core
+//!
+//! UI-agnostic queue domain primitives for Qobuz Connect.
+
+pub mod admission;
+pub mod pending;
+pub mod queue;
+pub mod reducer;
+pub mod telemetry;
+
+pub use admission::{
+    evaluate_remote_queue_admission, resolve_handoff_intent, AdmissionDecision, HandoffIntent,
+    TrackOrigin,
+};
+pub use pending::{PendingActionError, PendingActionSlot, PendingCorrelation, PendingQueueAction};
+pub use queue::{QConnectQueueState, QueueEvent, QueueItem, QueueVersion};
+pub use reducer::{apply_event, ReducerOutcome};

crates/qconnect-core/src/pending.rs

@@ -0,0 +1,77 @@
+use crate::queue::QueueVersion;
+use serde::{Deserialize, Serialize};
+use thiserror::Error;
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct PendingQueueAction {
+    pub uuid: String,
+    pub queue_version_ref: QueueVersion,
+    pub emit_result_event: bool,
+    pub is_ask_for_state_action: bool,
+    pub concurrency_error: bool,
+    pub sent_at_ms: u64,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum PendingCorrelation {
+    NoPending,
+    EventWithoutActionUuid,
+    Matched,
+    Concurrent,
+}
+
+#[derive(Debug, Error)]
+pub enum PendingActionError {
+    #[error("pending queue action already active: {0}")]
+    AlreadyPending(String),
+}
+
+#[derive(Debug, Default)]
+pub struct PendingActionSlot {
+    current: Option<PendingQueueAction>,
+}
+
+impl PendingActionSlot {
+    pub fn start(&mut self, action: PendingQueueAction) -> Result<(), PendingActionError> {
+        if let Some(existing) = &self.current {
+            return Err(PendingActionError::AlreadyPending(existing.uuid.clone()));
+        }
+        self.current = Some(action);
+        Ok(())
+    }
+
+    pub fn current(&self) -> Option<&PendingQueueAction> {
+        self.current.as_ref()
+    }
+
+    pub fn current_mut(&mut self) -> Option<&mut PendingQueueAction> {
+        self.current.as_mut()
+    }
+
+    pub fn correlate(&self, event_action_uuid: Option<&str>) -> PendingCorrelation {
+        let Some(pending) = &self.current else {
+            return PendingCorrelation::NoPending;
+        };
+        let Some(event_uuid) = event_action_uuid else {
+            return PendingCorrelation::EventWithoutActionUuid;
+        };
+
+        if pending.uuid == event_uuid {
+            PendingCorrelation::Matched
+        } else {
+            PendingCorrelation::Concurrent
+        }
+    }
+
+    pub fn mark_concurrency_error(&mut self) -> bool {
+        if let Some(current) = &mut self.current {
+            current.concurrency_error = true;
+            return true;
+        }
+        false
+    }
+
+    pub fn clear(&mut self) -> Option<PendingQueueAction> {
+        self.current.take()
+    }
+}

crates/qconnect-core/src/queue.rs

@@ -0,0 +1,141 @@
+use serde::{Deserialize, Serialize};
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize)]
+pub struct QueueVersion {
+    pub major: u64,
+    pub minor: u64,
+}
+
+impl QueueVersion {
+    pub const fn new(major: u64, minor: u64) -> Self {
+        Self { major, minor }
+    }
+
+    pub const fn next_minor(self) -> Self {
+        Self {
+            major: self.major,
+            minor: self.minor.saturating_add(1),
+        }
+    }
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct QueueItem {
+    pub track_context_uuid: String,
+    pub track_id: u64,
+    pub queue_item_id: u64,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct QConnectQueueState {
+    pub version: QueueVersion,
+    pub queue_items: Vec<QueueItem>,
+    pub shuffle_mode: bool,
+    pub shuffle_order: Option<Vec<usize>>,
+    pub autoplay_mode: bool,
+    pub autoplay_loading: bool,
+    pub autoplay_items: Vec<QueueItem>,
+    pub updated_at_ms: u64,
+}
+
+impl Default for QConnectQueueState {
+    fn default() -> Self {
+        Self {
+            version: QueueVersion::default(),
+            queue_items: Vec::new(),
+            shuffle_mode: false,
+            shuffle_order: None,
+            autoplay_mode: false,
+            autoplay_loading: false,
+            autoplay_items: Vec::new(),
+            updated_at_ms: 0,
+        }
+    }
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub enum QueueEvent {
+    QueueStateReplaced {
+        action_uuid: Option<String>,
+        state: QConnectQueueState,
+    },
+    TracksAdded {
+        action_uuid: Option<String>,
+        version: QueueVersion,
+        tracks: Vec<QueueItem>,
+        shuffle_seed: Option<u64>,
+        autoplay_reset: bool,
+        autoplay_loading: bool,
+    },
+    TracksLoaded {
+        action_uuid: Option<String>,
+        version: QueueVersion,
+        tracks: Vec<QueueItem>,
+        queue_position: Option<u64>,
+        shuffle_mode: Option<bool>,
+        shuffle_seed: Option<u64>,
+        shuffle_pivot_queue_item_id: Option<u64>,
+        autoplay_reset: bool,
+        autoplay_loading: bool,
+    },
+    TracksInserted {
+        action_uuid: Option<String>,
+        version: QueueVersion,
+        tracks: Vec<QueueItem>,
+        insert_after: Option<u64>,
+        shuffle_seed: Option<u64>,
+        autoplay_reset: bool,
+        autoplay_loading: bool,
+    },
+    TracksRemoved {
+        action_uuid: Option<String>,
+        version: QueueVersion,
+        queue_item_ids: Vec<u64>,
+        autoplay_reset: bool,
+        autoplay_loading: bool,
+    },
+    TracksReordered {
+        action_uuid: Option<String>,
+        version: QueueVersion,
+        queue_item_ids: Vec<u64>,
+        insert_after: Option<u64>,
+        autoplay_reset: bool,
+        autoplay_loading: bool,
+    },
+    QueueCleared {
+        action_uuid: Option<String>,
+        version: QueueVersion,
+    },
+    ShuffleModeSet {
+        action_uuid: Option<String>,
+        version: QueueVersion,
+        shuffle_mode: bool,
+        shuffle_seed: Option<u64>,
+        shuffle_pivot_queue_item_id: Option<u64>,
+        autoplay_reset: bool,
+        autoplay_loading: bool,
+    },
+    AutoplayModeSet {
+        action_uuid: Option<String>,
+        version: QueueVersion,
+        autoplay_mode: bool,
+        autoplay_reset: bool,
+        autoplay_loading: bool,
+    },
+    AutoplayTracksLoaded {
+        action_uuid: Option<String>,
+        version: QueueVersion,
+        tracks: Vec<QueueItem>,
+    },
+    AutoplayTracksRemoved {
+        action_uuid: Option<String>,
+        version: QueueVersion,
+        queue_item_ids: Vec<u64>,
+    },
+    QueueError {
+        action_uuid: Option<String>,
+        version: Option<QueueVersion>,
+        code: String,
+        message: String,
+    },
+}

crates/qconnect-core/src/reducer.rs

@@ -0,0 +1,520 @@
+use std::collections::{HashMap, HashSet};
+
+use crate::queue::{QConnectQueueState, QueueEvent, QueueItem};
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct ReducerOutcome {
+    pub queue_changed: bool,
+    pub version_changed: bool,
+    pub event_name: &'static str,
+}
+
+impl ReducerOutcome {
+    const fn unchanged(event_name: &'static str) -> Self {
+        Self {
+            queue_changed: false,
+            version_changed: false,
+            event_name,
+        }
+    }
+
+    const fn changed(event_name: &'static str, version_changed: bool) -> Self {
+        Self {
+            queue_changed: true,
+            version_changed,
+            event_name,
+        }
+    }
+}
+
+pub fn apply_event(
+    state: &mut QConnectQueueState,
+    event: &QueueEvent,
+    now_ms: u64,
+) -> ReducerOutcome {
+    match event {
+        QueueEvent::QueueStateReplaced { state: next, .. } => {
+            *state = next.clone();
+            state.updated_at_ms = now_ms;
+            ReducerOutcome::changed("SRVR_CTRL_QUEUE_STATE", true)
+        }
+        QueueEvent::TracksAdded {
+            version,
+            tracks,
+            shuffle_seed,
+            autoplay_reset,
+            autoplay_loading,
+            ..
+        } => {
+            let start_idx = state.queue_items.len();
+            state.queue_items.extend(tracks.iter().cloned());
+
+            if state.shuffle_mode {
+                if let Some(seed) = shuffle_seed {
+                    state.shuffle_order =
+                        Some(build_shuffle_order(state.queue_items.len(), *seed, None));
+                } else if let Some(order) = state.shuffle_order.as_mut() {
+                    for idx in start_idx..state.queue_items.len() {
+                        order.push(idx);
+                    }
+                }
+            }
+
+            if *autoplay_reset {
+                state.autoplay_items.clear();
+            }
+            state.autoplay_loading = *autoplay_loading;
+
+            let version_changed = state.version != *version;
+            state.version = *version;
+            state.updated_at_ms = now_ms;
+            ReducerOutcome::changed("SRVR_CTRL_QUEUE_TRACKS_ADDED", version_changed)
+        }
+        QueueEvent::TracksLoaded {
+            version,
+            tracks,
+            queue_position,
+            shuffle_mode,
+            shuffle_seed,
+            shuffle_pivot_queue_item_id,
+            autoplay_reset,
+            autoplay_loading,
+            ..
+        } => {
+            state.queue_items = tracks.clone();
+
+            if let Some(enabled) = shuffle_mode {
+                state.shuffle_mode = *enabled;
+            }
+
+            if state.shuffle_mode {
+                let seed = shuffle_seed.unwrap_or(0);
+                let pivot_index = shuffle_pivot_queue_item_id
+                    .and_then(|pivot_id| {
+                        state
+                            .queue_items
+                            .iter()
+                            .position(|item| item.queue_item_id == pivot_id)
+                    })
+                    .or_else(|| queue_position.and_then(|value| usize::try_from(value).ok()));
+                state.shuffle_order = Some(build_shuffle_order(
+                    state.queue_items.len(),
+                    seed,
+                    pivot_index,
+                ));
+            } else {
+                state.shuffle_order = None;
+            }
+
+            if *autoplay_reset {
+                state.autoplay_items.clear();
+            }
+            state.autoplay_loading = *autoplay_loading;
+
+            let version_changed = state.version != *version;
+            state.version = *version;
+            state.updated_at_ms = now_ms;
+            ReducerOutcome::changed("SRVR_CTRL_QUEUE_TRACKS_LOADED", version_changed)
+        }
+        QueueEvent::TracksInserted {
+            version,
+            tracks,
+            insert_after,
+            shuffle_seed,
+            autoplay_reset,
+            autoplay_loading,
+            ..
+        } => {
+            let insert_index = insertion_index(&state.queue_items, *insert_after);
+            let insert_count = tracks.len();
+            let mut next_queue =
+                Vec::with_capacity(state.queue_items.len().saturating_add(insert_count));
+            next_queue.extend_from_slice(&state.queue_items[..insert_index]);
+            next_queue.extend(tracks.iter().cloned());
+            next_queue.extend_from_slice(&state.queue_items[insert_index..]);
+            state.queue_items = next_queue;
+
+            if state.shuffle_mode {
+                if let Some(seed) = shuffle_seed {
+                    state.shuffle_order =
+                        Some(build_shuffle_order(state.queue_items.len(), *seed, None));
+                } else if let Some(order) = state.shuffle_order.as_mut() {
+                    for idx in order.iter_mut() {
+                        if *idx >= insert_index {
+                            *idx += insert_count;
+                        }
+                    }
+                    for idx in insert_index..insert_index + insert_count {
+                        order.push(idx);
+                    }
+                }
+            }
+
+            if *autoplay_reset {
+                state.autoplay_items.clear();
+            }
+            state.autoplay_loading = *autoplay_loading;
+
+            let version_changed = state.version != *version;
+            state.version = *version;
+            state.updated_at_ms = now_ms;
+            ReducerOutcome::changed("SRVR_CTRL_QUEUE_TRACKS_INSERTED", version_changed)
+        }
+        QueueEvent::TracksRemoved {
+            version,
+            queue_item_ids,
+            autoplay_reset,
+            autoplay_loading,
+            ..
+        } => {
+            let before = state.queue_items.clone();
+            let to_remove: HashSet<u64> = queue_item_ids.iter().copied().collect();
+            let old_len = before.len();
+
+            let mut old_to_new = vec![None; old_len];
+            let mut next_queue = Vec::with_capacity(old_len);
+            for (old_idx, item) in before.iter().cloned().enumerate() {
+                if !to_remove.contains(&item.queue_item_id) {
+                    old_to_new[old_idx] = Some(next_queue.len());
+                    next_queue.push(item);
+                }
+            }
+
+            let removed = old_len != next_queue.len();
+            state.queue_items = next_queue;
+
+            if let Some(order) = state.shuffle_order.as_mut() {
+                let mapped: Vec<usize> = order
+                    .iter()
+                    .filter_map(|old_idx| old_to_new.get(*old_idx).and_then(|value| *value))
+                    .collect();
+                *order = mapped;
+            }
+
+            if *autoplay_reset {
+                state.autoplay_items.clear();
+            }
+            state.autoplay_loading = *autoplay_loading;
+
+            let version_changed = state.version != *version;
+            state.version = *version;
+            state.updated_at_ms = now_ms;
+
+            if removed {
+                ReducerOutcome::changed("SRVR_CTRL_QUEUE_TRACKS_REMOVED", version_changed)
+            } else {
+                ReducerOutcome::unchanged("SRVR_CTRL_QUEUE_TRACKS_REMOVED")
+            }
+        }
+        QueueEvent::TracksReordered {
+            version,
+            queue_item_ids,
+            insert_after,
+            autoplay_reset,
+            autoplay_loading,
+            ..
+        } => {
+            let before = state.queue_items.clone();
+            let to_move: HashSet<u64> = queue_item_ids.iter().copied().collect();
+            let mut by_id: HashMap<u64, QueueItem> = before
+                .iter()
+                .cloned()
+                .map(|item| (item.queue_item_id, item))
+                .collect();
+
+            let moving: Vec<QueueItem> = queue_item_ids
+                .iter()
+                .filter_map(|id| by_id.remove(id))
+                .collect();
+            let mut remaining: Vec<QueueItem> = before
+                .iter()
+                .filter(|item| !to_move.contains(&item.queue_item_id))
+                .cloned()
+                .collect();
+            let insert_index = insertion_index(&remaining, *insert_after);
+
+            let reordered = !moving.is_empty();
+            if reordered {
+                remaining.splice(insert_index..insert_index, moving);
+                state.queue_items = remaining;
+
+                if let Some(order) = state.shuffle_order.as_mut() {
+                    let id_by_old_index: HashMap<usize, u64> = before
+                        .iter()
+                        .enumerate()
+                        .map(|(idx, item)| (idx, item.queue_item_id))
+                        .collect();
+                    let new_index_by_id: HashMap<u64, usize> = state
+                        .queue_items
+                        .iter()
+                        .enumerate()
+                        .map(|(idx, item)| (item.queue_item_id, idx))
+                        .collect();
+
+                    let mapped: Vec<usize> = order
+                        .iter()
+                        .filter_map(|old_idx| id_by_old_index.get(old_idx))
+                        .filter_map(|id| new_index_by_id.get(id))
+                        .copied()
+                        .collect();
+                    *order = mapped;
+                }
+            }
+
+            if *autoplay_reset {
+                state.autoplay_items.clear();
+            }
+            state.autoplay_loading = *autoplay_loading;
+
+            let version_changed = state.version != *version;
+            state.version = *version;
+            state.updated_at_ms = now_ms;
+
+            if reordered {
+                ReducerOutcome::changed("SRVR_CTRL_QUEUE_TRACKS_REORDERED", version_changed)
+            } else {
+                ReducerOutcome::unchanged("SRVR_CTRL_QUEUE_TRACKS_REORDERED")
+            }
+        }
+        QueueEvent::QueueCleared { version, .. } => {
+            state.queue_items.clear();
+            state.autoplay_items.clear();
+            state.shuffle_mode = false;
+            state.shuffle_order = None;
+            state.autoplay_loading = false;
+            let version_changed = state.version != *version;
+            state.version = *version;
+            state.updated_at_ms = now_ms;
+            ReducerOutcome::changed("SRVR_CTRL_QUEUE_CLEARED", version_changed)
+        }
+        QueueEvent::ShuffleModeSet {
+            version,
+            shuffle_mode,
+            shuffle_seed,
+            shuffle_pivot_queue_item_id,
+            autoplay_reset,
+            autoplay_loading,
+            ..
+        } => {
+            state.shuffle_mode = *shuffle_mode;
+            if *shuffle_mode {
+                let seed = shuffle_seed.unwrap_or(0);
+                let pivot_index = shuffle_pivot_queue_item_id.and_then(|pivot_id| {
+                    state
+                        .queue_items
+                        .iter()
+                        .position(|item| item.queue_item_id == pivot_id)
+                });
+                state.shuffle_order = Some(build_shuffle_order(
+                    state.queue_items.len(),
+                    seed,
+                    pivot_index,
+                ));
+            } else {
+                state.shuffle_order = None;
+            }
+
+            if *autoplay_reset {
+                state.autoplay_items.clear();
+            }
+            state.autoplay_loading = *autoplay_loading;
+
+            let version_changed = state.version != *version;
+            state.version = *version;
+            state.updated_at_ms = now_ms;
+            ReducerOutcome::changed("SRVR_CTRL_SHUFFLE_MODE_SET", version_changed)
+        }
+        QueueEvent::AutoplayModeSet {
+            version,
+            autoplay_mode,
+            autoplay_reset,
+            autoplay_loading,
+            ..
+        } => {
+            state.autoplay_mode = *autoplay_mode;
+            state.autoplay_loading = *autoplay_loading;
+            if *autoplay_reset {
+                state.autoplay_items.clear();
+            }
+
+            let version_changed = state.version != *version;
+            state.version = *version;
+            state.updated_at_ms = now_ms;
+            ReducerOutcome::changed("SRVR_CTRL_AUTOPLAY_MODE_SET", version_changed)
+        }
+        QueueEvent::AutoplayTracksLoaded {
+            version, tracks, ..
+        } => {
+            state.autoplay_items = tracks.clone();
+            let version_changed = state.version != *version;
+            state.version = *version;
+            state.updated_at_ms = now_ms;
+            ReducerOutcome::changed("SRVR_CTRL_AUTOPLAY_TRACKS_LOADED", version_changed)
+        }
+        QueueEvent::AutoplayTracksRemoved {
+            version,
+            queue_item_ids,
+            ..
+        } => {
+            let to_remove: HashSet<u64> = queue_item_ids.iter().copied().collect();
+            let before = state.autoplay_items.len();
+            state
+                .autoplay_items
+                .retain(|track| !to_remove.contains(&track.queue_item_id));
+            let changed = before != state.autoplay_items.len();
+
+            let version_changed = state.version != *version;
+            state.version = *version;
+            state.updated_at_ms = now_ms;
+
+            if changed {
+                ReducerOutcome::changed("SRVR_CTRL_AUTOPLAY_TRACKS_REMOVED", version_changed)
+            } else {
+                ReducerOutcome::unchanged("SRVR_CTRL_AUTOPLAY_TRACKS_REMOVED")
+            }
+        }
+        QueueEvent::QueueError { version, .. } => {
+            if let Some(version) = version {
+                let version_changed = state.version != *version;
+                state.version = *version;
+                state.updated_at_ms = now_ms;
+                return ReducerOutcome {
+                    queue_changed: false,
+                    version_changed,
+                    event_name: "SRVR_CTRL_QUEUE_ERROR_MESSAGE",
+                };
+            }
+            ReducerOutcome::unchanged("SRVR_CTRL_QUEUE_ERROR_MESSAGE")
+        }
+    }
+}
+
+fn insertion_index(items: &[QueueItem], insert_after: Option<u64>) -> usize {
+    insert_after
+        .and_then(|queue_item_id| {
+            items
+                .iter()
+                .position(|item| item.queue_item_id == queue_item_id)
+                .map(|idx| idx + 1)
+        })
+        .unwrap_or(items.len())
+}
+
+fn build_shuffle_order(count: usize, seed: u64, pivot_index: Option<usize>) -> Vec<usize> {
+    let mut order: Vec<usize> = (0..count).collect();
+    if count <= 1 {
+        return order;
+    }
+
+    let mut rng = if seed == 0 {
+        0x9e37_79b9_7f4a_7c15
+    } else {
+        seed
+    };
+
+    for idx in (1..count).rev() {
+        rng = rng.wrapping_mul(6364136223846793005).wrapping_add(1);
+        let swap_idx = (rng % ((idx + 1) as u64)) as usize;
+        order.swap(idx, swap_idx);
+    }
+
+    if let Some(pivot) = pivot_index.filter(|value| *value < count) {
+        if let Some(pos) = order.iter().position(|value| *value == pivot) {
+            order.rotate_left(pos);
+        }
+    }
+
+    order
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::queue::{QueueEvent, QueueVersion};
+
+    fn item(id: u64) -> QueueItem {
+        QueueItem {
+            track_context_uuid: "ctx".to_string(),
+            track_id: id,
+            queue_item_id: id,
+        }
+    }
+
+    #[test]
+    fn tracks_loaded_replaces_queue_and_builds_shuffle() {
+        let mut state = QConnectQueueState::default();
+        let event = QueueEvent::TracksLoaded {
+            action_uuid: None,
+            version: QueueVersion::new(1, 1),
+            tracks: vec![item(10), item(20), item(30)],
+            queue_position: Some(1),
+            shuffle_mode: Some(true),
+            shuffle_seed: Some(42),
+            shuffle_pivot_queue_item_id: Some(20),
+            autoplay_reset: true,
+            autoplay_loading: true,
+        };
+
+        let outcome = apply_event(&mut state, &event, 1234);
+        assert!(outcome.queue_changed);
+        assert_eq!(state.queue_items.len(), 3);
+        assert!(state.shuffle_mode);
+        assert!(state.shuffle_order.is_some());
+        assert!(state.autoplay_loading);
+    }
+
+    #[test]
+    fn tracks_removed_updates_queue_and_shuffle() {
+        let mut state = QConnectQueueState {
+            queue_items: vec![item(1), item(2), item(3)],
+            shuffle_mode: true,
+            shuffle_order: Some(vec![2, 1, 0]),
+            ..Default::default()
+        };
+
+        let event = QueueEvent::TracksRemoved {
+            action_uuid: None,
+            version: QueueVersion::new(2, 0),
+            queue_item_ids: vec![2],
+            autoplay_reset: false,
+            autoplay_loading: false,
+        };
+
+        let outcome = apply_event(&mut state, &event, 22);
+        assert!(outcome.queue_changed);
+        assert_eq!(state.queue_items.len(), 2);
+        assert_eq!(state.queue_items[0].queue_item_id, 1);
+        assert_eq!(state.queue_items[1].queue_item_id, 3);
+        assert_eq!(state.shuffle_order, Some(vec![1, 0]));
+    }
+
+    #[test]
+    fn tracks_reordered_moves_subset() {
+        let mut state = QConnectQueueState {
+            queue_items: vec![item(1), item(2), item(3), item(4)],
+            shuffle_mode: true,
+            shuffle_order: Some(vec![0, 1, 2, 3]),
+            ..Default::default()
+        };
+
+        let event = QueueEvent::TracksReordered {
+            action_uuid: None,
+            version: QueueVersion::new(3, 0),
+            queue_item_ids: vec![4],
+            insert_after: Some(1),
+            autoplay_reset: false,
+            autoplay_loading: false,
+        };
+
+        let outcome = apply_event(&mut state, &event, 33);
+        assert!(outcome.queue_changed);
+        assert_eq!(
+            state
+                .queue_items
+                .iter()
+                .map(|entry| entry.queue_item_id)
+                .collect::<Vec<_>>(),
+            vec![1, 4, 2, 3]
+        );
+    }
+}

crates/qconnect-core/src/telemetry.rs

@@ -0,0 +1,12 @@
+pub const EVENT_PENDING_ACTION_STARTED: &str = "pending_action_started";
+pub const EVENT_PENDING_ACTION_COMPLETED: &str = "pending_action_completed";
+pub const EVENT_PENDING_ACTION_TIMEOUT: &str = "pending_action_timeout";
+pub const EVENT_PENDING_ACTION_CANCELED_BY_CONCURRENT_REMOTE_EVENT: &str =
+    "pending_action_canceled_by_concurrent_remote_event";
+pub const EVENT_QUEUE_EVENT_UUID_MATCH: &str = "queue_event_uuid_match";
+pub const EVENT_QUEUE_EVENT_UUID_MISMATCH: &str = "queue_event_uuid_mismatch";
+pub const EVENT_QUEUE_RESYNC_TRIGGERED: &str = "queue_resync_triggered";
+
+pub fn queue_reducer_event_name(event_name: &str) -> String {
+    format!("queue_reducer_applied_{event_name}")
+}