librespot/core/src/spclient.rs

use std::time::Duration;

use bytes::Bytes;
use futures_util::future::IntoStream;
use http::header::HeaderValue;
use hyper::{
    client::ResponseFuture,
    header::{ACCEPT, AUTHORIZATION, CONTENT_TYPE, RANGE},
    Body, HeaderMap, Method, Request,
};
use protobuf::Message;
use rand::Rng;
use thiserror::Error;

use crate::{
    apresolve::SocketAddress,
    cdn_url::CdnUrl,
    error::ErrorKind,
    protocol::{
        canvaz::EntityCanvazRequest, connect::PutStateRequest,
        extended_metadata::BatchedEntityRequest,
    },
    Error, FileId, SpotifyId,
};

component! {
    SpClient : SpClientInner {
        accesspoint: Option<SocketAddress> = None,
        strategy: RequestStrategy = RequestStrategy::default(),
    }
}

pub type SpClientResult = Result<Bytes, Error>;

#[derive(Debug, Error)]
pub enum SpClientError {
    #[error("missing attribute {0}")]
    Attribute(String),
}

impl From<SpClientError> for Error {
    fn from(err: SpClientError) -> Self {
        Self::failed_precondition(err)
    }
}

#[derive(Copy, Clone, Debug)]
pub enum RequestStrategy {
    TryTimes(usize),
    Infinitely,
}

impl Default for RequestStrategy {
    fn default() -> Self {
        RequestStrategy::TryTimes(10)
    }
}

impl SpClient {
    pub fn set_strategy(&self, strategy: RequestStrategy) {
        self.lock(|inner| inner.strategy = strategy)
    }

    pub async fn flush_accesspoint(&self) {
        self.lock(|inner| inner.accesspoint = None)
    }

    pub async fn get_accesspoint(&self) -> Result<SocketAddress, Error> {
        // Memoize the current access point.
        let ap = self.lock(|inner| inner.accesspoint.clone());
        let tuple = match ap {
            Some(tuple) => tuple,
            None => {
                let tuple = self.session().apresolver().resolve("spclient").await?;
                self.lock(|inner| inner.accesspoint = Some(tuple.clone()));
                info!(
                    "Resolved \"{}:{}\" as spclient access point",
                    tuple.0, tuple.1
                );
                tuple
            }
        };
        Ok(tuple)
    }

    pub async fn base_url(&self) -> Result<String, Error> {
        let ap = self.get_accesspoint().await?;
        Ok(format!("https://{}:{}", ap.0, ap.1))
    }

    pub async fn request_with_protobuf(
        &self,
        method: &Method,
        endpoint: &str,
        headers: Option<HeaderMap>,
        message: &dyn Message,
    ) -> SpClientResult {
        let body = protobuf::text_format::print_to_string(message);

        let mut headers = headers.unwrap_or_else(HeaderMap::new);
        headers.insert(
            CONTENT_TYPE,
            HeaderValue::from_static("application/protobuf"),
        );

        self.request(method, endpoint, Some(headers), Some(body))
            .await
    }

    pub async fn request_as_json(
        &self,
        method: &Method,
        endpoint: &str,
        headers: Option<HeaderMap>,
        body: Option<String>,
    ) -> SpClientResult {
        let mut headers = headers.unwrap_or_else(HeaderMap::new);
        headers.insert(ACCEPT, HeaderValue::from_static("application/json"));

        self.request(method, endpoint, Some(headers), body).await
    }

    pub async fn request(
        &self,
        method: &Method,
        endpoint: &str,
        headers: Option<HeaderMap>,
        body: Option<String>,
    ) -> SpClientResult {
        let mut tries: usize = 0;
        let mut last_response;

        let body = body.unwrap_or_else(String::new);

        loop {
            tries += 1;

            // Reconnection logic: retrieve the endpoint every iteration, so we can try
            // another access point when we are experiencing network issues (see below).
            let mut url = self.base_url().await?;
            url.push_str(endpoint);

            // Add metrics. There is also an optional `partner` key with a value like
            // `vodafone-uk` but we've yet to discover how we can find that value.
            let separator = match url.find('?') {
                Some(_) => "&",
                None => "?",
            };
            url.push_str(&format!("{}product=0", separator));

            let mut request = Request::builder()
                .method(method)
                .uri(url)
                .body(Body::from(body.clone()))?;

            // Reconnection logic: keep getting (cached) tokens because they might have expired.
            let headers_mut = request.headers_mut();
            if let Some(ref hdrs) = headers {
                *headers_mut = hdrs.clone();
            }
            headers_mut.insert(
                AUTHORIZATION,
                HeaderValue::from_str(&format!(
                    "Bearer {}",
                    self.session()
                        .token_provider()
                        .get_token("playlist-read")
                        .await?
                        .access_token
                ))?,
            );

            last_response = self.session().http_client().request_body(request).await;

            if last_response.is_ok() {
                return last_response;
            }

            // Break before the reconnection logic below, so that the current access point
            // is retained when max_tries == 1. Leave it up to the caller when to flush.
            if let RequestStrategy::TryTimes(max_tries) = self.lock(|inner| inner.strategy) {
                if tries >= max_tries {
                    break;
                }
            }

            // Reconnection logic: drop the current access point if we are experiencing issues.
            // This will cause the next call to base_url() to resolve a new one.
            if let Err(ref network_error) = last_response {
                match network_error.kind {
                    ErrorKind::Unavailable | ErrorKind::DeadlineExceeded => {
                        // Keep trying the current access point three times before dropping it.
                        if tries % 3 == 0 {
                            self.flush_accesspoint().await
                        }
                    }
                    _ => break, // if we can't build the request now, then we won't ever
                }
            }

            // When retrying, avoid hammering the Spotify infrastructure by sleeping a while.
            // The backoff time is chosen randomly from an ever-increasing range.
            let max_seconds = u64::pow(tries as u64, 2) * 3;
            let backoff = Duration::from_secs(rand::thread_rng().gen_range(1..=max_seconds));
            warn!(
                "Unable to complete API request, waiting {} seconds before retrying...",
                backoff.as_secs(),
            );
            debug!("Error was: {:?}", last_response);
            tokio::time::sleep(backoff).await;
        }

        last_response
    }

