Harden Spotify API interaction

Author: udoprog

Cargo.lock

@@ -11,26 +11,27 @@ argwerk = "0.20.0"
 eudex = "0.1.1"
 chrono = { version = "0.4.19", features = ["serde"] }
 chrono-tz = { version = "0.5.3", features = ["serde"] }
-mysql_async = "0.26.0"
+mysql_async = "0.27.0"
 diesel = { version = "1.4.5", features = ["sqlite", "chrono"] }
 diesel_migrations = "1.4.0"
 # set the bundled feature to use the bundled libsqlite3
 libsqlite3-sys = { version = "0.17.3", features = ["bundled", "unlock_notify"] }
-log = "0.4.13"
+log = "0.4.14"
 relative-path = { version = "1.3.2", features = ["serde"] }
-serde = { version = "1.0.120", features = ["rc"] }
-serde_yaml = "0.8.15"
-serde_json = "1.0.61"
+serde = { version = "1.0.123", features = ["rc"] }
+serde_yaml = "0.8.17"
+serde_json = "1.0.62"
 serde_cbor = "0.11.1"
 serde_urlencoded = "0.7.0"
 reqwest = "0.11.0"
 Inflector = "0.11.4"
 warp = "0.3.0"
-tokio = { version = "1.1.0", features = ["full"] }
-tokio-tungstenite = { version = "0.13.0", features = ["tls"] }
-tokio-stream = "0.1.2"
+tokio = { version = "1.2.0", features = ["full"] }
+tokio-tungstenite = { version = "0.14.0", features = ["rustls-tls"] }
+tungstenite = "0.13.0"
+tokio-stream = "0.1.3"
 url = { version = "2.2.0", features = ["serde"] }
-rand = "0.8.2"
+rand = "0.8.3"
 fixed-map = "0.7.1"
 log4rs = "1.0.0"
 handlebars = "3.5.2"
@@ -42,7 +43,7 @@ bytes = "1.0.1"
 uuid = { version = "0.8.2", features = ["serde", "v4"] }
 smallvec = { version = "1.6.1", features = ["serde"] }
 dirs = "3.0.1"
-backoff = "0.2.1"
+backoff = "0.3.0"
 rust-embed = { version = "5.9.0", features = ["interpolate-folder-path"] }
 mime = "0.3.16"
 mime_guess = "2.0.3"
@@ -62,14 +63,13 @@ regex = "1.4.3"
 backtrace = "0.3.56"
 futures-util = "0.3.12"
 futures-core = "0.3.12"
-tracing = "0.1.22"
+tracing = "0.1.23"
 tracing-core = "0.1.17"
 tracing-futures = { version = "0.2.4", default-features = false, features = ["std-future"] }
 slab = "0.4.2"
 irc = "0.15.0"
 ignore = "0.4.17"
-notify = "5.0.0-pre.4"
-tungstenite = "0.12.0"
+notify = "5.0.0-pre.5"
 
 runestick = { version = "0.8.0", optional = true }
 rune = { version = "0.8.0", features = ["diagnostics"], optional = true }

bot/Cargo.toml

@@ -12,6 +12,7 @@ pub use self::model::user::PrivateUser;
 use crate::api::RequestBuilder;
 use crate::oauth2;
 use crate::prelude::*;
+use crate::spotify_id::SpotifyId;
 use anyhow::Result;
 use bytes::Bytes;
 use reqwest::{header, Client, Method, StatusCode};
@@ -22,6 +23,7 @@ use url::Url;
 mod model;
 
 const API_URL: &str = "https://api.spotify.com/v1";
+const DEFAULT_LIMIT: usize = 50;
 
 /// API integration.
 #[derive(Clone, Debug)]
@@ -56,9 +58,10 @@ impl Spotify {
     }
 
     /// Get my playlists.
