Something went wrong. Try again.
experiments of a tiny cytube-like player with yt-dlp
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197119811991200120112021203120412051206120712081209121012111212121312141215121612171218121912201221122212231224122512261227122812291230123112321233123412351236123712381239124012411242124312441245124612471248124912501251125212531254125512561257125812591260126112621263126412651266126712681269127012711272127312741275127612771278127912801281128212831284128512861287128812891290129112921293129412951296129712981299130013011302130313041305130613071308130913101311131213131314131513161317131813191320132113221323132413251326//! Room state management.//!//! Each room is identified by a UUIDv7, persists metadata in//! SQLite, and holds ephemeral playback state in memory via a pure//! [`PlaybackState`] machine.//!//! All mutations flow through a per-room mpsc event-loop actor://!//! 1. Gather: fetch data (DB, wall clock)//! 2. Transition: `PlaybackState::transition()` (pure)//! 3. Commit: execute returned [`Effect`]s
use std::collections::{HashMap, HashSet};use std::hash::{Hash, Hasher};use std::path::PathBuf;use std::sync::Arc;use std::time::Duration;
use chrono::{DateTime, Utc};use tokio::sync::{mpsc, watch, RwLock};
use crate::names::generate_room_name;use crate::state::{Effect, Event, PlaybackState, QueuedTrack};use crate::store::{PlaylistEntry, RoomStore};use crate::types::{ ChatMessage, HistoryEntry, PlaylistEntryId, QueueSummary, RoomHandle, RoomId, RoomPublishers, TrackId, TrackState,};
/// All room operations, processed sequentially by the per-room actor.pub(crate) enum RoomCommand { QueueUrl(String), DownloadReady { url: String, media: crate::ingest::IngestedTrack, }, DownloadFailed { url: String, error: String, }, Skip, TrackEnded { item_id: TrackId, }, Chat { user_name: String, content: String, msg_type: String, created_at: String, }, RegisterClient { resp: tokio::sync::oneshot::Sender<u64>, }, UnregisterClient, AddPlaylistEntry { url: String, added_by: Option<String>, }, RemovePlaylistEntry { id: PlaylistEntryId, }, SetAutoplay { enabled: bool, }, 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 { from: usize, to: usize, }, ShuffleQueue, RemoveQueueItem { track_id: TrackId, },}
/// Per-room event-loop actor. Owns all mutable state.struct RoomActor { rx: mpsc::Receiver<RoomCommand>, room_id: RoomId, room_name: String, publishers: RoomPublishers, store: Arc<dyn RoomStore>, cache_dir: PathBuf, /// Clone of the command sender, used by spawned download tasks to /// deliver DownloadReady / DownloadFailed back to the actor. cmd_tx: mpsc::Sender<RoomCommand>,
// Mutable state (owned by this actor, no locks needed) state: PlaybackState, client_count: u64, next_user_id: u64, last_active_tx: watch::Sender<DateTime<Utc>>, /// Users who voted to skip the current active track. Cleared /// whenever the active track changes. skip_votes: HashSet<u64>,
/// Running downloads keyed by URL (one task per URL — deduplicated). /// In-flight tracking only; completed results live in `state.media_cache`. download_tasks: HashMap<String, tokio::task::JoinHandle<()>>,}
/// Thread-safe registry of all active rooms with persistent store.#[derive(Clone)]pub(crate) struct Registry { store: Arc<dyn RoomStore>, cache_dir: PathBuf, inner: Arc<RwLock<Inner>>,}
struct Inner { rooms: HashMap<RoomId, RoomHandle>,}
impl Registry { pub(crate) fn new(store: Arc<dyn RoomStore>, cache_dir: PathBuf) -> Self { Self { store, inner: Arc::new(RwLock::new(Inner { rooms: HashMap::new(), })), cache_dir, } }
/// 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(); let _ = self.store.create_room(&id.as_str_buf(), &room_name).await;
let publishers = RoomPublishers::new(); let (tx, rx) = mpsc::channel(256); let (last_active_tx, last_active_rx) = watch::channel(now);
spawn_actor( rx, id, room_name.clone(), publishers.clone(), self.store.clone(), self.cache_dir.clone(), tx.clone(), PlaybackState::default(), last_active_tx, );
self.inner.write().await.rooms.insert( id, RoomHandle { cmd_tx: tx, publishers, last_active: last_active_rx, }, );
(id, room_name) }
/// Get the room handle. Rehydrates from store if the room exists but /// its actor was swept (idle timeout). pub(crate) async fn handle(&self, id: &RoomId) -> Option<RoomHandle> { // Fast path: room is already active if let Some(handle) = self.inner.read().await.rooms.get(id).cloned() { return Some(handle); }
// Slow path: try to rehydrate from store let snapshot = self.store.load(&id.as_str_buf()).await.ok()??;
let (state, effects) = PlaybackState::from_snapshot(snapshot.clone());
let publishers = RoomPublishers::new(); let (tx, rx) = mpsc::channel(256); let (last_active_tx, last_active_rx) = watch::channel(Utc::now());
spawn_actor( rx, *id, snapshot.room_name, publishers.clone(), self.store.clone(), self.cache_dir.clone(), tx.clone(), state, last_active_tx, );
// Persist any stale-active effects before returning the handle. if !effects.is_empty() { if let Err(e) = self.store.persist(&id.as_str_buf(), &effects).await { tracing::warn!(room = %id, error = %e, "rehydration: failed to persist stale effects"); } }
// Trigger initial state publish + started_at resolution. let _ = tx.send(RoomCommand::PublishState).await;
let handle = RoomHandle { cmd_tx: tx, publishers, last_active: last_active_rx, }; self.inner.write().await.rooms.insert(*id, handle.clone()); Some(handle) }
/// 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<RoomId> { if let Some(id) = RoomId::parse(r) { return Some(id); } let id_str = self.store.name_to_id(r).await.ok().flatten()?; RoomId::parse(&id_str) }
/// Load room name + history (for SSR rendering). /// Fetch recent chat messages for a room (newest first). pub(crate) async fn recent_chat( &self, id: &RoomId, limit: i64, ) -> Result<Vec<ChatMessage>, sqlx::Error> { self.store.recent_chat(&id.as_str_buf(), limit).await }
/// Push a raw URL onto a room's queue. Metadata extraction (yt-dlp) runs /// asynchronously in a spawned task. The caller returns immediately. pub(crate) async fn push_url(&self, id: &RoomId, url: String) { let handle = match self.handle(id).await { Some(h) => h, None => return, }; let _ = handle.cmd_tx.send(RoomCommand::QueueUrl(url)).await; }
/// Skip the current track. pub(crate) async fn skip(&self, room_id: &RoomId) { if let Some(handle) = self.handle(room_id).await { let _ = handle.cmd_tx.send(RoomCommand::Skip).await; } }
/// Sweep idle rooms that have been inactive longer than `timeout`. pub(crate) async fn sweep_idle(&self, timeout: Duration) -> Vec<RoomId> { let now = Utc::now(); let mut swept = Vec::new();
let idle_rooms: Vec<RoomId> = { let inner = self.inner.read().await; inner .rooms .iter() .filter(|(_, h)| { let idle = now - *h.last_active.borrow(); idle > chrono::TimeDelta::from_std(timeout).unwrap_or_default() }) .map(|(id, _)| *id) .collect() };
for id in &idle_rooms { if let Some(handle) = self.handle(id).await { if tokio::time::timeout( Duration::from_secs(1), // generous 1s window for graceful shutdown handle.cmd_tx.send(RoomCommand::Shutdown), ) .await .is_err() { tracing::warn!("sweep: room {id} channel full, forcing removal"); } } self.inner.write().await.rooms.remove(id); // Keep the room in the store so it can be rehydrated on rejoin. swept.push(*id); tracing::info!("swept idle room {id}"); }
swept }
pub(crate) async fn add_playlist_entry(&self, id: &RoomId, url: String) { if let Some(handle) = self.handle(id).await { let _ = handle .cmd_tx .send(RoomCommand::AddPlaylistEntry { url, added_by: None, }) .await; } }
pub(crate) async fn remove_playlist_entry(&self, id: &RoomId, entry_id: PlaylistEntryId) { if let Some(handle) = self.handle(id).await { let _ = handle .cmd_tx .send(RoomCommand::RemovePlaylistEntry { id: entry_id }) .await; } }}
/// Compute a filesystem-safe hash of a URL for use as a filename.fn file_hash(url: &str) -> String { let mut h = std::collections::hash_map::DefaultHasher::new(); url.hash(&mut h); format!("{:x}", h.finish())}
/// Construct a [`RoomActor`] and spawn it on the tokio runtime.////// Shared helper used by both [`Registry::create`] and [`Registry::handle`]/// to avoid duplicating the field initialization.#[allow(clippy::too_many_arguments)]fn spawn_actor( rx: mpsc::Receiver<RoomCommand>, room_id: RoomId, room_name: String, publishers: RoomPublishers, store: Arc<dyn RoomStore>, cache_dir: PathBuf, cmd_tx: mpsc::Sender<RoomCommand>, state: PlaybackState, last_active_tx: watch::Sender<DateTime<Utc>>,) { let actor = RoomActor { rx, room_id, room_name, publishers, store, cache_dir, cmd_tx, state, client_count: 0, next_user_id: 1, last_active_tx, skip_votes: HashSet::new(), download_tasks: HashMap::new(), }; tokio::spawn(actor.run());}
impl RoomActor { /// True if the URL has been resolved in the state machine's media cache. /// Failed URLs return false (they need a fresh download). fn is_resolved(&self, url: &str) -> bool { matches!( self.state.media_cache.get(url), Some(crate::state::MediaStatus::Resolved { .. }) ) }
/// Run the event loop. Returns when rx is closed or Shutdown received. async fn run(mut self) { while let Some(cmd) = self.rx.recv().await { self.last_active_tx.send_replace(Utc::now()); if self.process(cmd).await { break; } } // Abort any in-flight downloads. for (_, handle) in self.download_tasks.drain() { handle.abort(); } tracing::info!(room = %self.room_id, "room actor shut down"); }
/// Process a single command. /// /// Returns `true` if the actor should shut down. #[must_use] async fn process(&mut self, cmd: RoomCommand) -> bool { self.process_cmd(cmd).await }
async fn process_cmd(&mut self, cmd: RoomCommand) -> bool { match cmd { RoomCommand::QueueUrl(url) => self.handle_queue_url(url).await, RoomCommand::DownloadReady { url, media } => { self.handle_download_ready(url, media).await } RoomCommand::DownloadFailed { url, error } => { self.handle_download_failed(url, error).await } RoomCommand::Skip => self.handle_skip().await, RoomCommand::TrackEnded { item_id } => self.handle_track_ended(item_id).await, RoomCommand::Chat { user_name, content, msg_type, created_at, } => { self.handle_chat(user_name, content, msg_type, created_at) .await } RoomCommand::RegisterClient { resp } => self.handle_register_client(resp).await, RoomCommand::UnregisterClient => self.handle_unregister_client().await, RoomCommand::AddPlaylistEntry { url, added_by } => { self.handle_add_playlist_entry(url, added_by).await } RoomCommand::RemovePlaylistEntry { id } => self.handle_remove_playlist_entry(id).await, RoomCommand::SetAutoplay { enabled } => self.handle_set_autoplay(enabled).await, 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, RoomCommand::ShuffleQueue => self.handle_shuffle_queue().await, RoomCommand::RemoveQueueItem { track_id } => { self.handle_remove_queue_item(track_id).await } } }
async fn persist_effects(&self, effects: &[Effect]) { if let Err(e) = self .store .persist(&self.room_id.as_str_buf(), effects) .await { tracing::warn!(room = %self.room_id, error = %e, "persist failed"); } }
/// Queue a URL. If the room is idle the track advances to active /// immediately. When the download is already cached we resolve /// `started_at` before publishing — single snapshot, no flicker. async fn handle_queue_url(&mut self, url: String) -> bool { tracing::debug!(room = %self.room_id, %url, "track url queued");
let item_id = TrackId::new(); let now = Utc::now().timestamp_millis();
// Gather: check whether the state machine has cached metadata. // Cache hits override the placeholder so the track goes out // fully-formed; the state machine still re-checks via fill_from_cache. let cached = self.state.media_cache.get(&url).and_then(|s| match s { crate::state::MediaStatus::Resolved { title, duration, thumbnail, source: _, } => Some((title.clone(), duration.clone(), thumbnail.clone())), crate::state::MediaStatus::Failed(_) => None, }); let (title, duration, thumbnail, pending) = match cached { Some((t, d, th)) => (t, d, th, false), None => ("Loading...".into(), "--:--".into(), None, true), };
// Pure transition. let track = QueuedTrack { id: item_id, title, url: url.clone(), duration, thumbnail, pending, }; let effects = self.state.transition(&Event::TrackQueued(track), now); self.persist_effects(&effects).await;
// If the track became active and the file is cached, resolve BEFORE // publishing — the snapshot goes out with a real started_at. let became_active = effects .iter() .any(|e| matches!(e, Effect::PersistStarted(_))); if became_active && !pending { self.state.resolve_started_at(now); }
self.execute_effects(effects).await; self.ensure_downloaded(&url).await; self.try_start_active().await; false }
/// Download completed. Update the state machine's media cache and propagate /// resolved metadata to all tracks referencing this URL. Persist the /// metadata to media_cache and broadcast. async fn handle_download_ready( &mut self, url: String, media: crate::ingest::IngestedTrack, ) -> bool { tracing::debug!(room = %self.room_id, %url, title = %media.title, "download ready");
let title = media.title; let duration = format_duration(media.duration); let thumbnail = media.thumbnail; let source = media.source; let now = Utc::now().timestamp_millis();
// Update the state machine's media cache AND all tracks/playlist // entries sharing this URL via a single pure transition. let effects = self.state.transition( &Event::DownloadResolved { url: url.clone(), title: title.clone(), duration: duration.clone(), thumbnail: thumbnail.clone(), source: source.clone(), }, now, ); self.persist_effects(&effects).await;
// Check whether the active track was waiting for this file. let needs_start = self .state .active .as_ref() .is_some_and(|a| a.url == url && a.started_at_wall == 0);
self.execute_effects(effects).await;
if needs_start { self.state.resolve_started_at(now); self.publish_state_snapshot().await; // second publish, now with started_at self.ensure_next_downloaded().await; }
self.maybe_auto_fill().await; false }
/// Common post-transition cleanup: resolve the active track's started_at /// if the file is cached, start any needed downloads, and auto-fill from /// the playlist if the room became idle. /// /// All operations are idempotent — safe to call after any state change. async fn post_advance(&mut self) -> bool { self.try_start_active().await; self.ensure_active_download().await; self.ensure_next_downloaded().await; let _ = self.maybe_auto_fill().await; false }
async fn handle_download_failed(&mut self, url: String, error: String) -> bool { tracing::debug!(room = %self.room_id, %url, %error, "download failed");
// Remove the failed task so a retry can start a fresh download. self.download_tasks.remove(&url);
let now = Utc::now().timestamp_millis();
// Mark the URL as Failed in the state machine's media cache. This // stops ensure_downloaded from retrying and lets TrackQueued keep a // placeholder for a URL we know is bad. let mut effects = Vec::new(); effects.extend(self.state.transition( &Event::DownloadFailed { url: url.clone(), error: error.clone(), }, now, ));
// Only affect tracks whose URL matches the failed download. let affected: Vec<TrackId> = self .state .queue .iter() .filter(|t| t.url == url) .map(|t| t.id) .chain( self.state .active .iter() .filter(|a| a.url == url) .map(|a| a.id), ) .collect();
// If no tracks reference this URL, the cache update is enough. if affected.is_empty() { return false; }
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 { item_id: *id, url: url.clone(), title: failed_title.clone(), duration: "--:--".into(), thumbnail: None, source: "ytdlp".into(), }, now, )); effects.extend( self.state .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; self.post_advance().await }
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; self.execute_effects(effects).await; self.post_advance().await }
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; self.execute_effects(effects).await; 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<String>) -> bool { let id = PlaylistEntryId::new(); let entry = PlaylistEntry { id, url: url.clone(), title: "Loading...".into(), duration: "--:--".into(), thumbnail: None, source: "direct".into(), added_by, added_at: crate::util::now_iso(), pending: true, failed: None, }; // Send the new entry through the state machine. The state emits // Effect::AddPlaylistEntry (for persistence) and Effect::PublishSnapshot // (for clients), so the actor doesn't have to track the playlist itself. let effects = self.state.transition(&Event::PlaylistEntryAdded(entry), 0); if let Err(e) = self .store .persist(&self.room_id.as_str_buf(), &effects) .await { tracing::warn!("failed to persist playlist entry: {e}"); return false; } self.execute_effects(effects).await; self.ensure_downloaded(&url).await; false }
async fn handle_remove_playlist_entry(&mut self, id: PlaylistEntryId) -> bool { let effects = self.state.transition(&Event::PlaylistEntryRemoved(id), 0); if effects.is_empty() { return false; } if let Err(e) = self .store .persist(&self.room_id.as_str_buf(), &effects) .await { tracing::warn!("failed to persist playlist removal: {e}"); return false; } self.execute_effects(effects).await; false }
/// Toggle the room's auto-play setting. The state machine decides /// whether the change is observable (no-op if same value, or if /// re-enabling an empty playlist). async fn handle_set_autoplay(&mut self, enabled: bool) -> bool { let effects = self.state.transition(&Event::SetAutoplay { enabled }, 0); if effects.is_empty() { return false; } self.persist_effects(&effects).await; self.execute_effects(effects).await; false }
/// Set the skip-vote threshold. Persists and publishes when the /// value actually changes. async fn handle_set_skip_threshold(&mut self, percent: u8) -> bool { let effects = self .state .transition(&Event::SetSkipThreshold { percent }, 0); if effects.is_empty() { return false; } self.persist_effects(&effects).await; self.execute_effects(effects).await; false }
/// Reorder a single track in the queue. Out-of-range or no-op moves /// return no effects from the state machine; everything is persisted /// before the snapshot goes out. async fn handle_move_queue_item(&mut self, from: usize, to: usize) -> bool { let effects = self.state.transition(&Event::MoveQueueItem { from, to }, 0); if effects.is_empty() { return false; } self.persist_effects(&effects).await; self.execute_effects(effects).await; false }
/// Shuffle the queue using a server-picked seed. The seed comes from the /// wall clock so concurrent shuffles are independent, and clients can't /// predict or replay each other's shuffles. async fn handle_shuffle_queue(&mut self) -> bool { let seed = Utc::now().timestamp_nanos_opt().unwrap_or(0) as u64; let effects = self.state.transition(&Event::ShuffleQueue { seed }, 0); if effects.is_empty() { return false; } self.persist_effects(&effects).await; self.execute_effects(effects).await; false }
/// Remove a queued track. The state machine no-ops if the track is the /// active one (you can't delete a playing track) or if it isn't in /// the queue at all. async fn handle_remove_queue_item(&mut self, track_id: TrackId) -> bool { let effects = self .state .transition(&Event::RemoveQueueItem { track_id }, 0); if effects.is_empty() { return false; } self.persist_effects(&effects).await; self.execute_effects(effects).await; false }
async fn handle_chat( &mut self, user_name: String, content: String, msg_type: String, created_at: String, ) -> bool { let effects = vec![Effect::PersistChat { user_name: user_name.clone(), content: content.clone(), msg_type: msg_type.clone(), created_at: created_at.clone(), }]; if let Ok(output) = self .store .persist(&self.room_id.as_str_buf(), &effects) .await { if let Some(msg_id) = output.chat_message_id { let msg = ChatMessage { id: msg_id, user_name, content, msg_type, created_at, }; let _ = self.publishers.chat_tx.send(msg); } } false }
async fn handle_register_client(&mut self, resp: tokio::sync::oneshot::Sender<u64>) -> bool { let id = self.next_user_id; 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 }
async fn handle_unregister_client(&mut self) -> bool { self.client_count = self.client_count.saturating_sub(1); self.publish_state_snapshot().await; false }
async fn handle_publish_state(&mut self) -> bool { self.resolve_active_if_ready().await; self.publish_state_snapshot().await; false }
async fn handle_shutdown(&mut self) -> bool { tracing::info!(room = %self.room_id, "actor shutdown requested"); for (_, handle) in self.download_tasks.drain() { handle.abort(); } true }
/// Execute actor-side effects from a state transition. /// /// Persistence effects are handled by [`RoomStore::persist`] before this /// is called — only abort and publishing effects remain. async fn execute_effects(&mut self, effects: Vec<Effect>) { for effect in effects { match effect { Effect::AbortDownload => { // Abort all running downloads (only one should be active). for (_, handle) in self.download_tasks.drain() { handle.abort(); } } Effect::PublishSnapshot => { self.publish_state_snapshot().await; } // All persistence effects handled by store.persist() in process(). _ => {} } } }
/// If the room is idle and has playlist entries, auto-fill the next track. /// Returns `true` if a track was added (so callers can chain post-advance /// work). async fn maybe_auto_fill(&mut self) -> bool { // Auto-fill only applies when the room is idle. If something is // playing or queued, leave the playlist alone. if self.state.active.is_some() || !self.state.queue.is_empty() { return false; } // The state machine owns the playlist and cursor; it advances the cursor // as a side effect of returning Some. We capture the entry, build a // track, and dispatch it through TrackQueuedFromPlaylist. let entry = match self.state.next_playlist_entry() { Some(e) => e, None => return false, };
let id = TrackId::new(); let now = Utc::now().timestamp_millis(); let url = entry.url.clone(); let track = QueuedTrack { id, title: entry.title, url, duration: entry.duration, thumbnail: entry.thumbnail, pending: false, }; let effects = self .state .transition(&Event::TrackQueuedFromPlaylist(track, 0), now); if let Err(e) = self .store .persist(&self.room_id.as_str_buf(), &effects) .await { tracing::warn!("auto-fill: failed to persist: {e}"); } self.execute_effects(effects).await;
// The newly advanced track starts with started_at: 0. If the file is // already cached, resolve it immediately so the frontend loads. self.try_start_active().await; self.ensure_active_download().await; true }
/// Common post-transition cleanup: resolve the active track's started_at /// If the active track has a placeholder `started_at_wall` (0), resolve it /// to now. Used after rehydration where the file was already downloaded. async fn resolve_active_if_ready(&mut self) { if let Some(active) = &self.state.active { if active.started_at_wall == 0 && self.is_resolved(&active.url) { self.state.resolve_started_at(Utc::now().timestamp_millis()); } } }
/// Publish the current state snapshot to all connected clients. /// /// Reads from in-memory [`PlaybackState`], no SQLite queries. async fn publish_state_snapshot(&self) { let current = self.state.active.as_ref().map(|t| TrackState { id: t.id, hash: file_hash(&t.url), title: t.title.clone(), url: t.url.clone(), duration: t.duration.clone(), thumbnail: t.thumbnail.clone(), started_at: t.started_at_wall, });
let queue: Vec<QueueSummary> = self .state .queue .iter() .map(|q| QueueSummary { id: q.id, hash: file_hash(&q.url), title: q.title.clone(), url: q.url.clone(), duration: q.duration.clone(), thumbnail: q.thumbnail.clone(), pending: q.pending, }) .collect();
let history: Vec<HistoryEntry> = self .state .history .iter() .map(|h| HistoryEntry { id: h.id, hash: file_hash(&h.url), title: h.title.clone(), url: h.url.clone(), duration: h.duration.clone(), thumbnail: h.thumbnail.clone(), played_at: h.played_at.clone(), }) .collect();
let client_count = self.client_count;
tracing::trace!( room = %self.room_id, queue_len = %queue.len(), history_len = %history.len(), clients = %client_count, "publish_state_snapshot", );
let snapshot = serde_json::json!({ "room_name": self.room_name, "current_track": current, "queue": queue, "history": history, "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); }
/// Ensure a URL has been downloaded (or is in progress). Deduplicated by /// URL: if the URL is already downloaded or downloading, this is a no-op. /// Safe to call from both queue and playlist paths. async fn ensure_downloaded(&mut self, url: &str) { let url = url.to_string(); if self.is_resolved(&url) || self.download_tasks.contains_key(&url) { return; } // Hash the URL for a stable file path, avoiding filename conflicts. let hash = file_hash(&url); let dest = self.cache_dir.join(&hash); let cmd_tx = self.cmd_tx.clone(); let url_clone = url.clone(); let handle = tokio::spawn(async move { match crate::ingest::download(&url_clone, &dest).await { Ok(media) => { let _ = cmd_tx .send(RoomCommand::DownloadReady { url: url_clone, media, }) .await; } Err(e) => { let _ = cmd_tx .send(RoomCommand::DownloadFailed { url: url_clone, error: e.to_string(), }) .await; } } }); self.download_tasks.insert(url, handle); }
/// If the active track has a completed download cached, resolve its /// started_at and publish the updated snapshot. Called after a track /// advances (skip or track_ended). async fn try_start_active(&mut self) { let should_start = self .state .active .as_ref() .map(|t| t.started_at_wall == 0 && self.is_resolved(&t.url)) .unwrap_or(false); if should_start { let now = chrono::Utc::now().timestamp_millis(); self.state.resolve_started_at(now); self.publish_state_snapshot().await; self.ensure_next_downloaded().await; } }
/// If there's a next queued track that isn't already downloaded or /// downloading, start its download pre-emptively. Called after a track /// starts playing so the next track is ready when this one ends. async fn ensure_next_downloaded(&mut self) { let needs_download = self .state .queue .first() .map(|next| { !self.is_resolved(&next.url) && !self.download_tasks.contains_key(&next.url) }) .unwrap_or(false); if needs_download { let url = self .state .queue .first() .map(|t| t.url.clone()) .expect("queue must be non-empty after needs_download check"); self.ensure_downloaded(&url).await; } }
/// If the active track hasn't been downloaded yet, start its download. /// Called after Skip / TrackEnded when the next queue item becomes active. async fn ensure_active_download(&mut self) { let needs_download = self .state .active .as_ref() .map(|t| { t.started_at_wall == 0 && !self.is_resolved(&t.url) && !self.download_tasks.contains_key(&t.url) }) .unwrap_or(false); if needs_download { let url = self .state .active .as_ref() .map(|t| t.url.clone()) .expect("active must be Some after needs_download check"); self.ensure_downloaded(&url).await; } }}
/// Format a duration in seconds as "M:SS" or "H:MM:SS".fn format_duration(secs: f64) -> String { let total = secs as u64; let hours = total / 3600; let minutes = (total % 3600) / 60; let seconds = total % 60; if hours > 0 { format!("{hours}:{minutes:02}:{seconds:02}") } else { format!("{minutes}:{seconds:02}") }}
/// 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::*; use crate::types::ActiveTrackInfo;
fn tid(n: u128) -> TrackId { TrackId(uuid::Uuid::from_u128(n)) }
fn pid(n: u128) -> PlaylistEntryId { PlaylistEntryId(uuid::Uuid::from_u128(n)) }
#[test] fn test_auto_fill_skipped_when_active_track_present() { let state = PlaybackState { active: Some(ActiveTrackInfo { id: tid(1), title: "Active".into(), url: "https://example.com".into(), duration: "3:00".into(), thumbnail: None, started_at_wall: 0, }), ..PlaybackState::default() }; // The actor's maybe_auto_fill guard: don't draw from the playlist // while a track is playing. The state machine has no such guard — // next_playlist_entry can be called any time. The actor enforces. let pending = !state.active.is_some() && state.queue.is_empty(); assert!(!pending); }
#[test] fn test_auto_fill_skipped_when_queue_not_empty() { let mut state = PlaybackState::default(); state.queue.push(QueuedTrack { id: tid(1), title: "Queued".into(), url: "https://example.com".into(), duration: "3:00".into(), thumbnail: None, pending: false, }); let pending = state.active.is_some() || !state.queue.is_empty(); assert!(pending); }
#[test] fn test_next_playlist_entry_disabled_returns_none() { let mut state = PlaybackState { autoplay_enabled: false, playlist: vec![PlaylistEntry { id: pid(1), url: "https://example.com/p".into(), title: "P".into(), duration: "3:00".into(), thumbnail: None, source: "direct".into(), added_by: None, added_at: "now".into(), pending: false, failed: None, }], ..PlaybackState::default() }; // When autoplay is explicitly disabled, no entries drawn. assert!(state.next_playlist_entry().is_none()); }
#[test] fn test_next_playlist_entry_returns_and_advances_when_idle_and_enabled() { let mut state = PlaybackState { autoplay_enabled: true, playlist: vec![PlaylistEntry { id: pid(1), url: "https://example.com/p".into(), title: "P".into(), duration: "3:00".into(), thumbnail: None, source: "direct".into(), added_by: None, added_at: "now".into(), pending: false, failed: None, }], ..PlaybackState::default() }; let entry = state.next_playlist_entry().unwrap(); assert_eq!(entry.id, pid(1)); // Single-entry playlist: cursor wraps to 0. assert_eq!(state.playlist_cursor, 0); }
#[test] fn test_next_playlist_entry_wraps_around() { let mut state = PlaybackState { autoplay_enabled: true, playlist: vec![ PlaylistEntry { id: pid(1), url: "https://example.com/a".into(), title: "A".into(), duration: "3:00".into(), thumbnail: None, source: "direct".into(), added_by: None, added_at: "now".into(), pending: false, failed: None, }, PlaylistEntry { id: pid(2), url: "https://example.com/b".into(), title: "B".into(), duration: "4:00".into(), thumbnail: None, source: "direct".into(), added_by: None, added_at: "now".into(), pending: false, failed: None, }, ], ..PlaybackState::default() }; // Cursor at 1: draws second entry, advances to 2 → wraps to 0. state.playlist_cursor = 1; let entry = state.next_playlist_entry().unwrap(); assert_eq!(entry.id, pid(2)); assert_eq!(state.playlist_cursor, 0); // Cursor at 0: draws first entry. 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)); }}