diff --git a/crates/qconnect-core/Cargo.toml b/crates/qconnect-core/Cargo.toml new file mode 100644 index 000000000..97b94634a --- /dev/null +++ b/crates/qconnect-core/Cargo.toml @@ -0,0 +1,10 @@ +[package] +name = "qconnect-core" +version.workspace = true +edition.workspace = true +license.workspace = true +description = "UI-agnostic domain core for Qobuz Connect queue and concurrency" + +[dependencies] +serde = { workspace = true } +thiserror = { workspace = true } diff --git a/crates/qconnect-core/src/admission.rs b/crates/qconnect-core/src/admission.rs new file mode 100644 index 000000000..21b34ba3b --- /dev/null +++ b/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 + } + } +} diff --git a/crates/qconnect-core/src/lib.rs b/crates/qconnect-core/src/lib.rs new file mode 100644 index 000000000..8332a0955 --- /dev/null +++ b/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}; diff --git a/crates/qconnect-core/src/pending.rs b/crates/qconnect-core/src/pending.rs new file mode 100644 index 000000000..bf952418a --- /dev/null +++ b/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, +} + +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 { + self.current.take() + } +} diff --git a/crates/qconnect-core/src/queue.rs b/crates/qconnect-core/src/queue.rs new file mode 100644 index 000000000..34c60b967 --- /dev/null +++ b/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, + pub shuffle_mode: bool, + pub shuffle_order: Option>, + pub autoplay_mode: bool, + pub autoplay_loading: bool, + pub autoplay_items: Vec, + 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, + state: QConnectQueueState, + }, + TracksAdded { + action_uuid: Option, + version: QueueVersion, + tracks: Vec, + shuffle_seed: Option, + autoplay_reset: bool, + autoplay_loading: bool, + }, + TracksLoaded { + action_uuid: Option, + version: QueueVersion, + tracks: Vec, + queue_position: Option, + shuffle_mode: Option, + shuffle_seed: Option, + shuffle_pivot_queue_item_id: Option, + autoplay_reset: bool, + autoplay_loading: bool, + }, + TracksInserted { + action_uuid: Option, + version: QueueVersion, + tracks: Vec, + insert_after: Option, + shuffle_seed: Option, + autoplay_reset: bool, + autoplay_loading: bool, + }, + TracksRemoved { + action_uuid: Option, + version: QueueVersion, + queue_item_ids: Vec, + autoplay_reset: bool, + autoplay_loading: bool, + }, + TracksReordered { + action_uuid: Option, + version: QueueVersion, + queue_item_ids: Vec, + insert_after: Option, + autoplay_reset: bool, + autoplay_loading: bool, + }, + QueueCleared { + action_uuid: Option, + version: QueueVersion, + }, + ShuffleModeSet { + action_uuid: Option, + version: QueueVersion, + shuffle_mode: bool, + shuffle_seed: Option, + shuffle_pivot_queue_item_id: Option, + autoplay_reset: bool, + autoplay_loading: bool, + }, + AutoplayModeSet { + action_uuid: Option, + version: QueueVersion, + autoplay_mode: bool, + autoplay_reset: bool, + autoplay_loading: bool, + }, + AutoplayTracksLoaded { + action_uuid: Option, + version: QueueVersion, + tracks: Vec, + }, + AutoplayTracksRemoved { + action_uuid: Option, + version: QueueVersion, + queue_item_ids: Vec, + }, + QueueError { + action_uuid: Option, + version: Option, + code: String, + message: String, + }, +} diff --git a/crates/qconnect-core/src/reducer.rs b/crates/qconnect-core/src/reducer.rs new file mode 100644 index 000000000..7409e84aa --- /dev/null +++ b/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 = 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 = 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 = queue_item_ids.iter().copied().collect(); + let mut by_id: HashMap = before + .iter() + .cloned() + .map(|item| (item.queue_item_id, item)) + .collect(); + + let moving: Vec = queue_item_ids + .iter() + .filter_map(|id| by_id.remove(id)) + .collect(); + let mut remaining: Vec = 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 = before + .iter() + .enumerate() + .map(|(idx, item)| (idx, item.queue_item_id)) + .collect(); + let new_index_by_id: HashMap = state + .queue_items + .iter() + .enumerate() + .map(|(idx, item)| (item.queue_item_id, idx)) + .collect(); + + let mapped: Vec = 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 = 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) -> 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) -> Vec { + let mut order: Vec = (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![1, 4, 2, 3] + ); + } +} diff --git a/crates/qconnect-core/src/telemetry.rs b/crates/qconnect-core/src/telemetry.rs new file mode 100644 index 000000000..791f7630a --- /dev/null +++ b/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}") +}