-    pub async fn playlist(&self, id: String, market: Option<&str>) -> Result<FullPlaylist> {
+    pub async fn playlist(&self, id: SpotifyId, market: Option<&str>) -> Result<FullPlaylist> {
         let req = self
-            .request(Method::GET, &["playlists", id.as_str()])
+            .request(Method::GET, &["playlists", id.to_string().as_str()])
+            .query_param("limit", &DEFAULT_LIMIT.to_string())
             .optional_query_param("market", market);
 
         req.execute().await?.json()
@@ -190,14 +193,11 @@ impl Spotify {
 
     /// Get my songs.
     pub async fn my_tracks(&self) -> Result<Page<SavedTrack>> {
-        let req = self.request(Method::GET, &["me", "tracks"]);
-        req.execute().await?.json()
-    }
+        let req = self
+            .request(Method::GET, &["me", "tracks"])
+            .query_param("limit", &DEFAULT_LIMIT.to_string());
 
-    /// Get my songs.
-    pub fn my_tracks_stream(&self) -> PageStream<SavedTrack> {
-        let req = self.request(Method::GET, &["me", "tracks"]);
-        self.page_stream(async move { req.execute().await?.json() })
+        req.execute().await?.json()
     }
 
     /// Get the full track by ID.
@@ -252,6 +252,7 @@ fn device_control<C>(status: StatusCode, _: &C) -> Result<Option<bool>> {
     }
 }
 
+/// A page converted into a stream which will perform pagination under the hood.
 pub struct PageStream<T> {
     client: Client,
     token: oauth2::SyncToken,
@@ -283,15 +284,16 @@ where
             None => return Poll::Ready(None),
         };
 
-        if let Poll::Ready(page) = future.as_mut().poll(cx)? {
-            self.as_mut().next = match page.next.map(|s| str::parse(s.as_str())).transpose()? {
-                Some(next) => Some(Box::pin(self.next_page(next))),
-                None => None,
-            };
+        let page = match future.as_mut().poll(cx)? {
+            Poll::Ready(page) => page,
+            Poll::Pending => return Poll::Pending,
+        };
 
-            return Poll::Ready(Some(Ok(page.items)));
-        }
+        self.as_mut().next = match page.next.map(|s| str::parse(s.as_str())).transpose()? {
+            Some(next) => Some(Box::pin(self.next_page(next))),
+            None => None,
+        };
 
-        Poll::Pending
+        Poll::Ready(Some(Ok(page.items)))
     }
 }

bot/src/api/spotify/mod.rs

