From 14c136cd83314d876a7c4fc1cd5381fde88095c5 Mon Sep 17 00:00:00 2001 From: karitham Date: Mon, 1 Jun 2026 16:03:53 +0200 Subject: [PATCH] room: resolve_ref helper accepting UUID or name --- src/room.rs | 149 ++++++++++++++++++++++++++++++++++++++++++++++------ 1 file changed, 134 insertions(+), 15 deletions(-) diff --git a/src/room.rs b/src/room.rs index f627e5f..ebb7187 100644 --- a/src/room.rs +++ b/src/room.rs @@ -10,7 +10,7 @@ //! 2. Transition: `PlaybackState::transition()` (pure) //! 3. Commit: execute returned [`Effect`]s -use std::collections::HashMap; +use std::collections::{HashMap, HashSet}; use std::hash::{Hash, Hasher}; use std::path::PathBuf; use std::sync::Arc; @@ -65,6 +65,13 @@ pub(crate) enum RoomCommand { SetSkipThreshold { percent: u8, }, + /// Cast a skip vote from this user. The actor counts votes and + /// fires `Skip` when the room's threshold is met. `user_id` is + /// the session id assigned at register time; the WebSocket layer + /// attaches it so the actor doesn't need to track sessions. + VoteSkip { + user_id: u64, + }, PublishState, Shutdown, MoveQueueItem { @@ -94,6 +101,9 @@ struct RoomActor { client_count: u64, next_user_id: u64, last_active_tx: watch::Sender>, + /// Users who voted to skip the current active track. Cleared + /// whenever the active track changes. + skip_votes: HashSet, /// Running downloads keyed by URL (one task per URL — deduplicated). /// In-flight tracking only; completed results live in `state.media_cache`. @@ -123,8 +133,10 @@ impl Registry { } } - /// Create a new room with a UUIDv7 ID, persist to store. - pub(crate) async fn create(&self) -> RoomId { + /// Create a new room with a UUIDv7 ID, persist to store. Returns + /// the internal UUID plus the human-readable name so the caller can + /// build a `/r/{name}` URL. + pub(crate) async fn create(&self) -> (RoomId, String) { let id = RoomId::new(); let room_name = generate_room_name(&id.as_str_buf()); let now = Utc::now(); @@ -155,7 +167,7 @@ impl Registry { }, ); - id + (id, room_name) } /// Get the room handle. Rehydrates from store if the room exists but @@ -206,18 +218,16 @@ impl Registry { Some(handle) } - /// Check whether a room exists in the store. - #[must_use] - pub(crate) async fn exists(&self, id: &RoomId) -> bool { - if self.inner.read().await.rooms.contains_key(id) { - return true; + /// Resolve a room reference (UUID or name) to a `RoomId`. Tries + /// UUID parse first for cheapness, then falls back to a name + /// lookup. The public API accepts either, so shared `/r/{name}` + /// links and direct UUID clients both work. + pub(crate) async fn resolve_ref(&self, r: &str) -> Option { + if let Some(id) = RoomId::parse(r) { + return Some(id); } - self.store - .load(&id.as_str_buf()) - .await - .ok() - .flatten() - .is_some() + let id_str = self.store.name_to_id(r).await.ok().flatten()?; + RoomId::parse(&id_str) } /// Load room name + history (for SSR rendering). @@ -343,6 +353,7 @@ fn spawn_actor( client_count: 0, next_user_id: 1, last_active_tx, + skip_votes: HashSet::new(), download_tasks: HashMap::new(), }; tokio::spawn(actor.run()); @@ -411,6 +422,7 @@ impl RoomActor { RoomCommand::SetSkipThreshold { percent } => { self.handle_set_skip_threshold(percent).await } + RoomCommand::VoteSkip { user_id } => self.handle_vote_skip(user_id).await, RoomCommand::PublishState => self.handle_publish_state().await, RoomCommand::Shutdown => self.handle_shutdown().await, RoomCommand::MoveQueueItem { from, to } => self.handle_move_queue_item(from, to).await, @@ -588,6 +600,9 @@ impl RoomActor { } let failed_title = format!("Failed to load: {error}"); + // Capture the current active id so we know whether the failure + // removed it. If so, skip-votes for the old track are stale. + let prev_active = self.state.active.as_ref().map(|a| a.id); for id in &affected { effects.extend(self.state.transition( &Event::MetadataUpdated { @@ -605,6 +620,9 @@ impl RoomActor { .transition(&Event::TrackFailed { item_id: *id }, now), ); } + if prev_active.is_some_and(|id| affected.contains(&id)) { + self.skip_votes.clear(); + } self.persist_effects(&effects).await; self.execute_effects(effects).await; @@ -614,6 +632,8 @@ impl RoomActor { async fn handle_skip(&mut self) -> bool { tracing::info!(room = %self.room_id, "skip requested"); + // Skip always ends the active track; old votes are stale. + self.skip_votes.clear(); let now = Utc::now().timestamp_millis(); let effects = self.state.transition(&Event::Skip, now); self.persist_effects(&effects).await; @@ -624,6 +644,8 @@ impl RoomActor { async fn handle_track_ended(&mut self, item_id: TrackId) -> bool { tracing::debug!(room = %self.room_id, %item_id, "track ended"); + // Track end means the active advanced; old votes are stale. + self.skip_votes.clear(); let now = Utc::now().timestamp_millis(); let effects = self.state.transition(&Event::TrackEnded { item_id }, now); self.persist_effects(&effects).await; @@ -631,6 +653,35 @@ impl RoomActor { self.post_advance().await } + /// Cast a skip vote. Fires `Skip` once `votes * 100 >= + /// threshold * client_count`. Always publishes so clients see the + /// current count. + async fn handle_vote_skip(&mut self, user_id: u64) -> bool { + if self.state.active.is_none() { + // No active track to skip; ignore the vote. + return false; + } + let was_new = self.skip_votes.insert(user_id); + if !was_new { + // Already voted; the count is unchanged. Still publish so + // the user sees their vote is recorded. + self.publish_state_snapshot().await; + return false; + } + let threshold = self.state.skip_threshold as u64; + let count = self.client_count.max(1); + let votes = self.skip_votes.len() as u64; + if vote_threshold_reached(votes, count, threshold) { + // Threshold met: clear votes and fire the skip. The Skip + // handler will publish the snapshot. + self.skip_votes.clear(); + return self.handle_skip().await; + } + // Not yet; just publish the new count. + self.publish_state_snapshot().await; + false + } + async fn handle_add_playlist_entry(&mut self, url: String, added_by: Option) -> bool { let id = PlaylistEntryId::new(); let entry = PlaylistEntry { @@ -784,6 +835,12 @@ impl RoomActor { self.next_user_id = self.next_user_id.wrapping_add(1); self.client_count = self.client_count.saturating_add(1); let _ = resp.send(id); + // If autoplay is on and nothing is playing, draw the next + // playlist track so a fresh joiner doesn't have to wait for the + // idle sweep. + if self.state.autoplay_enabled && self.state.active.is_none() { + let _ = self.maybe_auto_fill().await; + } self.publish_state_snapshot().await; false } @@ -947,6 +1004,7 @@ impl RoomActor { "playlist": self.state.playlist, "autoplay_enabled": self.state.autoplay_enabled, "skip_threshold": self.state.skip_threshold, + "skip_votes": self.skip_votes.len() as u64, "clients": client_count, }); self.publishers.state_tx.send_replace(snapshot); @@ -1067,6 +1125,16 @@ fn format_duration(secs: f64) -> String { } } +/// Test whether a vote count is enough to skip the current track. +/// +/// Strict integer math so 34% with 3 clients needs 2 votes (1*100=100 +/// vs 34*3=102). Integer percentages 0-100; out-of-range is clamped. +fn vote_threshold_reached(votes: u64, client_count: u64, threshold_percent: u64) -> bool { + let count = client_count.max(1); + let threshold = threshold_percent.min(100); + votes * 100 >= threshold * count +} + #[cfg(test)] mod tests { use super::*; @@ -1198,4 +1266,55 @@ mod tests { let entry = state.next_playlist_entry().unwrap(); assert_eq!(entry.id, pid(1)); } + + #[test] + fn vote_threshold_34_pct_of_3_clients_needs_2_votes() { + // 1/3 = 33.3% < 34%, so one vote is not enough. + assert!(!vote_threshold_reached(1, 3, 34)); + // 2/3 = 66.7% >= 34%. + assert!(vote_threshold_reached(2, 3, 34)); + } + + #[test] + fn vote_threshold_50_pct_of_4_clients_needs_2_votes() { + // 2/4 = 50% >= 50%. + assert!(vote_threshold_reached(2, 4, 50)); + // 1/4 = 25% < 50%. + assert!(!vote_threshold_reached(1, 4, 50)); + } + + #[test] + fn vote_threshold_zero_is_immediate() { + // 0% threshold: 0 * 100 = 0 >= 0 * count = 0, so it fires + // immediately. Not useful in practice — the actor should never + // call this with threshold=0 in a way that fires pre-vote. + // The 1-vote case is the meaningful real-world one. + assert!(vote_threshold_reached(1, 5, 0)); + } + + #[test] + fn vote_threshold_hundred_requires_all_votes() { + // 100% threshold: every client must vote. + assert!(!vote_threshold_reached(2, 3, 100)); + assert!(vote_threshold_reached(3, 3, 100)); + } + + #[test] + fn vote_threshold_clamps_out_of_range_percent() { + // >100% treated as 100%. + assert!(vote_threshold_reached(1, 1, 250)); + // Single client with single vote always passes any non-zero + // threshold. + assert!(vote_threshold_reached(1, 1, 1)); + } + + #[test] + fn vote_threshold_handles_zero_clients_as_one() { + // Edge: no clients. Should never happen, but the math must + // not divide by zero. With count=1, votes=0: 0 >= 100*1 = 100 + // is false at 100%, true at 0% and anything in between that + // the votes count is below. + assert!(!vote_threshold_reached(0, 0, 100)); + assert!(!vote_threshold_reached(0, 0, 50)); + } } -- 2.51.2