    pub async fn put_connect_state(
        &self,
        connection_id: String,
        state: PutStateRequest,
    ) -> SpClientResult {
        let endpoint = format!("/connect-state/v1/devices/{}", self.session().device_id());

        let mut headers = HeaderMap::new();
        headers.insert("X-Spotify-Connection-Id", connection_id.parse()?);

        self.request_with_protobuf(&Method::PUT, &endpoint, Some(headers), &state)
            .await
    }

    pub async fn get_metadata(&self, scope: &str, id: SpotifyId) -> SpClientResult {
        let endpoint = format!("/metadata/4/{}/{}", scope, id.to_base16());
        self.request(&Method::GET, &endpoint, None, None).await
    }

    pub async fn get_track_metadata(&self, track_id: SpotifyId) -> SpClientResult {
        self.get_metadata("track", track_id).await
    }

    pub async fn get_episode_metadata(&self, episode_id: SpotifyId) -> SpClientResult {
        self.get_metadata("episode", episode_id).await
    }

    pub async fn get_album_metadata(&self, album_id: SpotifyId) -> SpClientResult {
        self.get_metadata("album", album_id).await
    }

    pub async fn get_artist_metadata(&self, artist_id: SpotifyId) -> SpClientResult {
        self.get_metadata("artist", artist_id).await
    }

    pub async fn get_show_metadata(&self, show_id: SpotifyId) -> SpClientResult {
        self.get_metadata("show", show_id).await
    }

    pub async fn get_lyrics(&self, track_id: SpotifyId) -> SpClientResult {
        let endpoint = format!("/color-lyrics/v1/track/{}", track_id.to_base62());

        self.request_as_json(&Method::GET, &endpoint, None, None)
            .await
    }

    pub async fn get_lyrics_for_image(
        &self,
        track_id: SpotifyId,
        image_id: FileId,
    ) -> SpClientResult {
        let endpoint = format!(
            "/color-lyrics/v2/track/{}/image/spotify:image:{}",
            track_id.to_base62(),
            image_id
        );

        self.request_as_json(&Method::GET, &endpoint, None, None)
            .await
    }

    // TODO: Find endpoint for newer canvas.proto and upgrade to that.
    pub async fn get_canvases(&self, request: EntityCanvazRequest) -> SpClientResult {
        let endpoint = "/canvaz-cache/v0/canvases";
        self.request_with_protobuf(&Method::POST, endpoint, None, &request)
            .await
    }

    pub async fn get_extended_metadata(&self, request: BatchedEntityRequest) -> SpClientResult {
        let endpoint = "/extended-metadata/v0/extended-metadata";
        self.request_with_protobuf(&Method::POST, endpoint, None, &request)
            .await
    }

    pub async fn get_audio_storage(&self, file_id: FileId) -> SpClientResult {
        let endpoint = format!(
            "/storage-resolve/files/audio/interactive/{}",
            file_id.to_base16()
        );
        self.request(&Method::GET, &endpoint, None, None).await
    }

    pub fn stream_from_cdn(
        &self,
        cdn_url: &CdnUrl,
        offset: usize,
        length: usize,
    ) -> Result<IntoStream<ResponseFuture>, Error> {
        let url = cdn_url.try_get_url()?;
        let req = Request::builder()
            .method(&Method::GET)
            .uri(url)
            .header(
                RANGE,
                HeaderValue::from_str(&format!("bytes={}-{}", offset, offset + length - 1))?,
            )
            .body(Body::empty())?;

        let stream = self.session().http_client().request_stream(req)?;

        Ok(stream)
    }

    pub async fn request_url(&self, url: String) -> SpClientResult {
        let request = Request::builder()
            .method(&Method::GET)
            .uri(url)
            .body(Body::empty())?;

        self.session().http_client().request_body(request).await
    }

    // Audio preview in 96 kbps MP3, unencrypted
    pub async fn get_audio_preview(&self, preview_id: &FileId) -> SpClientResult {
        let attribute = "audio-preview-url-template";
        let template = self
            .session()
            .get_user_attribute(attribute)
            .ok_or_else(|| SpClientError::Attribute(attribute.to_string()))?;

        let mut url = template.replace("{id}", &preview_id.to_base16());
        let separator = match url.find('?') {
            Some(_) => "&",
            None => "?",
        };
        url.push_str(&format!("{}cid={}", separator, self.session().client_id()));

        self.request_url(url).await
    }

    // The first 128 kB of a track, unencrypted
    pub async fn get_head_file(&self, file_id: FileId) -> SpClientResult {
        let attribute = "head-files-url";
        let template = self
            .session()
            .get_user_attribute(attribute)
            .ok_or_else(|| SpClientError::Attribute(attribute.to_string()))?;

        let url = template.replace("{file_id}", &file_id.to_base16());

        self.request_url(url).await
    }

    pub async fn get_image(&self, image_id: FileId) -> SpClientResult {
        let attribute = "image-url";
        let template = self
            .session()
            .get_user_attribute(attribute)
            .ok_or_else(|| SpClientError::Attribute(attribute.to_string()))?;
        let url = template.replace("{file_id}", &image_id.to_base16());

        self.request_url(url).await
    }
}