thunderstore 0.6.0

use crate::{Client, Error, IntoPackageIdent, IntoVersionIdent, Result, models::*};

use async_stream::try_stream;
use flate2::read::GzDecoder;
use futures_core::Stream;
use futures_util::{TryStreamExt, pin_mut};
use std::{fmt::Display, io::Read};
use tokio::sync::mpsc;

impl Client {
    /// Fetches [`PackageMetrics`] for a specific package.
    ///
    /// `community` is the slug of the community, which is usually in kebab-case.
    pub async fn get_metrics(
        &self,
        community: impl AsRef<str>,
        package: impl IntoPackageIdent<'_>,
    ) -> Result<PackageMetrics> {
        let url = self.v1_url(
            community,
            format_args!("/package-metrics/{}", package.into_id()?.path()),
        );
        self.get_json(url).await
    }

    /// Fetches the download count for a specific version of a package.
    ///
    /// `community` is the slug of the community, which is usually in kebab-case.
    pub async fn get_downloads(
        &self,
        community: impl AsRef<str>,
        version: impl IntoVersionIdent<'_>,
    ) -> Result<u64> {
        let url = self.v1_url(
            community,
            format_args!("/package-metrics/{}", version.into_id()?.path()),
        );
        let response: PackageVersionMetrics = self.get_json(url).await?;
        Ok(response.downloads)
    }

    /// Fetches all available packages in a community and collects them in a `Vec`.
    ///
    /// - `community` is the slug of the community, which is usually in kebab-case.
    ///
    /// Note that on popular communities like Lethal Company (`lethal-company`),
    /// this will fetch up to 170 MB of data.
    pub async fn list_packages_v1(&self, community: impl AsRef<str>) -> Result<Vec<PackageV1>> {
        let url = self.v1_url(community, "/package/");
        self.get_json(url).await
    }

    fn v1_url(&self, community: impl AsRef<str>, path: impl Display) -> String {
        format!("{}/c/{}/api/v1{}", self.base_url, community.as_ref(), path)
    }

    /// Asynchronously streams all available packages in a community.
    ///
    /// - `community` is the slug of the community, which is usually in kebab-case.
    ///
    /// If you just want a `Vec` of all packages, use [`Client::list_packages_v1`] instead.
    ///
    /// ## Examples
    ///
    /// ```no_run
    /// // provides `try_next`
    /// use futures_util::TryStreamExt;
    /// use futures_util::pin_mut;
    ///
    /// #[tokio::main]
    /// async fn main() -> Result<(), Box<dyn std::error::Error>> {
    ///     let client = thunderstore::Client::new();
    ///
    ///     let stream = client.stream_packages_v1("content-warning").await?;
    ///     pin_mut!(stream); // needed for iteration
    ///
    ///     while let Some(package) = stream.try_next().await? {
    ///         println!("got {}!", package.name);
    ///     }
    ///
    ///    Ok(())
    /// }
    /// ```
    pub async fn stream_packages_v1(
        &self,
        community: impl AsRef<str>,
    ) -> Result<impl Stream<Item = Result<PackageV1>>> {
        let url = self.v1_url(community.as_ref(), "/package/");
        let mut response = self.get(url).await?;

        Ok(try_stream! {
            let mut buffer = Vec::new();
            let mut string = String::new();

            let mut is_first = true;

            while let Some(chunk) = response.chunk().await? {
                buffer.extend_from_slice(&chunk);

                let chunk = match std::str::from_utf8(&buffer) {
                    Ok(chunk) => chunk,
                    Err(_) => continue,
                };

                if is_first {
                    is_first = false;
                    string.extend(chunk.chars().skip(1)); // remove leading [
                } else {
                    string.push_str(chunk);
                }

                buffer.clear();

                while let Some(index) = string.find("}]}") {
                    let (json, _) = string.split_at(index + 3);
                    yield serde_json::from_str::<PackageV1>(json)?;
                    string.replace_range(..index + 4, "");
                }
            }

            debug_assert!(string.is_empty(), "remaining string: {string:?}");
            debug_assert!(buffer.is_empty(), "remaining buffer: {buffer:?}");
        })
    }

    pub async fn stream_package_index_v1<'a>(
        &'a self,
        community: impl AsRef<str>,
    ) -> Result<impl Stream<Item = Result<Vec<PackageV1>>> + 'a> {
        let url = self.v1_url(community, "/package-listing-index/");

        let encoded_urls = self.get(url).await?.bytes().await?;
        let urls: Vec<String> = serde_json::from_reader(GzDecoder::new(&encoded_urls[..]))?;

        let (tx, mut rx) = mpsc::channel(urls.len());

        // need to get static lifetime for the spawned task
        let this = self.clone();

        let handle = tokio::spawn(async move {
            for url in urls {
                let bytes = this.get(url).await?.bytes().await?;
                tx.send(bytes)
                    .await
                    .map_err(|_| Error::ChannelClosedTooEarly)?;
            }
            Ok::<_, Error>(())
        });

        Ok(try_stream! {
            let mut buf = Vec::<u8>::new();

            while let Some(bytes) = rx.recv().await {
                GzDecoder::new(&bytes[..]).read_to_end(&mut buf)?;
                let packages: Vec<PackageV1> = serde_json::from_slice(&buf)?;

                buf.clear();

                yield packages;
            }

            handle.await.expect("request task panicked")?;
        })
    }

    pub async fn get_package_index_v1(&self, community: impl AsRef<str>) -> Result<Vec<PackageV1>> {
        let stream = self.stream_package_index_v1(community).await?;
        pin_mut!(stream);

        let mut packages = Vec::new();
        while let Some(package) = stream.try_next().await? {
            packages.extend(package);
        }

        Ok(packages)
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use futures_util::{TryStreamExt, pin_mut};

    #[tokio::test]
    async fn stream_packages_v1() -> Result<()> {
        let client = Client::new();

        let stream = client.stream_packages_v1("muck").await?;
        pin_mut!(stream);

        let mut count = 0;
        while let Some(_) = stream.try_next().await? {
            count += 1;
        }

        assert!(count > 0);

        Ok(())
    }

    #[tokio::test]
    async fn stream_package_listing_index() -> Result<()> {
        let client = Client::new();
        let stream = client.stream_package_index_v1("muck").await?;
        pin_mut!(stream);

        let mut count = 0;
        while let Some(_) = stream.try_next().await? {
            count += 1;
        }

        assert!(count > 0);

        Ok(())
    }

    #[tokio::test]
    async fn get_package_listing_index() -> Result<()> {
        let client = Client::new();
        let packages = client.get_package_index_v1("muck").await?;

        assert!(packages.len() > 0);

        println!("{packages:#?}");

        Ok(())
    }
}