···6262 SetPlaylistEnabled(bool),
6363 PublishState,
6464 Shutdown,
6565+ MoveQueueItem {
6666+ from: usize,
6767+ to: usize,
6868+ },
6969+ ShuffleQueue,
6570}
66716772/// Per-room event-loop actor. Owns all mutable state.
···399404 }
400405 RoomCommand::PublishState => self.handle_publish_state().await,
401406 RoomCommand::Shutdown => self.handle_shutdown().await,
407407+ RoomCommand::MoveQueueItem { from, to } => self.handle_move_queue_item(from, to).await,
408408+ RoomCommand::ShuffleQueue => self.handle_shuffle_queue().await,
402409 }
403410 }
404411···670677 if effects.is_empty() {
671678 return false;
672679 }
680680+ self.execute_effects(effects).await;
681681+ false
682682+ }
683683+684684+ /// Reorder a single track in the queue. Out-of-range or no-op moves
685685+ /// return no effects from the state machine; everything is persisted
686686+ /// before the snapshot goes out.
687687+ async fn handle_move_queue_item(&mut self, from: usize, to: usize) -> bool {
688688+ let effects = self.state.transition(&Event::MoveQueueItem { from, to }, 0);
689689+ if effects.is_empty() {
690690+ return false;
691691+ }
692692+ self.persist_effects(&effects).await;
693693+ self.execute_effects(effects).await;
694694+ false
695695+ }
696696+697697+ /// Shuffle the queue using a server-picked seed. The seed comes from the
698698+ /// wall clock so concurrent shuffles are independent, and clients can't
699699+ /// predict or replay each other's shuffles.
700700+ async fn handle_shuffle_queue(&mut self) -> bool {
701701+ let seed = Utc::now().timestamp_nanos_opt().unwrap_or(0) as u64;
702702+ let effects = self.state.transition(&Event::ShuffleQueue { seed }, 0);
703703+ if effects.is_empty() {
704704+ return false;
705705+ }
706706+ self.persist_effects(&effects).await;
673707 self.execute_effects(effects).await;
674708 false
675709 }
+262-7
src/state.rs
···9090 PlaylistEntryRemoved(PlaylistEntryId),
9191 /// Toggle the perpetual playlist on/off.
9292 SetPlaylistEnabled(bool),
9393+ /// Reorder a track within the up-next queue.
9494+ /// `from` and `to` are 0-based queue indices. `from == to` is a no-op.
9595+ /// Indices out of range or queue empty are no-ops.
9696+ MoveQueueItem { from: usize, to: usize },
9797+ /// Shuffle the up-next queue deterministically. Same seed → same result.
9898+ /// Empty/single-item queues are no-ops. Active track is not affected.
9999+ ShuffleQueue { seed: u64 },
93100}
9410195102/// Side effects to execute after a transition.
···135142 },
136143 /// Remove a playlist entry.
137144 RemovePlaylistEntry(PlaylistEntryId),
145145+ /// Renumber queue positions to match `new_order`. Emitted by both
146146+ /// move and shuffle operations; the store updates the `position`
147147+ /// column for every queued track so a fresh `SELECT ... ORDER BY
148148+ /// position` returns the new order.
149149+ ReorderQueue { new_order: Vec<TrackId> },
138150}
139151140152/// Pure playback state: no IO handles, no DB connections.
···189201 Event::PlaylistEntryAdded(entry) => self.handle_playlist_entry_added(entry),
190202 Event::PlaylistEntryRemoved(id) => self.handle_playlist_entry_removed(*id),
191203 Event::SetPlaylistEnabled(enabled) => self.handle_set_playlist_enabled(*enabled),
204204+ Event::MoveQueueItem { from, to } => self.handle_move_queue_item(*from, *to),
205205+ Event::ShuffleQueue { seed } => self.handle_shuffle_queue(*seed),
192206 }
193207 }
194208···488502 Some(entry)
489503 }
490504505505+ /// Reorder a track within the up-next queue. `from` and `to` are
506506+ /// 0-based queue indices. The active track is not affected. No-op if
507507+ /// either index is out of range, the queue is empty, or `from == to`.
508508+ fn handle_move_queue_item(&mut self, from: usize, to: usize) -> Vec<Effect> {
509509+ if from == to || from >= self.queue.len() || to >= self.queue.len() {
510510+ return Vec::new();
511511+ }
512512+ let track = self.queue.remove(from);
513513+ self.queue.insert(to, track);
514514+ let new_order: Vec<TrackId> = self.queue.iter().map(|t| t.id).collect();
515515+ vec![Effect::ReorderQueue { new_order }, Effect::PublishSnapshot]
516516+ }
517517+518518+ /// Shuffle the up-next queue deterministically. Same seed → same order.
519519+ /// Fisher-Yates with a splitmix64 PRNG seeded by the caller.
520520+ /// No-op for queues with fewer than 2 items.
521521+ fn handle_shuffle_queue(&mut self, seed: u64) -> Vec<Effect> {
522522+ if self.queue.len() < 2 {
523523+ return Vec::new();
524524+ }
525525+ let mut rng = SplitMix64::new(seed);
526526+ let n = self.queue.len();
527527+ // Fisher-Yates: walk from the end, swap each element with a random earlier one.
528528+ for i in (1..n).rev() {
529529+ let j = (rng.next() as usize) % (i + 1);
530530+ self.queue.swap(i, j);
531531+ }
532532+ let new_order: Vec<TrackId> = self.queue.iter().map(|t| t.id).collect();
533533+ vec![Effect::ReorderQueue { new_order }, Effect::PublishSnapshot]
534534+ }
535535+491536 /// Pop the first track from the queue and start playing it.
492537 ///
493538 /// Returns effects for starting the pipeline and publishing state.
···557602 }
558603559604 (state, effects)
605605+ }
606606+}
607607+608608+/// Tiny splitmix64 PRNG — deterministic, no dependencies, good enough for
609609+/// shuffling a user-visible queue. Not cryptographic.
610610+struct SplitMix64(u64);
611611+612612+impl SplitMix64 {
613613+ fn new(seed: u64) -> Self {
614614+ Self(seed)
615615+ }
616616+617617+ fn next(&mut self) -> u64 {
618618+ self.0 = self.0.wrapping_add(0x9E3779B97F4A7C15);
619619+ let mut z = self.0;
620620+ z = (z ^ (z >> 30)).wrapping_mul(0xBF58476D1CE4E5B9);
621621+ z = (z ^ (z >> 27)).wrapping_mul(0x94D049BB133111EB);
622622+ z ^ (z >> 31)
560623 }
561624}
562625···14721535 // Active track should be the queued one.
14731536 assert_eq!(s.active.as_ref().unwrap().title, "PL");
14741537 }
15381538+15391539+ // --- queue reorder / shuffle ---
15401540+15411541+ fn qid(n: u128) -> QueuedTrack {
15421542+ QueuedTrack {
15431543+ id: track_id(n),
15441544+ title: format!("Q{n}"),
15451545+ url: format!("https://example.com/q{n}"),
15461546+ duration: "1:00".into(),
15471547+ thumbnail: None,
15481548+ pending: false,
15491549+ }
15501550+ }
15511551+15521552+ fn queue_with(n: usize) -> (PlaybackState, Vec<TrackId>) {
15531553+ // Queue `n` tracks. The first becomes active (idle room); the rest
15541554+ // sit in the queue. Total TrackQueued events = n + 1.
15551555+ let mut s = PlaybackState::default();
15561556+ let active = qid(0);
15571557+ let mut ids = vec![active.id];
15581558+ s.transition(&Event::TrackQueued(active), 0);
15591559+ for i in 0..n {
15601560+ let t = qid((i + 1) as u128);
15611561+ ids.push(t.id);
15621562+ s.transition(&Event::TrackQueued(t), 0);
15631563+ }
15641564+ (s, ids)
15651565+ }
15661566+15671567+ #[test]
15681568+ fn test_move_queue_item_basic() {
15691569+ // Queue (after queue_with): [B, C, D]. Active is A. Move from 0 to 2 → [C, D, B].
15701570+ let (mut s, ids) = queue_with(3);
15711571+ let effects = s.transition(&Event::MoveQueueItem { from: 0, to: 2 }, 0);
15721572+ assert_eq!(s.queue[0].id, ids[2]);
15731573+ assert_eq!(s.queue[1].id, ids[3]);
15741574+ assert_eq!(s.queue[2].id, ids[1]);
15751575+ // Effect carries the new queue order so the store can renumber.
15761576+ assert!(effects.iter().any(|e| matches!(
15771577+ e,
15781578+ Effect::ReorderQueue { new_order } if new_order == &vec![ids[2], ids[3], ids[1]]
15791579+ )));
15801580+ assert!(effects.iter().any(|e| matches!(e, Effect::PublishSnapshot)));
15811581+ }
15821582+15831583+ #[test]
15841584+ fn test_move_queue_item_same_index_is_noop() {
15851585+ let (mut s, ids) = queue_with(3);
15861586+ let effects = s.transition(&Event::MoveQueueItem { from: 1, to: 1 }, 0);
15871587+ assert_eq!(s.queue[0].id, ids[1]);
15881588+ assert!(effects.is_empty());
15891589+ }
15901590+15911591+ #[test]
15921592+ fn test_move_queue_item_out_of_range_is_noop() {
15931593+ let (mut s, _ids) = queue_with(3);
15941594+ let effects = s.transition(&Event::MoveQueueItem { from: 5, to: 0 }, 0);
15951595+ assert_eq!(s.queue.len(), 3);
15961596+ assert!(effects.is_empty());
15971597+ let effects = s.transition(&Event::MoveQueueItem { from: 0, to: 5 }, 0);
15981598+ assert!(effects.is_empty());
15991599+ }
16001600+16011601+ #[test]
16021602+ fn test_move_queue_item_empty_queue_is_noop() {
16031603+ let mut s = PlaybackState::default();
16041604+ let effects = s.transition(&Event::MoveQueueItem { from: 0, to: 0 }, 0);
16051605+ assert!(effects.is_empty());
16061606+ }
16071607+16081608+ #[test]
16091609+ fn test_shuffle_queue_deterministic() {
16101610+ // Same seed → same order. Run twice and compare.
16111611+ let (mut s1, _) = queue_with(8);
16121612+ let (mut s2, _) = queue_with(8);
16131613+ s1.transition(&Event::ShuffleQueue { seed: 42 }, 0);
16141614+ s2.transition(&Event::ShuffleQueue { seed: 42 }, 0);
16151615+ let order1: Vec<TrackId> = s1.queue.iter().map(|t| t.id).collect();
16161616+ let order2: Vec<TrackId> = s2.queue.iter().map(|t| t.id).collect();
16171617+ assert_eq!(order1, order2);
16181618+ }
16191619+16201620+ #[test]
16211621+ fn test_shuffle_queue_different_seeds_diverge() {
16221622+ // Two different seeds → almost certainly different orders.
16231623+ // (Birthday-paradox: collision chance with 8 items is ~1/2000.)
16241624+ let (mut s1, _) = queue_with(8);
16251625+ let (mut s2, _) = queue_with(8);
16261626+ s1.transition(&Event::ShuffleQueue { seed: 1 }, 0);
16271627+ s2.transition(&Event::ShuffleQueue { seed: 2 }, 0);
16281628+ let order1: Vec<TrackId> = s1.queue.iter().map(|t| t.id).collect();
16291629+ let order2: Vec<TrackId> = s2.queue.iter().map(|t| t.id).collect();
16301630+ assert_ne!(order1, order2);
16311631+ }
16321632+16331633+ #[test]
16341634+ fn test_shuffle_queue_preserves_count() {
16351635+ let (mut s, ids) = queue_with(10);
16361636+ s.transition(&Event::ShuffleQueue { seed: 99 }, 0);
16371637+ assert_eq!(s.queue.len(), 10);
16381638+ // ids[0] is the active track; ids[1..] are the queue.
16391639+ let mut sorted: Vec<_> = s.queue.iter().map(|t| t.id).collect();
16401640+ sorted.sort_by_key(|id| id.0.as_u128());
16411641+ let mut expected: Vec<_> = ids[1..].to_vec();
16421642+ expected.sort_by_key(|id| id.0.as_u128());
16431643+ assert_eq!(sorted, expected);
16441644+ }
16451645+16461646+ #[test]
16471647+ fn test_shuffle_queue_singleton_is_noop() {
16481648+ let (mut s, ids) = queue_with(1);
16491649+ let effects = s.transition(&Event::ShuffleQueue { seed: 0 }, 0);
16501650+ assert_eq!(s.queue[0].id, ids[1]);
16511651+ assert!(effects.is_empty());
16521652+ }
16531653+16541654+ #[test]
16551655+ fn test_shuffle_queue_empty_is_noop() {
16561656+ let mut s = PlaybackState::default();
16571657+ let effects = s.transition(&Event::ShuffleQueue { seed: 0 }, 0);
16581658+ assert!(effects.is_empty());
16591659+ }
16601660+16611661+ #[test]
16621662+ fn test_shuffle_queue_does_not_affect_active() {
16631663+ // Active track stays put; the shuffle permutes only the queue Vec.
16641664+ let (mut s, _ids) = queue_with(5);
16651665+ let original_active = s.active.as_ref().unwrap().id;
16661666+ s.transition(&Event::ShuffleQueue { seed: 7 }, 0);
16671667+ assert_eq!(s.active.as_ref().unwrap().id, original_active);
16681668+ assert_eq!(s.queue.len(), 5);
16691669+ }
14751670}
1476167114771672/// Deterministic simulation testing: random event sequences with invariant
···15061701 MetadataUpdated(u64),
15071702 DownloadResolved(u64),
15081703 DownloadFailed(u64),
17041704+ MoveQueueItem { from: u8, to: u8 },
17051705+ ShuffleQueue { seed: u64 },
15091706 }
1510170715111708 fn op_strategy() -> impl Strategy<Value = Vec<Op>> {
17091709+ // Move and shuffle are weighted higher than the other ops so the new
17101710+ // handlers get meaningful coverage in the 1024-case run.
15121711 let op = prop_oneof![
15131513- (0u64..12).prop_map(Op::Queue),
15141514- Just(Op::Skip),
15151515- (0u64..12).prop_map(Op::TrackEnded),
15161516- (0u64..12).prop_map(Op::TrackFailed),
15171517- (0u64..12).prop_map(Op::MetadataUpdated),
15181518- (0u64..12).prop_map(Op::DownloadResolved),
15191519- (0u64..12).prop_map(Op::DownloadFailed),
17121712+ 2 => (0u64..12).prop_map(Op::Queue),
17131713+ 1 => Just(Op::Skip),
17141714+ 1 => (0u64..12).prop_map(Op::TrackEnded),
17151715+ 1 => (0u64..12).prop_map(Op::TrackFailed),
17161716+ 1 => (0u64..12).prop_map(Op::MetadataUpdated),
17171717+ 1 => (0u64..12).prop_map(Op::DownloadResolved),
17181718+ 1 => (0u64..12).prop_map(Op::DownloadFailed),
17191719+ 3 => (0u8..16, 0u8..16).prop_map(|(from, to)| Op::MoveQueueItem { from, to }),
17201720+ 2 => (0u64..1000).prop_map(|seed| Op::ShuffleQueue { seed }),
15201721 ];
15211722 proptest::collection::vec(op, 0..50)
15221723 }
···16221823 Some(&crate::state::MediaStatus::Failed(format!("err-{n}"))),
16231824 "DownloadFailed({url}) did not record Failed in media_cache",
16241825 );
18261826+ effects
18271827+ }
18281828+ Op::MoveQueueItem { from, to } => {
18291829+ // The handler is a no-op for out-of-range or equal indices.
18301830+ let queue_len = state.queue.len();
18311831+ let from_in = (*from as usize) < queue_len;
18321832+ let to_in = (*to as usize) < queue_len;
18331833+ let changed = from_in && to_in && from != to;
18341834+ let expected_ids: Vec<_> = if changed {
18351835+ let mut q = state.queue.clone();
18361836+ let t = q.remove(*from as usize);
18371837+ q.insert(*to as usize, t);
18381838+ q.into_iter().map(|t| t.id).collect()
18391839+ } else {
18401840+ state.queue.iter().map(|t| t.id).collect()
18411841+ };
18421842+ let effects = state.transition(
18431843+ &Event::MoveQueueItem {
18441844+ from: *from as usize,
18451845+ to: *to as usize,
18461846+ },
18471847+ 0,
18481848+ );
18491849+ let actual_ids: Vec<_> = state.queue.iter().map(|t| t.id).collect();
18501850+ assert_eq!(
18511851+ actual_ids, expected_ids,
18521852+ "MoveQueueItem {from},{to} produced wrong order",
18531853+ );
18541854+ if changed {
18551855+ assert!(effects.iter().any(|e| matches!(e, Effect::PublishSnapshot)));
18561856+ } else {
18571857+ assert!(effects.is_empty(), "no-op move should return no effects");
18581858+ }
18591859+ effects
18601860+ }
18611861+ Op::ShuffleQueue { seed } => {
18621862+ // Same seed → same order. Record the order, run the op,
18631863+ // and verify the queue contains the same track IDs
18641864+ // (it's a permutation, not a transformation).
18651865+ let pre: std::collections::HashSet<_> =
18661866+ state.queue.iter().map(|t| t.id).collect();
18671867+ let effects = state.transition(&Event::ShuffleQueue { seed: *seed }, 0);
18681868+ let post: std::collections::HashSet<_> =
18691869+ state.queue.iter().map(|t| t.id).collect();
18701870+ assert_eq!(pre, post, "shuffle must be a permutation");
18711871+ if state.queue.len() >= 2 {
18721872+ assert!(effects.iter().any(|e| matches!(e, Effect::PublishSnapshot)));
18731873+ } else {
18741874+ assert!(effects.is_empty(), "tiny queue should be a no-op");
18751875+ }
16251876 effects
16261877 }
16271878 };
···17972048 }
1798204917992050 proptest! {
20512051+ #![proptest_config(ProptestConfig {
20522052+ cases: 1024,
20532053+ ..ProptestConfig::default()
20542054+ })]
18002055 #[test]
18012056 fn dst_random_sequences_invariants_hold(ops in op_strategy()) {
18022057 apply_ops(&ops);
+28
src/store.rs
···292292 Effect::RemovePlaylistEntry(id) => {
293293 room.playlist.retain(|e| e.id != *id);
294294 }
295295+ Effect::ReorderQueue { new_order } => {
296296+ // The state machine's in-memory `room.queue` is already in
297297+ // the new order; this arm keeps the effect exhaustively
298298+ // matched. The SQLite path handles position renumbering.
299299+ let _ = new_order;
300300+ }
295301 _ => {}
296302 }
297303 }
···590596 .bind(id.as_str_buf())
591597 .execute(&mut *tx)
592598 .await?;
599599+ }
600600+ Effect::ReorderQueue { new_order } => {
601601+ // Renumber every queued track's position to its index in
602602+ // new_order. Limit to un-played, un-started rows so we
603603+ // never touch the active or history rows. Using indexed
604604+ // UPDATEs (one per row) is fine for queues of typical
605605+ // size (< 100).
606606+ for (i, track_id) in new_order.iter().enumerate() {
607607+ sqlx::query(
608608+ "UPDATE tracks
609609+ SET position = ?1
610610+ WHERE id = ?2
611611+ AND room_id = ?3
612612+ AND played = 0
613613+ AND started_at_ms IS NULL",
614614+ )
615615+ .bind(i as i64)
616616+ .bind(track_id.as_str_buf())
617617+ .bind(room_id)
618618+ .execute(&mut *tx)
619619+ .await?;
620620+ }
593621 }
594622 _ => {}
595623 }
+24-3
src/transport.rs
···2626#[derive(Deserialize)]
2727#[serde(tag = "type", rename_all = "snake_case")]
2828enum WsClientMessage {
2929- Chat { content: String },
2929+ Chat {
3030+ content: String,
3131+ },
3032 Skip,
3131- TrackEnded { item_id: TrackId },
3232- SetPlaylistEnabled { enabled: bool },
3333+ TrackEnded {
3434+ item_id: TrackId,
3535+ },
3636+ SetPlaylistEnabled {
3737+ enabled: bool,
3838+ },
3339 Ping,
4040+ /// Move a queue item from one position to another. 0-indexed.
4141+ MoveQueueItem {
4242+ from: usize,
4343+ to: usize,
4444+ },
4545+ /// Shuffle the entire queue (server picks the seed).
4646+ ShuffleQueue,
3447}
35483649/// Run a WebSocket session for a single client.
···127140 }
128141 WsClientMessage::Ping => {
129142 let _ = ws_tx.send(Message::Ping(Bytes::new())).await;
143143+ }
144144+ WsClientMessage::MoveQueueItem { from, to } => {
145145+ let _ = cmd_tx
146146+ .send(RoomCommand::MoveQueueItem { from, to })
147147+ .await;
148148+ }
149149+ WsClientMessage::ShuffleQueue => {
150150+ let _ = cmd_tx.send(RoomCommand::ShuffleQueue).await;
130151 }
131152 }
132153 }