···11+# Seeds for failure cases proptest has generated in the past. It is
22+# automatically read and these particular cases re-run before any
33+# novel cases are generated.
44+#
55+# It is recommended to check this file in to source control so that
66+# everyone who runs the test benefits from these saved cases.
77+cc e68e654f36d5af1cc26441258aebd42f4c2b2182b17658f78de107b2cbf070b9 # shrinks to ops = [Queue(0), Queue(1)]
+202-262
src/room.rs
···1111//! 3. Commit: execute returned [`Effect`]s
12121313use std::collections::HashMap;
1414+use std::hash::{Hash, Hasher};
1415use std::path::PathBuf;
1516use std::sync::Arc;
1617use std::time::Duration;
···2627 TrackId, TrackState,
2728};
28292929-#[cfg(test)]
3030-use crate::types::ChatMessageId;
3131-#[cfg(test)]
3232-use tokio::sync::oneshot;
3333-3430/// All room operations, processed sequentially by the per-room actor.
3531pub(crate) enum RoomCommand {
3632 QueueUrl(String),
3733 DownloadReady {
3838- item_id: TrackId,
3434+ url: String,
3935 media: crate::ingest::IngestedTrack,
4036 },
4137 DownloadFailed {
4242- item_id: TrackId,
3838+ url: String,
4339 error: String,
4440 },
4541 Skip,
···6359 RemovePlaylistEntry {
6460 id: PlaylistEntryId,
6561 },
6666- PlaylistDownloadReady {
6767- id: PlaylistEntryId,
6868- media: crate::ingest::IngestedTrack,
6969- },
7070- PlaylistDownloadFailed {
7171- id: PlaylistEntryId,
7272- error: String,
7373- },
7462 PublishState,
7563 Shutdown,
7664}
···9886 /// Index into the playlist for round-robin auto-fill.
9987 playlist_index: usize,
10088101101- /// Track handles of running downloads, keyed by track ID.
102102- download_tasks: HashMap<TrackId, tokio::task::JoinHandle<()>>,
103103- /// Completed downloads whose tracks haven't advanced to active yet.
104104- download_results: HashMap<TrackId, crate::ingest::IngestedTrack>,
8989+ /// Running downloads keyed by URL (one task per URL — deduplicated).
9090+ download_tasks: HashMap<String, tokio::task::JoinHandle<()>>,
9191+ /// Completed downloads keyed by URL. Shared across all tracks/entries
9292+ /// referencing the same media URL.
9393+ download_results: HashMap<String, crate::ingest::IngestedTrack>,
10594}
1069510796/// Thread-safe registry of all active rooms with persistent store.
···246235 let _ = handle.cmd_tx.send(RoomCommand::QueueUrl(url)).await;
247236 }
248237249249- /// Remove a room (from store and in-memory state).
250250- #[cfg(test)]
251251- pub(crate) async fn remove(&self, id: &RoomId) {
252252- if let Some(handle) = self.handle(id).await {
253253- let _ = tokio::time::timeout(
254254- Duration::from_secs(1), // generous 1s window for graceful shutdown
255255- handle.cmd_tx.send(RoomCommand::Shutdown),
256256- )
257257- .await;
258258- }
259259- self.inner.write().await.rooms.remove(id);
260260- let _ = self.store.delete_room(&id.as_str_buf()).await;
261261- }
262262-263263- /// Send a chat message: validate, send to actor, return best-effort ChatMessage.
264264- #[cfg(test)]
265265- pub(crate) async fn send_chat(
266266- &self,
267267- room_id: &RoomId,
268268- user_name: &str,
269269- content: &str,
270270- msg_type: &str,
271271- ) -> Result<ChatMessage, anyhow::Error> {
272272- if content.is_empty() {
273273- anyhow::bail!("content is empty");
274274- }
275275- if content.len() > 2000 {
276276- anyhow::bail!("content exceeds 2000 characters");
277277- }
278278- if user_name.is_empty() {
279279- anyhow::bail!("username is empty");
280280- }
281281- if user_name.len() > 32 {
282282- anyhow::bail!("username exceeds 32 characters");
283283- }
284284- if let Some(handle) = self.handle(room_id).await {
285285- let _ = handle
286286- .cmd_tx
287287- .send(RoomCommand::Chat {
288288- user_name: user_name.to_string(),
289289- content: content.to_string(),
290290- msg_type: msg_type.to_string(),
291291- created_at: chrono::Utc::now()
292292- .format("%Y-%m-%dT%H:%M:%S%.3fZ")
293293- .to_string(),
294294- })
295295- .await;
296296- }
297297- Ok(ChatMessage {
298298- id: ChatMessageId::new(),
299299- user_name: user_name.to_string(),
300300- content: content.to_string(),
301301- msg_type: msg_type.to_string(),
302302- created_at: chrono::Utc::now()
303303- .format("%Y-%m-%dT%H:%M:%S%.3fZ")
304304- .to_string(),
305305- })
306306- }
307307-308308- /// Get the most recent chat messages from the store.
309309- /// Register a client in the room.
310310- #[cfg(test)]
311311- pub(crate) async fn register_client(&self, room_id: &RoomId) -> Option<u64> {
312312- let handle = self.handle(room_id).await?;
313313- let (resp_tx, resp_rx) = oneshot::channel();
314314- handle
315315- .cmd_tx
316316- .send(RoomCommand::RegisterClient { resp: resp_tx })
317317- .await
318318- .ok()?;
319319- resp_rx.await.ok()
320320- }
321321-322238 /// Skip the current track.
323239 pub(crate) async fn skip(&self, room_id: &RoomId) {
324240 if let Some(handle) = self.handle(room_id).await {
···411327 Some((entry, index.wrapping_add(1)))
412328}
413329330330+/// Compute a filesystem-safe hash of a URL for use as a filename.
331331+fn file_hash(url: &str) -> String {
332332+ let mut h = std::collections::hash_map::DefaultHasher::new();
333333+ url.hash(&mut h);
334334+ format!("{:x}", h.finish())
335335+}
336336+414337/// Construct a [`RoomActor`] and spawn it on the tokio runtime.
415338///
416339/// Shared helper used by both [`Registry::create`] and [`Registry::handle`]
···475398 async fn process_cmd(&mut self, cmd: RoomCommand) -> bool {
476399 match cmd {
477400 RoomCommand::QueueUrl(url) => self.handle_queue_url(url).await,
478478- RoomCommand::DownloadReady { item_id, media } => {
479479- self.handle_download_ready(item_id, media).await
480480- }
481481- RoomCommand::DownloadFailed { item_id, error } => {
482482- self.handle_download_failed(item_id, error).await
483483- }
484484- RoomCommand::Skip => {
485485- let result = self.handle_skip().await;
486486- if !result {
487487- self.maybe_auto_fill().await;
488488- }
489489- result
401401+ RoomCommand::DownloadReady { url, media } => {
402402+ self.handle_download_ready(url, media).await
490403 }
491491- RoomCommand::TrackEnded { item_id } => {
492492- let result = self.handle_track_ended(item_id).await;
493493- if !result {
494494- self.maybe_auto_fill().await;
495495- }
496496- result
404404+ RoomCommand::DownloadFailed { url, error } => {
405405+ self.handle_download_failed(url, error).await
497406 }
407407+ RoomCommand::Skip => self.handle_skip().await,
408408+ RoomCommand::TrackEnded { item_id } => self.handle_track_ended(item_id).await,
498409 RoomCommand::Chat {
499410 user_name,
500411 content,
···510421 self.handle_add_playlist_entry(url, added_by).await
511422 }
512423 RoomCommand::RemovePlaylistEntry { id } => self.handle_remove_playlist_entry(id).await,
513513- RoomCommand::PlaylistDownloadReady { id, media } => {
514514- self.handle_playlist_download_ready(id, media).await
515515- }
516516- RoomCommand::PlaylistDownloadFailed { id, error } => {
517517- self.handle_playlist_download_failed(id, error).await
518518- }
519424 RoomCommand::PublishState => self.handle_publish_state().await,
520425 RoomCommand::Shutdown => self.handle_shutdown().await,
521426 }
···531436 }
532437 }
533438439439+ /// Queue a URL. If the room is idle the track advances to active
440440+ /// immediately. When the download is already cached we resolve
441441+ /// `started_at` before publishing — single snapshot, no flicker.
534442 async fn handle_queue_url(&mut self, url: String) -> bool {
535443 tracing::debug!(room = %self.room_id, %url, "track url queued");
536444537445 let item_id = TrackId::new();
538446 let now = Utc::now().timestamp_millis();
539447448448+ // Gather: check whether we already have the file.
449449+ let cached = self.download_results.get(&url);
450450+ let (title, duration, thumbnail, pending) = if let Some(c) = cached {
451451+ (
452452+ c.title.clone(),
453453+ format_duration(c.duration),
454454+ c.thumbnail.clone(),
455455+ false,
456456+ )
457457+ } else {
458458+ ("Loading...".into(), "--:--".into(), None, true)
459459+ };
460460+461461+ // Pure transition.
540462 let track = QueuedTrack {
541463 id: item_id,
542542- title: "Loading...".into(),
464464+ title,
543465 url: url.clone(),
544544- duration: "--:--".into(),
545545- thumbnail: None,
546546- pending: true,
466466+ duration,
467467+ thumbnail,
468468+ pending,
547469 };
548548-549470 let effects = self.state.transition(&Event::TrackQueued(track), now);
550471 self.persist_effects(&effects).await;
472472+473473+ // If the track became active and the file is cached, resolve BEFORE
474474+ // publishing — the snapshot goes out with a real started_at.
475475+ let became_active = effects
476476+ .iter()
477477+ .any(|e| matches!(e, Effect::PersistStarted(_)));
478478+ if became_active && !pending {
479479+ self.state.resolve_started_at(now);
480480+ }
481481+551482 self.execute_effects(effects).await;
552552- self.start_download(item_id, &url).await;
483483+ self.ensure_downloaded(&url).await;
484484+ self.try_start_active().await;
553485 false
554486 }
555487488488+ /// Download completed. Cache the result, update all in-memory state
489489+ /// (tracks + playlist entries referencing this URL), persist metadata
490490+ /// to media_cache, and broadcast.
556491 async fn handle_download_ready(
557492 &mut self,
558558- item_id: TrackId,
493493+ url: String,
559494 media: crate::ingest::IngestedTrack,
560495 ) -> bool {
561561- tracing::debug!(room = %self.room_id, %item_id, title = %media.title, "download ready");
496496+ tracing::debug!(room = %self.room_id, %url, title = %media.title, "download ready");
562497563563- self.download_results.insert(item_id, media.clone());
498498+ self.download_results.insert(url.clone(), media.clone());
499499+ let title = media.title;
500500+ let duration = format_duration(media.duration);
501501+ let thumbnail = media.thumbnail;
502502+ let source = media.source;
503503+ let now = Utc::now().timestamp_millis();
564504565565- let now = Utc::now().timestamp_millis();
505505+ // Update all tracks via the pure state machine (URL-keyed event).
566506 let effects = self.state.transition(
567567- &Event::MetadataUpdated {
568568- item_id,
569569- title: media.title,
570570- duration: format_duration(media.duration),
571571- thumbnail: media.thumbnail,
572572- source: media.source,
507507+ &Event::MetadataResolved {
508508+ url: url.clone(),
509509+ title: title.clone(),
510510+ duration: duration.clone(),
511511+ thumbnail: thumbnail.clone(),
512512+ source: source.clone(),
573513 },
574514 now,
575515 );
516516+ self.persist_effects(&effects).await;
576517577577- self.persist_effects(&effects).await;
578578- self.execute_effects(effects).await;
518518+ // Update playlist entries in memory (they aren't owned by the state
519519+ // machine, so we update them directly).
520520+ for entry in &mut self.playlist {
521521+ if entry.url == url {
522522+ entry.title = title.clone();
523523+ entry.duration = duration.clone();
524524+ entry.thumbnail = thumbnail.clone();
525525+ entry.source = source.clone();
526526+ entry.pending = false;
527527+ }
528528+ }
579529580580- let is_active_and_pending = self
530530+ // Check whether the active track was waiting for this file.
531531+ let needs_start = self
581532 .state
582533 .active
583534 .as_ref()
584584- .map(|t| t.id == item_id && t.started_at_wall == 0)
585585- .unwrap_or(false);
535535+ .is_some_and(|a| a.url == url && a.started_at_wall == 0);
586536587587- if is_active_and_pending {
588588- let now = Utc::now().timestamp_millis();
537537+ self.execute_effects(effects).await;
538538+539539+ if needs_start {
589540 self.state.resolve_started_at(now);
590590- self.publish_state_snapshot().await;
541541+ self.publish_state_snapshot().await; // second publish, now with started_at
591542 self.ensure_next_downloaded().await;
592543 }
593544545545+ self.maybe_auto_fill().await;
594546 false
595547 }
596548597597- async fn handle_download_failed(&mut self, item_id: TrackId, error: String) -> bool {
598598- tracing::debug!(room = %self.room_id, %item_id, %error, "download failed");
549549+ /// Common post-transition cleanup: resolve the active track's started_at
550550+ /// if the file is cached, start any needed downloads, and auto-fill from
551551+ /// the playlist if the room became idle.
552552+ ///
553553+ /// All operations are idempotent — safe to call after any state change.
554554+ async fn post_advance(&mut self) -> bool {
555555+ self.try_start_active().await;
556556+ self.ensure_active_download().await;
557557+ self.ensure_next_downloaded().await;
558558+ let _ = self.maybe_auto_fill().await;
559559+ false
560560+ }
561561+562562+ async fn handle_download_failed(&mut self, url: String, error: String) -> bool {
563563+ tracing::debug!(room = %self.room_id, %url, %error, "download failed");
564564+565565+ // Remove the failed task so a retry can start a fresh download.
566566+ self.download_tasks.remove(&url);
599567600568 let now = Utc::now().timestamp_millis();
601601- let effects = self.state.transition(
602602- &Event::MetadataUpdated {
603603- item_id,
604604- title: format!("Failed to load: {error}"),
605605- duration: "--:--".into(),
606606- thumbnail: None,
607607- source: "ytdlp".into(),
608608- },
609609- now,
610610- );
569569+570570+ // Only affect tracks whose URL matches the failed download.
571571+ let affected: Vec<TrackId> = self
572572+ .state
573573+ .queue
574574+ .iter()
575575+ .filter(|t| t.url == url)
576576+ .map(|t| t.id)
577577+ .chain(
578578+ self.state
579579+ .active
580580+ .iter()
581581+ .filter(|a| a.url == url)
582582+ .map(|a| a.id),
583583+ )
584584+ .collect();
585585+586586+ // If no tracks reference this URL, nothing to do.
587587+ if affected.is_empty() {
588588+ return false;
589589+ }
590590+591591+ let mut effects = Vec::new();
592592+ let failed_title = format!("Failed to load: {error}");
593593+ for id in &affected {
594594+ effects.extend(self.state.transition(
595595+ &Event::MetadataUpdated {
596596+ item_id: *id,
597597+ url: url.clone(),
598598+ title: failed_title.clone(),
599599+ duration: "--:--".into(),
600600+ thumbnail: None,
601601+ source: "ytdlp".into(),
602602+ },
603603+ now,
604604+ ));
605605+ effects.extend(
606606+ self.state
607607+ .transition(&Event::TrackFailed { item_id: *id }, now),
608608+ );
609609+ }
611610612611 self.persist_effects(&effects).await;
613612 self.execute_effects(effects).await;
614614- false
613613+ self.post_advance().await
615614 }
616615617616 async fn handle_skip(&mut self) -> bool {
···619618620619 let now = Utc::now().timestamp_millis();
621620 let effects = self.state.transition(&Event::Skip, now);
622622-623621 self.persist_effects(&effects).await;
624622 self.execute_effects(effects).await;
625625- self.try_start_active().await;
626626- self.ensure_active_download().await;
627627- false
623623+ self.post_advance().await
628624 }
629625630626 async fn handle_track_ended(&mut self, item_id: TrackId) -> bool {
···632628633629 let now = Utc::now().timestamp_millis();
634630 let effects = self.state.transition(&Event::TrackEnded { item_id }, now);
635635-636631 self.persist_effects(&effects).await;
637632 self.execute_effects(effects).await;
638638- self.try_start_active().await;
639639- self.ensure_active_download().await;
640640- false
633633+ self.post_advance().await
641634 }
642635643636 async fn handle_add_playlist_entry(&mut self, url: String, added_by: Option<String>) -> bool {
···652645 added_by,
653646 added_at: crate::util::now_iso(),
654647 pending: true,
655655- stream_url: None,
656648 };
657649 let effects = vec![Effect::AddPlaylistEntry {
658650 id,
···672664 }
673665 self.playlist.push(entry);
674666 self.publish_state_snapshot().await;
675675- self.start_playlist_download(id, &url).await;
667667+ self.ensure_downloaded(&url).await;
676668 false
677669 }
678670···691683 false
692684 }
693685694694- async fn handle_playlist_download_ready(
695695- &mut self,
696696- id: PlaylistEntryId,
697697- media: crate::ingest::IngestedTrack,
698698- ) -> bool {
699699- tracing::debug!(room = %self.room_id, %id, title = %media.title, "playlist download ready");
700700-701701- if let Some(entry) = self.playlist.iter_mut().find(|e| e.id == id) {
702702- entry.title = media.title.clone();
703703- entry.duration = format_duration(media.duration);
704704- entry.thumbnail = media.thumbnail.clone();
705705- entry.source = media.source.clone();
706706- entry.pending = false;
707707- }
708708- let effects = vec![Effect::UpdatePlaylistEntry {
709709- id,
710710- title: media.title,
711711- duration: format_duration(media.duration),
712712- thumbnail: media.thumbnail,
713713- source: media.source,
714714- }];
715715- self.persist_effects(&effects).await;
716716- self.publish_state_snapshot().await;
717717- self.maybe_auto_fill().await;
718718- false
719719- }
720720-721721- async fn handle_playlist_download_failed(
722722- &mut self,
723723- id: PlaylistEntryId,
724724- error: String,
725725- ) -> bool {
726726- tracing::warn!(room = %self.room_id, %id, %error, "playlist download failed");
727727- let effects = vec![Effect::RemovePlaylistEntry(id)];
728728- self.persist_effects(&effects).await;
729729- self.playlist.retain(|e| e.id != id);
730730- self.publish_state_snapshot().await;
731731- false
732732- }
733733-734686 async fn handle_chat(
735687 &mut self,
736688 user_name: String,
···815767 }
816768817769 /// If the room is idle and has playlist entries, auto-fill the next track.
818818- async fn maybe_auto_fill(&mut self) {
770770+ /// Returns `true` if a track was added (so callers can chain post-advance
771771+ /// work).
772772+ async fn maybe_auto_fill(&mut self) -> bool {
819773 let entry = match next_playlist_track(&self.state, &self.playlist, self.playlist_index) {
820774 Some((e, idx)) => {
821775 self.playlist_index = idx;
822776 e.clone()
823777 }
824824- None => return,
778778+ None => return false,
825779 };
826780827781 let id = TrackId::new();
828782 let now = Utc::now().timestamp_millis();
829829- let track_url = entry.url.clone();
783783+ let url = entry.url.clone();
830784 let track = QueuedTrack {
831785 id,
832786 title: entry.title,
833833- url: entry.url,
787787+ url,
834788 duration: entry.duration,
835789 thumbnail: entry.thumbnail,
836790 pending: false,
···845799 }
846800 self.execute_effects(effects).await;
847801848848- // Start the download for this new track.
849849- self.start_download(id, &track_url).await;
802802+ // The newly advanced track starts with started_at: 0. If the file is
803803+ // already cached, resolve it immediately so the frontend loads.
804804+ self.try_start_active().await;
805805+ self.ensure_active_download().await;
806806+ true
850807 }
851808809809+ /// Common post-transition cleanup: resolve the active track's started_at
852810 /// If the active track has a placeholder `started_at_wall` (0), resolve it
853811 /// to now. Used after rehydration where the file was already downloaded.
854812 async fn resolve_active_if_ready(&mut self) {
855813 if let Some(active) = &self.state.active {
856814 if active.started_at_wall == 0
857857- && (self.download_results.contains_key(&active.id)
815815+ && (self.download_results.contains_key(&active.url)
858816 || self
859817 .cache_dir
860860- .join(active.id.as_str_buf())
818818+ .join(file_hash(&active.url))
861819 .with_extension("mp4")
862820 .exists())
863821 {
···872830 async fn publish_state_snapshot(&self) {
873831 let current = self.state.active.as_ref().map(|t| TrackState {
874832 id: t.id,
833833+ hash: file_hash(&t.url),
875834 title: t.title.clone(),
876835 url: t.url.clone(),
877836 duration: t.duration.clone(),
···885844 .iter()
886845 .map(|q| QueueSummary {
887846 id: q.id,
847847+ hash: file_hash(&q.url),
888848 title: q.title.clone(),
889849 url: q.url.clone(),
890850 duration: q.duration.clone(),
···899859 .iter()
900860 .map(|h| HistoryEntry {
901861 id: h.id,
862862+ hash: file_hash(&h.url),
902863 title: h.title.clone(),
903864 url: h.url.clone(),
904865 duration: h.duration.clone(),
···925886 self.publishers.state_tx.send_replace(snapshot);
926887 }
927888928928- /// Spawn a download task for a queued track.
929929- async fn start_download(&mut self, item_id: TrackId, url: &str) {
889889+ /// Ensure a URL has been downloaded (or is in progress). Deduplicated by
890890+ /// URL: if the URL is already downloaded or downloading, this is a no-op.
891891+ /// Safe to call from both queue and playlist paths.
892892+ async fn ensure_downloaded(&mut self, url: &str) {
893893+ let url = url.to_string();
894894+ if self.download_results.contains_key(&url) || self.download_tasks.contains_key(&url) {
895895+ return;
896896+ }
897897+ // Hash the URL for a stable file path, avoiding filename conflicts.
898898+ let hash = file_hash(&url);
899899+ let dest = self.cache_dir.join(&hash);
930900 let cmd_tx = self.cmd_tx.clone();
931931- let dest = self.cache_dir.join(item_id.as_str_buf());
932932- let url = url.to_string();
901901+ let url_clone = url.clone();
933902 let handle = tokio::spawn(async move {
934934- match crate::ingest::download(&url, &dest).await {
903903+ match crate::ingest::download(&url_clone, &dest).await {
935904 Ok(media) => {
936905 let _ = cmd_tx
937937- .send(RoomCommand::DownloadReady { item_id, media })
938938- .await;
939939- }
940940- Err(e) => {
941941- let _ = cmd_tx
942942- .send(RoomCommand::DownloadFailed {
943943- item_id,
944944- error: e.to_string(),
906906+ .send(RoomCommand::DownloadReady {
907907+ url: url_clone,
908908+ media,
945909 })
946910 .await;
947911 }
948948- }
949949- });
950950- self.download_tasks.insert(item_id, handle);
951951- }
952952-953953- /// Spawn a download task for a playlist entry.
954954- async fn start_playlist_download(&mut self, id: PlaylistEntryId, url: &str) {
955955- let cmd_tx = self.cmd_tx.clone();
956956- let dest = self.cache_dir.join(format!("playlist_{}", id.as_str_buf()));
957957- let url = url.to_string();
958958- tokio::spawn(async move {
959959- match crate::ingest::download(&url, &dest).await {
960960- Ok(media) => {
961961- let _ = cmd_tx
962962- .send(RoomCommand::PlaylistDownloadReady { id, media })
963963- .await;
964964- }
965912 Err(e) => {
966913 let _ = cmd_tx
967967- .send(RoomCommand::PlaylistDownloadFailed {
968968- id,
914914+ .send(RoomCommand::DownloadFailed {
915915+ url: url_clone,
969916 error: e.to_string(),
970917 })
971918 .await;
972919 }
973920 }
974921 });
922922+ self.download_tasks.insert(url, handle);
975923 }
976924977925 /// If the active track has a completed download cached, resolve its
···982930 .state
983931 .active
984932 .as_ref()
985985- .map(|t| t.started_at_wall == 0 && self.download_results.contains_key(&t.id))
933933+ .map(|t| t.started_at_wall == 0 && self.download_results.contains_key(&t.url))
986934 .unwrap_or(false);
987935 if should_start {
988936 let now = chrono::Utc::now().timestamp_millis();
···1001949 .queue
1002950 .first()
1003951 .map(|next| {
10041004- !self.download_results.contains_key(&next.id)
10051005- && !self.download_tasks.contains_key(&next.id)
952952+ !self.download_results.contains_key(&next.url)
953953+ && !self.download_tasks.contains_key(&next.url)
1006954 })
1007955 .unwrap_or(false);
1008956 if needs_download {
10091009- let next = self
957957+ let url = self
1010958 .state
1011959 .queue
1012960 .first()
961961+ .map(|t| t.url.clone())
1013962 .expect("queue must be non-empty after needs_download check");
10141014- let id = next.id;
10151015- let url = next.url.clone();
10161016- self.start_download(id, &url).await;
963963+ self.ensure_downloaded(&url).await;
1017964 }
1018965 }
1019966···1026973 .as_ref()
1027974 .map(|t| {
1028975 t.started_at_wall == 0
10291029- && !self.download_results.contains_key(&t.id)
10301030- && !self.download_tasks.contains_key(&t.id)
976976+ && !self.download_results.contains_key(&t.url)
977977+ && !self.download_tasks.contains_key(&t.url)
1031978 })
1032979 .unwrap_or(false);
1033980 if needs_download {
10341034- // Clone before borrow to satisfy the borrow checker.
10351035- let active = self
981981+ let url = self
1036982 .state
1037983 .active
1038984 .as_ref()
985985+ .map(|t| t.url.clone())
1039986 .expect("active must be Some after needs_download check");
10401040- let id = active.id;
10411041- let url = active.url.clone();
10421042- self.start_download(id, &url).await;
987987+ self.ensure_downloaded(&url).await;
1043988 }
1044989 }
1045990}
···10911036 added_by: None,
10921037 added_at: "now".into(),
10931038 pending: false,
10941094- stream_url: None,
10951039 }];
10961040 assert!(next_playlist_track(&state, &playlist, 0).is_none());
10971041 }
···11171061 added_by: None,
11181062 added_at: "now".into(),
11191063 pending: false,
11201120- stream_url: None,
11211064 }];
11221065 assert!(next_playlist_track(&state, &playlist, 0).is_none());
11231066 }
···11421085 added_by: None,
11431086 added_at: "now".into(),
11441087 pending: false,
11451145- stream_url: None,
11461088 }];
11471089 let (entry, index) = next_playlist_track(&state, &playlist, 0).unwrap();
11481090 assert_eq!(entry.id, pid(1));
···11631105 added_by: None,
11641106 added_at: "now".into(),
11651107 pending: false,
11661166- stream_url: None,
11671108 },
11681109 PlaylistEntry {
11691110 id: pid(2),
···11751116 added_by: None,
11761117 added_at: "now".into(),
11771118 pending: false,
11781178- stream_url: None,
11791119 },
11801120 ];
11811121 let (entry, index) = next_playlist_track(&state, &playlist, 1).unwrap();
+621-64
src/state.rs
···3737 Skip,
3838 /// Frontend reported the track finished playing.
3939 TrackEnded { item_id: TrackId },
4040- /// Asynchronous download + metadata extraction completed.
4040+ /// A download failed. Remove the track from wherever it sits and advance.
4141+ TrackFailed { item_id: TrackId },
4242+ /// Download completed: update metadata for ALL queued/active tracks
4343+ /// and playlist entries that share this URL. No item_id needed —
4444+ /// the URL is the deduplication key.
4545+ MetadataResolved {
4646+ url: String,
4747+ title: String,
4848+ duration: String,
4949+ thumbnail: Option<String>,
5050+ source: String,
5151+ },
5252+ /// Asynchronous metadata extraction completed for a specific track.
4153 MetadataUpdated {
4254 item_id: TrackId,
5555+ url: String,
4356 title: String,
4457 duration: String,
4558 thumbnail: Option<String>,
···6376 /// Persist a newly queued track to the store.
6477 PersistQueuedTrack(QueuedTrack),
6578 /// Persist metadata update after async extraction completes.
7979+ /// `url` targets media_cache (canonical metadata store).
6680 PersistMetadata {
6781 item_id: TrackId,
8282+ url: String,
6883 title: String,
6984 duration: String,
7085 thumbnail: Option<String>,
···86101 thumbnail: Option<String>,
87102 source: String,
88103 },
8989- /// Update a playlist entry's metadata after extraction.
9090- UpdatePlaylistEntry {
9191- id: PlaylistEntryId,
9292- title: String,
9393- duration: String,
9494- thumbnail: Option<String>,
9595- source: String,
9696- },
97104 /// Remove a playlist entry.
98105 RemovePlaylistEntry(PlaylistEntryId),
99106}
···119126 Event::TrackQueued(track) => self.handle_queued(track),
120127 Event::Skip => self.handle_skip(now),
121128 Event::TrackEnded { item_id } => self.handle_track_ended(*item_id, now),
129129+ Event::TrackFailed { item_id } => self.handle_track_failed(*item_id, now),
130130+ Event::MetadataResolved {
131131+ url,
132132+ title,
133133+ duration,
134134+ thumbnail,
135135+ source,
136136+ } => self.handle_metadata_resolved(url, title, duration, thumbnail, source),
122137 Event::MetadataUpdated {
123138 item_id,
139139+ url,
124140 title,
125141 duration,
126142 thumbnail,
127143 source,
128128- } => self.handle_metadata_updated(*item_id, title, duration, thumbnail, source),
144144+ } => self.handle_metadata_updated(*item_id, url, title, duration, thumbnail, source),
129145 }
130146 }
131147···148164149165 /// User requested skip.
150166 fn handle_skip(&mut self, _now: i64) -> Vec<Effect> {
151151- let mut effects = Vec::new();
152152-153153- if let Some(active) = self.active.take() {
154154- effects.push(Effect::AbortDownload);
155155- effects.push(Effect::PersistFinished(active.id));
156156- self.history.insert(
157157- 0,
158158- FinishedTrack {
159159- id: active.id,
160160- title: active.title,
161161- url: active.url,
162162- duration: active.duration,
163163- thumbnail: active.thumbnail,
164164- // RoomActor overrides via PersistFinished handler which sets
165165- // the real played_at timestamp in the DB.
166166- played_at: String::new(),
167167- },
168168- );
169169- }
170170-171171- if effects.is_empty() {
172172- return effects;
173173- }
174174-175175- if let Some(next) = self.queue.first().cloned() {
176176- effects.extend(self.advance(next));
177177- } else if !effects.is_empty() {
178178- effects.push(Effect::PublishSnapshot);
179179- }
180180-181181- effects
167167+ self.finish_active_and_advance(true)
182168 }
183169184170 /// Frontend reported the current track finished playing.
185171 fn handle_track_ended(&mut self, item_id: TrackId, _now: i64) -> Vec<Effect> {
186172 // Only react if this matches the currently active track.
187187- let is_current = self
188188- .active
189189- .as_ref()
190190- .map(|t| t.id == item_id)
191191- .unwrap_or(false);
192192-173173+ let is_current = self.active.as_ref().is_some_and(|t| t.id == item_id);
193174 if !is_current {
194175 return Vec::new(); // stale event, ignore
195176 }
177177+ self.finish_active_and_advance(false)
178178+ }
196179180180+ /// Shared logic: move the active track to history, then advance the next
181181+ /// queued track (if any). `abort` controls whether `AbortDownload` is emitted.
182182+ fn finish_active_and_advance(&mut self, abort: bool) -> Vec<Effect> {
197183 let mut effects = Vec::new();
198184199199- if let Some(active) = self.active.take() {
200200- effects.push(Effect::PersistFinished(active.id));
201201- self.history.insert(
202202- 0,
203203- FinishedTrack {
204204- id: active.id,
205205- title: active.title,
206206- url: active.url,
207207- duration: active.duration,
208208- thumbnail: active.thumbnail,
209209- // RoomActor overrides via PersistFinished handler which sets
210210- // the real played_at timestamp in the DB.
211211- played_at: String::new(),
212212- },
213213- );
185185+ let Some(active) = self.active.take() else {
186186+ return effects;
187187+ };
188188+189189+ if abort {
190190+ effects.push(Effect::AbortDownload);
214191 }
192192+ effects.push(Effect::PersistFinished(active.id));
193193+ self.history.insert(
194194+ 0,
195195+ FinishedTrack {
196196+ id: active.id,
197197+ title: active.title,
198198+ url: active.url,
199199+ duration: active.duration,
200200+ thumbnail: active.thumbnail,
201201+ // RoomActor overrides via PersistFinished handler which sets
202202+ // the real played_at timestamp in the DB.
203203+ played_at: String::new(),
204204+ },
205205+ );
215206216207 if let Some(next) = self.queue.first().cloned() {
217208 effects.extend(self.advance(next));
···222213 effects
223214 }
224215216216+ /// A download failed. Remove the track from wherever it sits (active or
217217+ /// queue) and advance if it was the currently playing track.
218218+ fn handle_track_failed(&mut self, item_id: TrackId, _now: i64) -> Vec<Effect> {
219219+ if self.active.as_ref().is_some_and(|t| t.id == item_id) {
220220+ return self.finish_active_and_advance(true);
221221+ }
222222+223223+ if self.queue.iter().any(|t| t.id == item_id) {
224224+ self.queue.retain(|t| t.id != item_id);
225225+ return vec![Effect::PublishSnapshot];
226226+ }
227227+228228+ Vec::new()
229229+ }
230230+225231 /// Asynchronous metadata resolution completed. Updates the matching item
226232 /// in the queue and/or the active track in-place. Does not change playback
227233 /// state, only updates display fields.
228234 fn handle_metadata_updated(
229235 &mut self,
230236 item_id: TrackId,
237237+ url: &str,
231238 title: &str,
232239 duration: &str,
233240 thumbnail: &Option<String>,
···249256 vec![
250257 Effect::PersistMetadata {
251258 item_id,
259259+ url: url.to_string(),
260260+ title: title.to_string(),
261261+ duration: duration.to_string(),
262262+ thumbnail: thumbnail.clone(),
263263+ source: source.to_string(),
264264+ },
265265+ Effect::PublishSnapshot,
266266+ ]
267267+ }
268268+269269+ /// Download completed: update ALL queued and active tracks matching `url`.
270270+ /// Produces a single PersistMetadata effect (media_cache is URL-keyed)
271271+ /// and a PublishSnapshot to broadcast the updated metadata.
272272+ fn handle_metadata_resolved(
273273+ &mut self,
274274+ url: &str,
275275+ title: &str,
276276+ duration: &str,
277277+ thumbnail: &Option<String>,
278278+ source: &str,
279279+ ) -> Vec<Effect> {
280280+ for item in &mut self.queue {
281281+ if item.url == url {
282282+ item.title = title.to_string();
283283+ item.duration = duration.to_string();
284284+ item.thumbnail.clone_from(thumbnail);
285285+ item.pending = false;
286286+ }
287287+ }
288288+ if let Some(ref mut active) = self.active {
289289+ if active.url == url {
290290+ active.title = title.to_string();
291291+ active.duration = duration.to_string();
292292+ active.thumbnail.clone_from(thumbnail);
293293+ }
294294+ }
295295+ vec![
296296+ Effect::PersistMetadata {
297297+ // item_id unused — media_cache is keyed by url
298298+ item_id: TrackId::nil(),
299299+ url: url.to_string(),
252300 title: title.to_string(),
253301 duration: duration.to_string(),
254302 thumbnail: thumbnail.clone(),
···587635 let effects = s.transition(
588636 &Event::MetadataUpdated {
589637 item_id: track_id(1),
638638+ url: "https://example.com/A".into(),
590639 title: "Track A (Real)".into(),
591640 duration: "4:20".into(),
592641 thumbnail: Some("https://img.example/a.jpg".into()),
···594643 },
595644 0,
596645 );
597597-598646 assert_eq!(
599647 effects,
600648 vec![
601649 Effect::PersistMetadata {
602650 item_id: track_id(1),
651651+ url: "https://example.com/A".into(),
603652 title: "Track A (Real)".into(),
604653 duration: "4:20".into(),
605654 thumbnail: Some("https://img.example/a.jpg".into()),
···609658 ]
610659 );
611660612612- // Queue should have empty (track was advanced to active), but active
661661+ // Queue should be empty (track was advanced to active), but active
613662 // should have updated metadata.
614663 assert_eq!(s.active.as_ref().unwrap().title, "Track A (Real)");
615664 assert_eq!(s.active.as_ref().unwrap().duration, "4:20");
···629678 let effects = s.transition(
630679 &Event::MetadataUpdated {
631680 item_id: track_id(2),
681681+ url: "https://example.com/B".into(),
632682 title: "Track B (Real)".into(),
633683 duration: "5:55".into(),
634684 thumbnail: None,
···641691 vec![
642692 Effect::PersistMetadata {
643693 item_id: track_id(2),
694694+ url: "https://example.com/B".into(),
644695 title: "Track B (Real)".into(),
645696 duration: "5:55".into(),
646697 thumbnail: None,
···659710 let effects = s.transition(
660711 &Event::MetadataUpdated {
661712 item_id: track_id(999),
713713+ url: "https://example.com/Ghost".into(),
662714 title: "Ghost".into(),
663715 duration: "0:00".into(),
664716 thumbnail: None,
···673725 vec![
674726 Effect::PersistMetadata {
675727 item_id: track_id(999),
728728+ url: "https://example.com/Ghost".into(),
676729 title: "Ghost".into(),
677730 duration: "0:00".into(),
678731 thumbnail: None,
···683736 );
684737 assert!(s.active.is_none());
685738 assert!(s.queue.is_empty());
739739+ }
740740+741741+ // --- TrackFailed tests ---
742742+743743+ #[test]
744744+ fn test_track_failed_removes_from_queue() {
745745+ let mut s = PlaybackState::default();
746746+ s.transition(&Event::TrackQueued(queued("A", 1)), 0);
747747+ s.transition(&Event::TrackQueued(queued("B", 2)), 0);
748748+ // B is in queue, behind active A.
749749+ let effects = s.transition(
750750+ &Event::TrackFailed {
751751+ item_id: track_id(2),
752752+ },
753753+ 0,
754754+ );
755755+756756+ assert!(!s.queue.iter().any(|t| t.id == track_id(2)));
757757+ assert_eq!(effects, vec![Effect::PublishSnapshot]);
758758+ // A still active; B removed.
759759+ assert_eq!(s.active.as_ref().unwrap().title, "A");
760760+ }
761761+762762+ #[test]
763763+ fn test_track_failed_while_active_acts_like_skip() {
764764+ let mut s = PlaybackState::default();
765765+ s.transition(&Event::TrackQueued(queued("A", 1)), 0);
766766+ // A is now active.
767767+ let effects = s.transition(
768768+ &Event::TrackFailed {
769769+ item_id: track_id(1),
770770+ },
771771+ 0,
772772+ );
773773+774774+ assert_eq!(
775775+ effects,
776776+ vec![
777777+ Effect::AbortDownload,
778778+ Effect::PersistFinished(track_id(1)),
779779+ Effect::PublishSnapshot,
780780+ ]
781781+ );
782782+ assert!(s.active.is_none());
783783+ assert_eq!(s.history.len(), 1);
784784+ assert_eq!(s.history[0].title, "A");
785785+ }
786786+787787+ #[test]
788788+ fn test_track_failed_active_with_next_advances() {
789789+ let mut s = PlaybackState::default();
790790+ s.transition(&Event::TrackQueued(queued("A", 1)), 0);
791791+ s.transition(&Event::TrackQueued(queued("B", 2)), 0);
792792+ // A active, B queued.
793793+ let effects = s.transition(
794794+ &Event::TrackFailed {
795795+ item_id: track_id(1),
796796+ },
797797+ 0,
798798+ );
799799+800800+ assert_eq!(s.active.as_ref().unwrap().title, "B");
801801+ assert!(s.queue.is_empty());
802802+ assert_eq!(s.history.len(), 1);
803803+ assert_eq!(s.history[0].title, "A");
804804+ // PersistFinished for A, PersistStarted for B, PublishSnapshot.
805805+ assert!(effects.contains(&Effect::AbortDownload));
806806+ assert!(effects.contains(&Effect::PersistFinished(track_id(1))));
807807+ assert!(effects.contains(&Effect::PersistStarted(track_id(2))));
808808+ assert!(effects.contains(&Effect::PublishSnapshot));
809809+ }
810810+811811+ #[test]
812812+ fn test_track_failed_stale_id_noop() {
813813+ let mut s = PlaybackState::default();
814814+ s.transition(&Event::TrackQueued(queued("A", 1)), 0);
815815+ let effects = s.transition(
816816+ &Event::TrackFailed {
817817+ item_id: track_id(999),
818818+ },
819819+ 0,
820820+ );
821821+822822+ assert!(effects.is_empty());
823823+ assert_eq!(s.active.as_ref().unwrap().title, "A");
824824+ }
825825+826826+ #[test]
827827+ fn test_track_failed_idle_noop() {
828828+ let mut s = PlaybackState::default();
829829+ let effects = s.transition(
830830+ &Event::TrackFailed {
831831+ item_id: track_id(1),
832832+ },
833833+ 0,
834834+ );
835835+836836+ assert!(effects.is_empty());
837837+ assert!(s.active.is_none());
838838+ assert!(s.queue.is_empty());
839839+ }
840840+841841+ #[test]
842842+ fn test_metadata_resolved_updates_all_matching_by_url() {
843843+ // Four tracks: first advances to active (different URL), three in queue
844844+ // where two share the same URL.
845845+ let mut s = PlaybackState::default();
846846+ s.transition(&Event::TrackQueued(queued("Active", 1)), 0);
847847+ let mut q1 = queued("B", 2);
848848+ q1.url = "https://example.com/shared".into();
849849+ q1.pending = true;
850850+ s.transition(&Event::TrackQueued(q1), 0);
851851+ let mut qc = queued("C-other", 3);
852852+ qc.pending = true;
853853+ s.transition(&Event::TrackQueued(qc), 0);
854854+ let mut q2 = queued("D", 4);
855855+ q2.url = "https://example.com/shared".into();
856856+ q2.pending = true;
857857+ s.transition(&Event::TrackQueued(q2), 0);
858858+859859+ // Queue: [B (shared), C (other), D (shared)] — indices 0, 1, 2.
860860+861861+ let effects = s.transition(
862862+ &Event::MetadataResolved {
863863+ url: "https://example.com/shared".into(),
864864+ title: "Real Title".into(),
865865+ duration: "3:30".into(),
866866+ thumbnail: Some("https://img.example/thumb.jpg".into()),
867867+ source: "ytdlp".into(),
868868+ },
869869+ 0,
870870+ );
871871+872872+ // Matching queue tracks (B at 0, D at 2) updated.
873873+ assert_eq!(s.queue[0].title, "Real Title");
874874+ assert_eq!(s.queue[0].duration, "3:30");
875875+ assert!(!s.queue[0].pending);
876876+877877+ assert_eq!(s.queue[2].title, "Real Title");
878878+ assert_eq!(s.queue[2].duration, "3:30");
879879+ assert!(!s.queue[2].pending);
880880+881881+ // Non-matching queue track (C at 1) unchanged.
882882+ assert_eq!(s.queue[1].title, "C-other");
883883+ assert!(s.queue[1].pending);
884884+885885+ // Active track has different URL, should be unchanged.
886886+ assert_eq!(s.active.as_ref().unwrap().title, "Active");
887887+888888+ assert_eq!(
889889+ effects,
890890+ vec![
891891+ Effect::PersistMetadata {
892892+ item_id: TrackId::nil(),
893893+ url: "https://example.com/shared".into(),
894894+ title: "Real Title".into(),
895895+ duration: "3:30".into(),
896896+ thumbnail: Some("https://img.example/thumb.jpg".into()),
897897+ source: "ytdlp".into(),
898898+ },
899899+ Effect::PublishSnapshot,
900900+ ]
901901+ );
902902+ }
903903+904904+ #[test]
905905+ fn test_metadata_resolved_updates_active_track() {
906906+ let mut s = PlaybackState::default();
907907+ // Queue one track — room is idle, so it advances to active.
908908+ s.transition(&Event::TrackQueued(queued("Active", 1)), 0);
909909+ s.active.as_mut().unwrap().url = "https://example.com/active".into();
910910+911911+ let effects = s.transition(
912912+ &Event::MetadataResolved {
913913+ url: "https://example.com/active".into(),
914914+ title: "Resolved Active".into(),
915915+ duration: "2:00".into(),
916916+ thumbnail: None,
917917+ source: "ytdlp".into(),
918918+ },
919919+ 0,
920920+ );
921921+922922+ assert_eq!(s.active.as_ref().unwrap().title, "Resolved Active");
923923+ assert_eq!(s.active.as_ref().unwrap().duration, "2:00");
924924+925925+ assert_eq!(
926926+ effects,
927927+ vec![
928928+ Effect::PersistMetadata {
929929+ item_id: TrackId::nil(),
930930+ url: "https://example.com/active".into(),
931931+ title: "Resolved Active".into(),
932932+ duration: "2:00".into(),
933933+ thumbnail: None,
934934+ source: "ytdlp".into(),
935935+ },
936936+ Effect::PublishSnapshot,
937937+ ]
938938+ );
939939+ }
940940+}
941941+942942+/// Deterministic simulation testing: random event sequences with invariant
943943+/// checks after every step and at rest.
944944+#[cfg(test)]
945945+mod dst {
946946+ use proptest::prelude::*;
947947+948948+ use super::*;
949949+950950+ fn tid(n: u64) -> TrackId {
951951+ TrackId(uuid::Uuid::from_u64_pair(0, n))
952952+ }
953953+954954+ fn queued(n: u64) -> QueuedTrack {
955955+ QueuedTrack {
956956+ id: tid(n),
957957+ title: format!("Track-{n}"),
958958+ url: format!("https://example.com/track-{n}"),
959959+ duration: "3:00".into(),
960960+ thumbnail: None,
961961+ pending: n % 3 == 0,
962962+ }
963963+ }
964964+965965+ #[derive(Debug, Clone)]
966966+ enum Op {
967967+ Queue(u64),
968968+ Skip,
969969+ TrackEnded(u64),
970970+ TrackFailed(u64),
971971+ MetadataResolved(u64),
972972+ MetadataUpdated(u64),
973973+ }
974974+975975+ fn op_strategy() -> impl Strategy<Value = Vec<Op>> {
976976+ let op = prop_oneof![
977977+ (0u64..12).prop_map(Op::Queue),
978978+ Just(Op::Skip),
979979+ (0u64..12).prop_map(Op::TrackEnded),
980980+ (0u64..12).prop_map(Op::TrackFailed),
981981+ (0u64..12).prop_map(Op::MetadataResolved),
982982+ (0u64..12).prop_map(Op::MetadataUpdated),
983983+ ];
984984+ proptest::collection::vec(op, 0..50)
985985+ }
986986+987987+ fn apply_ops(ops: &[Op]) {
988988+ let mut state = PlaybackState::default();
989989+ let mut queued_ids = std::collections::HashSet::new();
990990+ let mut history_snapshot = Vec::new();
991991+992992+ for op in ops {
993993+ let prev_active_id = state.active.as_ref().map(|a| a.id);
994994+995995+ let effects = match op {
996996+ Op::Queue(n) => {
997997+ if !queued_ids.insert(*n) {
998998+ continue;
999999+ }
10001000+10011001+ let t = queued(*n);
10021002+ state.transition(&Event::TrackQueued(t), 0)
10031003+ }
10041004+ Op::Skip => state.transition(&Event::Skip, 0),
10051005+ Op::TrackEnded(n) => state.transition(&Event::TrackEnded { item_id: tid(*n) }, 0),
10061006+ Op::TrackFailed(n) => {
10071007+ let id = tid(*n);
10081008+ let effects = state.transition(&Event::TrackFailed { item_id: id }, 0);
10091009+10101010+ // TrackFailed must only affect the named track.
10111011+ if prev_active_id.is_some_and(|aid| aid != id) {
10121012+ assert!(
10131013+ state.active.is_some(),
10141014+ "TrackFailed({id}) removed active track {aid}",
10151015+ aid = prev_active_id.unwrap()
10161016+ );
10171017+ assert_eq!(
10181018+ state.active.as_ref().map(|a| a.id),
10191019+ prev_active_id,
10201020+ "TrackFailed({id}) changed active track from {aid}",
10211021+ aid = prev_active_id.unwrap()
10221022+ );
10231023+ }
10241024+ effects
10251025+ }
10261026+ Op::MetadataResolved(n) => {
10271027+ let q = queued(*n);
10281028+ let effects = state.transition(
10291029+ &Event::MetadataResolved {
10301030+ url: q.url.clone(),
10311031+ title: format!("Resolved-{n}"),
10321032+ duration: "4:20".into(),
10331033+ thumbnail: None,
10341034+ source: "ytdlp".into(),
10351035+ },
10361036+ 0,
10371037+ );
10381038+ // Check per-event: all queue tracks matching this URL
10391039+ // must now be !pending.
10401040+ for t in &state.queue {
10411041+ if t.url == q.url {
10421042+ assert!(
10431043+ !t.pending,
10441044+ "queue track {id} still pending after MetadataResolved({url})",
10451045+ id = t.id,
10461046+ url = q.url,
10471047+ );
10481048+ }
10491049+ }
10501050+ effects
10511051+ }
10521052+ Op::MetadataUpdated(n) => {
10531053+ let q = queued(*n);
10541054+ state.transition(
10551055+ &Event::MetadataUpdated {
10561056+ item_id: q.id,
10571057+ url: q.url,
10581058+ title: format!("Updated-{n}"),
10591059+ duration: "5:00".into(),
10601060+ thumbnail: None,
10611061+ source: "ytdlp".into(),
10621062+ },
10631063+ 0,
10641064+ )
10651065+ }
10661066+ };
10671067+10681068+ check_invariants(&state, &effects, &queued_ids, &mut history_snapshot);
10691069+ }
10701070+10711071+ // Final-state crash test: applying any event to the final state
10721072+ // should never panic.
10731073+ let _ = state.transition(&Event::Skip, 0);
10741074+ let _ = state.transition(
10751075+ &Event::MetadataResolved {
10761076+ url: "https://example.com/ghost".into(),
10771077+ title: "Ghost".into(),
10781078+ duration: "0:00".into(),
10791079+ thumbnail: None,
10801080+ source: "ytdlp".into(),
10811081+ },
10821082+ 0,
10831083+ );
10841084+ // Determinism: same event on identical state should produce same effects.
10851085+ let mut clone = state.clone();
10861086+ let e1 = state.transition(&Event::Skip, 0);
10871087+ let e2 = clone.transition(&Event::Skip, 0);
10881088+ assert_eq!(e1, e2, "non-deterministic Skip on final state");
10891089+ }
10901090+10911091+ fn check_invariants(
10921092+ state: &PlaybackState,
10931093+ effects: &[Effect],
10941094+ queued_ids: &std::collections::HashSet<u64>,
10951095+ history_snapshot: &mut Vec<FinishedTrack>,
10961096+ ) {
10971097+ // === State structural invariants ===
10981098+10991099+ // 1. All track IDs across active + queue + history are unique.
11001100+ let mut ids = std::collections::HashSet::new();
11011101+ if let Some(ref a) = state.active {
11021102+ assert!(ids.insert(a.id), "duplicate id in active: {a}", a = a.id);
11031103+ }
11041104+ for t in &state.queue {
11051105+ assert!(ids.insert(t.id), "duplicate id in queue: {t}", t = t.id);
11061106+ }
11071107+ for h in &state.history {
11081108+ assert!(ids.insert(h.id), "duplicate id in history: {h}", h = h.id);
11091109+ }
11101110+11111111+ // 2. Active track is never also in queue or history.
11121112+ if let Some(ref a) = state.active {
11131113+ assert!(
11141114+ !state.queue.iter().any(|t| t.id == a.id),
11151115+ "active track {a} also in queue",
11161116+ a = a.id
11171117+ );
11181118+ assert!(
11191119+ !state.history.iter().any(|h| h.id == a.id),
11201120+ "active track {a} also in history",
11211121+ a = a.id
11221122+ );
11231123+ }
11241124+11251125+ // 3. Queue and history are disjoint.
11261126+ for h in &state.history {
11271127+ assert!(
11281128+ !state.queue.iter().any(|t| t.id == h.id),
11291129+ "track {h} in both queue and history",
11301130+ h = h.id
11311131+ );
11321132+ }
11331133+11341134+ // 4. Conservation: every track in active/queue/history was once queued.
11351135+ for id in &ids {
11361136+ let n = id.0.as_u64_pair().1;
11371137+ assert!(
11381138+ queued_ids.contains(&n),
11391139+ "orphan track id {id} was never queued"
11401140+ );
11411141+ }
11421142+11431143+ // 5. History is monotonic: never shrinks, and held tracks are stable.
11441144+ assert!(
11451145+ state.history.len() >= history_snapshot.len(),
11461146+ "history shrank from {} to {}",
11471147+ history_snapshot.len(),
11481148+ state.history.len()
11491149+ );
11501150+ for prev in history_snapshot.iter() {
11511151+ assert!(
11521152+ state.history.iter().any(|h| h.id == prev.id),
11531153+ "history lost track {id}",
11541154+ id = prev.id
11551155+ );
11561156+ }
11571157+ *history_snapshot = state.history.clone();
11581158+11591159+ // === Effect composition invariants ===
11601160+11611161+ // 6. Within a single transition, all Persist* effects precede
11621162+ // PublishSnapshot. There may be zero or one PublishSnapshot.
11631163+ let publish_pos = effects
11641164+ .iter()
11651165+ .position(|e| matches!(e, Effect::PublishSnapshot));
11661166+ if let Some(pos) = publish_pos {
11671167+ for e in &effects[pos + 1..] {
11681168+ assert!(
11691169+ !is_persist_effect(e),
11701170+ "persist effect {e:?} appears after PublishSnapshot"
11711171+ );
11721172+ }
11731173+ }
11741174+11751175+ // 7. State-changing effects must be followed by PublishSnapshot.
11761176+ let needs_publish = effects.iter().any(|e| {
11771177+ matches!(
11781178+ e,
11791179+ Effect::PersistStarted(_)
11801180+ | Effect::PersistFinished(_)
11811181+ | Effect::PersistQueuedTrack(_)
11821182+ | Effect::PersistMetadata { .. }
11831183+ | Effect::AbortDownload
11841184+ )
11851185+ });
11861186+ if needs_publish {
11871187+ assert!(
11881188+ effects.iter().any(|e| matches!(e, Effect::PublishSnapshot)),
11891189+ "state-changing effects without PublishSnapshot: {effects:?}"
11901190+ );
11911191+ }
11921192+11931193+ // 8. PersistStarted and PersistFinished for the same track can't
11941194+ // appear in the same transition.
11951195+ let started: std::collections::HashSet<TrackId> = effects
11961196+ .iter()
11971197+ .filter_map(|e| {
11981198+ if let Effect::PersistStarted(id) = e {
11991199+ Some(*id)
12001200+ } else {
12011201+ None
12021202+ }
12031203+ })
12041204+ .collect();
12051205+ for e in effects {
12061206+ if let Effect::PersistFinished(id) = e {
12071207+ assert!(
12081208+ !started.contains(id),
12091209+ "PersistStarted and PersistFinished for same track: {id}"
12101210+ );
12111211+ }
12121212+ }
12131213+12141214+ // 9. AbortDownload always accompanies PersistFinished (only fired
12151215+ // when the active track is aborted during skip/fail).
12161216+ if effects.iter().any(|e| matches!(e, Effect::AbortDownload)) {
12171217+ assert!(
12181218+ effects
12191219+ .iter()
12201220+ .any(|e| matches!(e, Effect::PersistFinished(_))),
12211221+ "AbortDownload without PersistFinished"
12221222+ );
12231223+ }
12241224+ }
12251225+12261226+ fn is_persist_effect(e: &Effect) -> bool {
12271227+ matches!(
12281228+ e,
12291229+ Effect::PersistStarted(_)
12301230+ | Effect::PersistFinished(_)
12311231+ | Effect::PersistQueuedTrack(_)
12321232+ | Effect::PersistMetadata { .. }
12331233+ | Effect::AddPlaylistEntry { .. }
12341234+ | Effect::RemovePlaylistEntry(_)
12351235+ )
12361236+ }
12371237+12381238+ proptest! {
12391239+ #[test]
12401240+ fn dst_random_sequences_invariants_hold(ops in op_strategy()) {
12411241+ apply_ops(&ops);
12421242+ }
6861243 }
6871244}
+99-113
src/store.rs
···1818use crate::types::{ChatMessage, ChatMessageId, PlaylistEntryId, TrackId};
19192020/// An entry in a room's perpetual playlist — used for auto-fill when the
2121-/// queue runs dry. Metadata is pre-resolved at add-time so auto-played
2222-/// tracks never show a "Loading..." placeholder.
2121+/// queue runs dry. Metadata is pre-resolved via the [`MEDIA_CACHE`] table so
2222+/// auto-played tracks never show a "Loading..." placeholder.
2323+///
2424+/// The `title`, `duration`, `thumbnail`, `source`, and `pending` fields are
2525+/// populated at load time via a JOIN with `media_cache`; they are NOT stored
2626+/// independently in the `playlist_entries` table.
2327#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize)]
2428pub(crate) struct PlaylistEntry {
2529 pub id: PlaylistEntryId,
···3034 pub source: String,
3135 pub added_by: Option<String>,
3236 pub added_at: String,
3737+ /// Derived from `media_cache.state` at load time.
3338 pub pending: bool,
3434- /// Pre-resolved direct stream URL, cached at extraction time.
3535- pub stream_url: Option<String>,
3639}
37403841/// Full snapshot of a room's state loaded from the store.
···110113 added_by: Option<String>,
111114 added_at: String,
112115 pending: bool,
113113- stream_url: Option<String>,
114116}
115117116118impl From<InMemoryPlaylistEntry> for PlaylistEntry {
···125127 added_by: e.added_by,
126128 added_at: e.added_at,
127129 pending: e.pending,
128128- stream_url: e.stream_url,
129130 }
130131 }
131132}
···231232 }
232233 Effect::PersistMetadata {
233234 item_id,
235235+ url: _,
234236 title,
235237 duration,
236238 thumbnail,
···285287 added_by: None,
286288 added_at: crate::util::now_iso(),
287289 pending: true,
288288- stream_url: None,
289290 });
290291 }
291291- Effect::UpdatePlaylistEntry {
292292- id,
293293- title,
294294- duration,
295295- thumbnail,
296296- source,
297297- } => {
298298- if let Some(entry) = room.playlist.iter_mut().find(|e| e.id == *id) {
299299- entry.title.clone_from(title);
300300- entry.duration.clone_from(duration);
301301- entry.thumbnail.clone_from(thumbnail);
302302- entry.source.clone_from(source);
303303- entry.pending = false;
304304- }
305305- }
306292 Effect::RemovePlaylistEntry(id) => {
307293 room.playlist.retain(|e| e.id != *id);
308294 }
···407393 .execute(&pool)
408394 .await?;
409395396396+ // Canonical resolved media keyed by URL. Tracks and playlist entries
397397+ // reference this table to avoid duplicating metadata and downloads.
398398+ sqlx::query(
399399+ "CREATE TABLE IF NOT EXISTS media_cache (
400400+ url TEXT PRIMARY KEY,
401401+ title TEXT NOT NULL DEFAULT '',
402402+ duration TEXT NOT NULL DEFAULT '--:--',
403403+ thumbnail TEXT,
404404+ source TEXT NOT NULL DEFAULT '',
405405+ file_path TEXT NOT NULL DEFAULT '',
406406+ state TEXT NOT NULL DEFAULT 'pending'
407407+ )",
408408+ )
409409+ .execute(&pool)
410410+ .await?;
411411+410412 sqlx::query(
411413 "CREATE TABLE IF NOT EXISTS tracks (
412414 id TEXT PRIMARY KEY,
413415 room_id TEXT NOT NULL REFERENCES rooms(id) ON DELETE CASCADE,
414416 url TEXT NOT NULL,
415415- title TEXT NOT NULL,
416416- duration TEXT NOT NULL DEFAULT '--:--',
417417- thumbnail TEXT,
418418- source TEXT NOT NULL DEFAULT '',
419417 position INTEGER NOT NULL,
420418 added_at TEXT NOT NULL,
421419 played INTEGER NOT NULL DEFAULT 0,
422420 played_at TEXT,
423423- started_at_ms INTEGER,
424424- pending INTEGER NOT NULL DEFAULT 1
421421+ started_at_ms INTEGER
425422 )",
426423 )
427424 .execute(&pool)
···453450 id TEXT PRIMARY KEY,
454451 room_id TEXT NOT NULL REFERENCES rooms(id) ON DELETE CASCADE,
455452 url TEXT NOT NULL,
456456- title TEXT NOT NULL DEFAULT '',
457457- duration TEXT NOT NULL DEFAULT '--:--',
458458- thumbnail TEXT,
459459- source TEXT NOT NULL DEFAULT '',
460453 added_by TEXT,
461461- added_at TEXT NOT NULL,
462462- pending INTEGER NOT NULL DEFAULT 1,
463463- stream_url TEXT
454454+ added_at TEXT NOT NULL
464455 )",
465456 )
466457 .execute(&pool)
···495486 .bind(room_id)
496487 .fetch_one(&mut *tx)
497488 .await?;
489489+490490+ // Ensure a media_cache row exists for deduplication.
491491+ sqlx::query("INSERT OR IGNORE INTO media_cache (url) VALUES (?1)")
492492+ .bind(&track.url)
493493+ .execute(&mut *tx)
494494+ .await?;
498495499496 sqlx::query(
500500- "INSERT INTO tracks
501501- (id, room_id, url, title, duration, thumbnail, source,
502502- position, added_at, pending)
503503- VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10)",
497497+ "INSERT INTO tracks (id, room_id, url, position, added_at)
498498+ VALUES (?1, ?2, ?3, ?4, ?5)",
504499 )
505500 .bind(track.id.as_str_buf())
506501 .bind(room_id)
507502 .bind(&track.url)
508508- .bind(&track.title)
509509- .bind(&track.duration)
510510- .bind(track.thumbnail.as_deref())
511511- .bind("")
512503 .bind(position)
513504 .bind(crate::util::now_iso())
514514- .bind(track.pending as i64)
515505 .execute(&mut *tx)
516506 .await?;
517507 }
···534524 .await?;
535525 }
536526 Effect::PersistMetadata {
537537- item_id,
527527+ item_id: _,
528528+ url,
538529 title,
539530 duration,
540531 thumbnail,
541532 source,
542533 } => {
543543- let src_str = source.to_string();
544534 sqlx::query(
545545- "UPDATE tracks
535535+ "UPDATE media_cache
546536 SET title = ?1, duration = ?2, thumbnail = ?3,
547547- source = ?4, pending = 0
548548- WHERE id = ?5",
537537+ source = ?4, state = 'ready'
538538+ WHERE url = ?5",
549539 )
550540 .bind(title.as_str())
551541 .bind(duration.as_str())
552542 .bind(thumbnail.as_deref())
553553- .bind(&src_str)
554554- .bind(item_id.as_str_buf())
543543+ .bind(source.as_str())
544544+ .bind(url.as_str())
555545 .execute(&mut *tx)
556546 .await?;
557547 }
···577567 .await?;
578568 output.chat_message_id = Some(id);
579569 }
580580- Effect::AddPlaylistEntry {
581581- id,
582582- url,
583583- title,
584584- duration,
585585- thumbnail,
586586- source,
587587- } => {
570570+ Effect::AddPlaylistEntry { id, url, .. } => {
571571+ // Ensure a media_cache row exists for deduplication.
572572+ sqlx::query("INSERT OR IGNORE INTO media_cache (url) VALUES (?1)")
573573+ .bind(url.as_str())
574574+ .execute(&mut *tx)
575575+ .await?;
576576+588577 sqlx::query(
589589- "INSERT INTO playlist_entries (id, room_id, url, title, duration, thumbnail, source, added_at, pending)
590590- VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, 1)",
578578+ "INSERT INTO playlist_entries (id, room_id, url, added_at)
579579+ VALUES (?1, ?2, ?3, ?4)",
591580 )
592581 .bind(id.as_str_buf())
593582 .bind(room_id)
594583 .bind(url.as_str())
595595- .bind(title.as_str())
596596- .bind(duration.as_str())
597597- .bind(thumbnail.as_deref())
598598- .bind(source.as_str())
599584 .bind(crate::util::now_iso())
600585 .execute(&mut *tx)
601586 .await?;
602587 }
603603- Effect::UpdatePlaylistEntry {
604604- id,
605605- title,
606606- duration,
607607- thumbnail,
608608- source,
609609- } => {
610610- sqlx::query(
611611- "UPDATE playlist_entries SET title = ?1, duration = ?2, thumbnail = ?3, source = ?4, pending = 0 WHERE id = ?5",
612612- )
613613- .bind(title.as_str())
614614- .bind(duration.as_str())
615615- .bind(thumbnail.as_deref())
616616- .bind(source.as_str())
617617- .bind(id.as_str_buf())
618618- .execute(&mut *tx)
619619- .await?;
620620- }
621588 Effect::RemovePlaylistEntry(id) => {
622589 sqlx::query("DELETE FROM playlist_entries WHERE id = ?")
623590 .bind(id.as_str_buf())
···633600 }
634601}
635602636636-/// Helper row type for deserialising `tracks` rows.
603603+/// Helper row type for deserialising tracks JOINed with media_cache.
637604#[derive(sqlx::FromRow)]
638605struct TrackRow {
639606 id: String,
640640- title: String,
641607 url: String,
608608+ played_at: Option<String>,
609609+ title: String,
642610 duration: String,
643611 thumbnail: Option<String>,
644644- played_at: Option<String>,
645645- pending: i64,
612612+ state: String,
646613}
647614648648-/// Helper row type for deserialising `playlist_entries` rows.
615615+/// Helper row type for deserialising playlist_entries JOINed with media_cache.
649616#[derive(sqlx::FromRow)]
650617struct PlaylistRow {
651618 id: String,
652619 url: String,
620620+ added_by: Option<String>,
621621+ added_at: String,
653622 title: String,
654623 duration: String,
655624 thumbnail: Option<String>,
656625 source: String,
657657- added_by: Option<String>,
658658- added_at: String,
659659- pending: i64,
660660- stream_url: Option<String>,
626626+ state: String,
661627}
662628663629/// Helper row type for deserialising `chat_messages` rows.
···742708 };
743709744710 // 2. Queue: not played, not started, ordered by position.
711711+ // JOIN media_cache for resolved metadata.
745712 let queue_rows: Vec<TrackRow> = sqlx::query_as(
746746- "SELECT id, url, title, duration, thumbnail, played_at, pending
747747- FROM tracks
748748- WHERE room_id = ? AND played = 0 AND started_at_ms IS NULL
749749- ORDER BY position",
713713+ "SELECT t.id, t.url, t.played_at,
714714+ COALESCE(m.title, 'Loading...') AS title,
715715+ COALESCE(m.duration, '--:--') AS duration,
716716+ m.thumbnail, m.state
717717+ FROM tracks t
718718+ LEFT JOIN media_cache m ON m.url = t.url
719719+ WHERE t.room_id = ? AND t.played = 0 AND t.started_at_ms IS NULL
720720+ ORDER BY t.position",
750721 )
751722 .bind(room_id)
752723 .fetch_all(&self.pool)
···754725755726 // 3. Active: not played, started (at most one).
756727 let active_row: Option<TrackRow> = sqlx::query_as(
757757- "SELECT id, url, title, duration, thumbnail, played_at, pending
758758- FROM tracks
759759- WHERE room_id = ? AND played = 0 AND started_at_ms IS NOT NULL
760760- ORDER BY position
728728+ "SELECT t.id, t.url, t.played_at,
729729+ COALESCE(m.title, 'Loading...') AS title,
730730+ COALESCE(m.duration, '--:--') AS duration,
731731+ m.thumbnail, m.state
732732+ FROM tracks t
733733+ LEFT JOIN media_cache m ON m.url = t.url
734734+ WHERE t.room_id = ? AND t.played = 0 AND t.started_at_ms IS NOT NULL
735735+ ORDER BY t.position
761736 LIMIT 1",
762737 )
763738 .bind(room_id)
···766741767742 // 4. History: played, newest first.
768743 let history_rows: Vec<TrackRow> = sqlx::query_as(
769769- "SELECT id, url, title, duration, thumbnail, played_at, pending
770770- FROM tracks
771771- WHERE room_id = ? AND played = 1
772772- ORDER BY played_at DESC",
744744+ "SELECT t.id, t.url, t.played_at,
745745+ COALESCE(m.title, 'Unknown') AS title,
746746+ COALESCE(m.duration, '--:--') AS duration,
747747+ m.thumbnail, m.state
748748+ FROM tracks t
749749+ LEFT JOIN media_cache m ON m.url = t.url
750750+ WHERE t.room_id = ? AND t.played = 1
751751+ ORDER BY t.played_at DESC",
773752 )
774753 .bind(room_id)
775754 .fetch_all(&self.pool)
···787766 .fetch_all(&self.pool)
788767 .await?;
789768790790- // 6. Playlist entries, ordered by id.
769769+ // 6. Playlist entries with media_cache metadata.
791770 let playlist_rows: Vec<PlaylistRow> = sqlx::query_as(
792792- "SELECT id, url, title, duration, thumbnail, source, added_by, added_at, pending, stream_url
793793- FROM playlist_entries
794794- WHERE room_id = ?
795795- ORDER BY id",
771771+ "SELECT p.id, p.url, p.added_by, p.added_at,
772772+ COALESCE(m.title, 'Loading...') AS title,
773773+ COALESCE(m.duration, '--:--') AS duration,
774774+ m.thumbnail, m.source, m.state
775775+ FROM playlist_entries p
776776+ LEFT JOIN media_cache m ON m.url = p.url
777777+ WHERE p.room_id = ?
778778+ ORDER BY p.id",
796779 )
797780 .bind(room_id)
798781 .fetch_all(&self.pool)
799782 .await?;
783783+784784+ let pending_from_state = |state: &str| state == "pending";
800785801786 let queue: Vec<QueuedTrack> = queue_rows
802787 .into_iter()
···806791 url: r.url,
807792 duration: r.duration,
808793 thumbnail: r.thumbnail,
809809- pending: r.pending != 0,
794794+ pending: pending_from_state(&r.state),
810795 })
811796 .collect();
812797···816801 url: r.url,
817802 duration: r.duration,
818803 thumbnail: r.thumbnail,
819819- pending: r.pending != 0,
804804+ pending: pending_from_state(&r.state),
820805 });
821806822807 let history: Vec<FinishedTrack> = history_rows
···853838 source: r.source,
854839 added_by: r.added_by,
855840 added_at: r.added_at,
856856- pending: r.pending != 0,
857857- stream_url: r.stream_url,
841841+ pending: pending_from_state(&r.state),
858842 })
859843 .collect();
860844···11181102 "r",
11191103 &[Effect::PersistMetadata {
11201104 item_id: tid(1),
11051105+ url: "https://e.com/t".into(),
11211106 title: "Real Title".into(),
11221107 duration: "4:20".into(),
11231108 thumbnail: Some("https://img.example/t.jpg".into()),
···11641149 "r",
11651150 &[Effect::PersistMetadata {
11661151 item_id: tid(1),
11521152+ url: "https://e.com/t".into(),
11671153 title: "Real Title".into(),
11681154 duration: "4:20".into(),
11691155 thumbnail: None,
+12
src/types.rs
···1616 Self(uuid::Uuid::now_v7())
1717 }
18181919+ /// Sentinel value: no specific track. Used in PersistMetadata effects
2020+ /// generated by URL-keyed operations where the effect targets media_cache,
2121+ /// not a particular track row.
2222+ pub(crate) fn nil() -> Self {
2323+ Self(uuid::Uuid::nil())
2424+ }
2525+1926 pub(crate) fn as_str_buf(&self) -> String {
2027 self.0.to_string()
2128 }
···103110#[derive(Debug, Clone, Serialize)]
104111pub(crate) struct TrackState {
105112 pub(crate) id: TrackId,
113113+ /// Stable hash of the media URL, used for file serving.
114114+ pub(crate) hash: String,
106115 pub(crate) title: String,
107116 pub(crate) url: String,
108117 pub(crate) duration: String,
···114123#[derive(Debug, Clone, Serialize)]
115124pub(crate) struct QueueSummary {
116125 pub(crate) id: TrackId,
126126+ /// Stable hash of the media URL, used for file serving.
127127+ pub(crate) hash: String,
117128 pub(crate) title: String,
118129 pub(crate) url: String,
119130 pub(crate) duration: String,
···130141#[derive(Debug, Clone, Serialize)]
131142pub(crate) struct HistoryEntry {
132143 pub(crate) id: TrackId,
144144+ pub(crate) hash: String,
133145 pub(crate) title: String,
134146 pub(crate) url: String,
135147 pub(crate) duration: String,
+12-23
src/web.rs
···2121use crate::room;
2222use crate::types::{PlaylistEntryId, RoomId};
23232424-const MEDIA_EXTENSIONS: &[&str] = &[
2525- "mp4", "m4a", "mp3", "webm", "mkv", "ogg", "wav", "flac", "aac", "opus", "mov",
2626-];
2727-2824/// Per-IP rate limiter for POST /api/ingest (max 1 req/5s per IP).
2925#[derive(Clone)]
3026pub(crate) struct RateLimiter {
···8985 .route("/api/v1/skip", post(skip_handler))
9086 .route("/api/v1/playlist/add", post(add_playlist_entry))
9187 .route("/api/v1/playlist/remove", post(remove_playlist_entry))
9292- .route(
9393- "/api/v1/rooms/{room_id}/media/{track_id}",
9494- get(media_handler),
9595- )
8888+ .route("/api/v1/rooms/{room_id}/media/{hash}", get(media_handler))
9689 .route("/api/v1/ws/{room_id}", get(ws_handler))
9790 // SPA: serve built frontend from static/dist/
9891 .nest_service(
···132125 Ok(Json(serde_json::json!({ "room_id": room.to_string() })))
133126}
134127135135-/// Find a media file in a room's cache directory by trying extensions.
136136-/// yt-dlp downloads to `{track_id}.{ext}` but the media handler receives
137137-/// the bare track ID (UUID) without extension.
138138-async fn find_media_file(dir: &std::path::Path, track_id: &str) -> Option<std::path::PathBuf> {
139139- // Try bare path first (was renamed or is an exact match).
140140- let bare = dir.join(track_id);
128128+/// Find a media file in the cache directory by hash (derived from URL).
129129+async fn find_media_file(dir: &std::path::Path, hash: &str) -> Option<std::path::PathBuf> {
130130+ // Try exact hash path, with mp4 extension.
131131+ let path = dir.join(hash).with_extension("mp4");
132132+ if path.exists() {
133133+ return Some(path);
134134+ }
135135+ // Fall back to bare hash (no extension), for any legacy files.
136136+ let bare = dir.join(hash);
141137 if bare.exists() {
142138 return Some(bare);
143143- }
144144- // Scan for `{track_id}.{ext}` with common media extensions.
145145- for ext in MEDIA_EXTENSIONS {
146146- let path = dir.join(format!("{track_id}.{ext}"));
147147- if path.exists() {
148148- return Some(path);
149149- }
150139 }
151140 None
152141}
···271260272261async fn media_handler(
273262 State(state): State<AppState>,
274274- Path((room_id, track_id)): Path<(String, String)>,
263263+ Path((room_id, hash)): Path<(String, String)>,
275264 headers: axum::http::HeaderMap,
276265) -> Response {
277266 let room_id = match RoomId::parse(&room_id) {
···283272 return (StatusCode::NOT_FOUND, "room not found").into_response();
284273 }
285274286286- let file_path = find_media_file(&state.cache_dir, &track_id).await;
275275+ let file_path = find_media_file(&state.cache_dir, &hash).await;
287276288277 let Some(file_path) = file_path else {
289278 return (StatusCode::NOT_FOUND, "media not found").into_response();