@@ -108,13 +108,13 @@ macro_rules! retry_until_ok {
 
             loop {
                 log::info!("{}", $id);
-                let res = async { $($f)* }.await;
+                let res: anyhow::Result<_> = async { $($f)* }.await;
 
                 match res {
                     Ok(output) => break output,
                     Err(e) => {
                         let duration = backoff.next();
-                        log_warn!(e, "\"{}\" failed, trying again in {:?}", $id, duration);
+                        log_warn!(e, "{} failed, trying again in {:?}", $id, duration);
                         tokio::time::sleep(duration).await;
                     }
                 }

bot/src/macros.rs

@@ -42,7 +42,8 @@ const PACKAGE: &str = env!("CARGO_PKG_NAME");
 #[cfg(feature = "windows")]
 mod internal {
     use super::FILE;
-    use log4rs::config::{Config, ConfigBuilder, Logger, LoggerBuilder, Root, RootBuilder};
+    use log4rs::config::runtime::{ConfigBuilder, LoggerBuilder, RootBuilder};
+    use log4rs::config::{Config, Logger, Root};
 
     pub(crate) fn logger_builder() -> LoggerBuilder {
         Logger::builder().appender(FILE).additive(false)

bot/src/main.rs

@@ -81,9 +81,9 @@ pub(super) async fn setup(
 
 #[derive(Debug, Error)]
 pub(super) enum ConnectError {
-    #[error("no device configured or available")]
-    NoDevice,
-    #[error("error when issuing {0} command")]
+    #[error("{0}: no device configured or available")]
+    NoDevice(&'static str),
+    #[error("{0}: error")]
     Error(&'static str, #[source] Error),
 }
 
@@ -92,7 +92,7 @@ impl ConnectError {
         match result {
             Err(e) => Err(ConnectError::Error(what, e.into())),
             Ok(true) => Ok(()),
-            Ok(false) => Err(ConnectError::NoDevice),
+            Ok(false) => Err(ConnectError::NoDevice(what)),
         }
     }
 }
@@ -171,6 +171,7 @@ impl ConnectPlayer {
         self.volume(player::ModifyVolume::Set(update)).await
     }
 
+    /// Modify the volume of the player.
     pub(super) async fn volume(&self, modify: player::ModifyVolume) -> Result<u32, ConnectError> {
         let volume = self.volume.load().await;
         let update = modify.apply(volume);
@@ -181,6 +182,7 @@ impl ConnectPlayer {
         Ok(update)
     }
 
+    /// Get the current volume of the player.
     pub(super) async fn current_volume(&self) -> u32 {
         self.volume.load().await
     }

bot/src/player/connect.rs

@@ -14,15 +14,15 @@ pub struct Item {
 impl Item {
     /// Human readable version of playback item.
     pub fn what(&self) -> String {
-        match self.track {
-            Track::Spotify { ref track } => {
+        match &self.track {
+            Track::Spotify { track } => {
                 if let Some(artists) = utils::human_artists(&track.artists) {
                     format!("\"{}\" by {}", track.name, artists)
                 } else {
                     format!("\"{}\"", track.name)
                 }
             }
-            Track::YouTube { ref video } => match video.snippet.as_ref() {
+            Track::YouTube { video } => match video.snippet.as_ref() {
                 Some(snippet) => match snippet.channel_title.as_ref() {
                     Some(channel_title) => {
                         format!("\"{}\" from \"{}\"", snippet.title, channel_title)
@@ -34,17 +34,11 @@ impl Item {
         }
     }
 
+    /// Test if the given item is playable.
     pub fn is_playable(&self) -> bool {
-        match self.track {
-            Track::Spotify { ref track } => {
-                match track.is_playable {
-                    Some(is_playable) => return is_playable,
-                    None => return false,
-                };
-            }
-            Track::YouTube { video: _ } => {
-                return true;
-            }
+        match &self.track {
+            Track::Spotify { track } => track.is_playable.unwrap_or(true),
+            Track::YouTube { video: _ } => true,
         }
     }
 }

bot/src/player/item.rs

@@ -36,11 +36,9 @@ impl PlaybackFuture {
         }
 
         let (mut fallback_stream, fallback) = settings.stream("fallback-uri").optional().await?;
-        self.internal
-            .write()
-            .await
-            .update_fallback_items(fallback)
-            .await;
+
+        let configure_fallback = Fuse::new(update_fallback_items_task(&self.internal, fallback));
+        tokio::pin!(configure_fallback);
 
         let (mut song_stream, song) = injector.stream::<Song>().await;
 
@@ -63,9 +61,6 @@ impl PlaybackFuture {
                         _ => None,
                     }).unwrap_or_default());
                 }
-                fallback = fallback_stream.recv() => {
-                    self.internal.write().await.update_fallback_items(fallback).await;
-                }
                 /* player */
                 _ = &mut song_timeout => {
                     self.internal.write().await.end_of_track().await?;
@@ -89,7 +84,36 @@ impl PlaybackFuture {
                 event = self.connect_stream.recv() => {
                     self.internal.write().await.handle_player_event(event).await?;
                 }
+                fallback = fallback_stream.recv() => {
+                    configure_fallback.set(Fuse::new(update_fallback_items_task(&self.internal, fallback)));
+                }
+                _ = configure_fallback.as_mut() => {
+                }
             }
         }
+
+        /// Update fallback item tasks.
+        async fn update_fallback_items_task(
+            internal: &RwLock<PlayerInternal>,
+            fallback: Option<Uri>,
+        ) {
+            let task = retry_until_ok! {
+                "Loading fallback items", {
+                    let task = internal.read().await.load_fallback_items(fallback.as_ref());
+                    let (what, items) = task.await?;
+
+                    log::info!(
+                        "Updated fallback queue with {} items from {}.",
+                        items.len(),
+                        what
+                    );
+
+                    internal.write().await.update_fallback_items(items).await;
+                    Ok(())
+                }
+            };
+
+            task.await
+        }
     }
 }