yab2 0.1.0-alpha.3

//! Yet Another Backblaze B2 Client
//! ===============================
//!
//! Opinionated Backblaze B2 Client.
//!
//! ## Features
//!
//! - Simple API making use of Rust's ownership for API constraints
//! - Automatic re-authentication and refreshing of Upload URLs
//!
//! ## Cargo Features
//!
//! - `fs` (enables optimized routine for uploading from filesystem)
//! - `pool` (enabled non-large `UploadURL` object pool for reuse)
//! - `reqwest_compression` (enables deflate/gzip features on `reqwest`)
//! - `large_buffers` (enables large buffer support, 64KiB instead of 8KiB)

#![allow(clippy::redundant_pattern_matching)]

#[macro_use]
extern crate serde;

use std::{borrow::Cow, future::Future, num::NonZeroU32, sync::Arc, time::Duration};

use tokio::sync::RwLock;

use headers::HeaderMapExt;
use reqwest::header::{HeaderMap, HeaderValue, AUTHORIZATION};
use reqwest::Method;

macro_rules! h {
    ($headers:ident.$key:literal => $value:expr) => {
        $headers.insert(
            reqwest::header::HeaderName::from_static($key), // NOTE: Header names must be lowercase
            reqwest::header::HeaderValue::from_str($value).expect("Unable to use header value"),
        );
    };
}

mod types;

pub mod error;
pub mod models;

pub use types::{sse, DownloadFileBy, FileRetention, ListFiles, NewFileInfo, NewLargeFileInfo, NewPartInfo};

/// Autogenerated builders for various types.
pub mod builders {
    pub use crate::types::{
        FileRetentionBuilder, ListFilesBuilder, NewFileInfoBuilder, NewLargeFileInfoBuilder, NewPartInfoBuilder,
    };
}

#[cfg(feature = "pool")]
pub mod pool;

#[cfg(feature = "fs")]
mod fs;

pub use error::B2Error;
pub use fs::NewFileFromPath;

struct ClientState {
    /// The builder used to create the client.
    config: ClientBuilder,

    /// The authorization data returned from the B2 API `b2_authorize_account` endpoint
    account: crate::models::B2Authorized,

    /// The authorization header to use for requests
    auth: HeaderValue,
}

impl ClientState {
    fn check_capability(&self, capability: &'static str) -> Result<(), B2Error> {
        if !self.account.allowed(capability) {
            return Err(B2Error::MissingCapability(capability));
        }

        Ok(())
    }

    fn url(&self, path: &str) -> String {
        format!("{}/b2api/v3/{}", self.account.api.storage.api_url, path)
    }

    #[inline]
    fn bucket_id<'a>(&'a self, bucket_id: Option<&'a str>) -> Result<&'a str, B2Error> {
        #[allow(clippy::unnecessary_lazy_evaluations)]
        bucket_id.or_else(|| self.account.api.storage.bucket_id.as_deref()).ok_or(B2Error::MissingBucketId)
    }

    fn check_prefix(&self, name: Option<&str>) -> Result<(), B2Error> {
        match (name, self.account.api.storage.name_prefix.as_ref()) {
            (Some(name), Some(prefix)) if !name.starts_with(prefix as &str) => Err(B2Error::InvalidPrefix),
            _ => Ok(()),
        }
    }
}

/// A client for interacting with the B2 API
#[derive(Clone)]
pub struct Client {
    state: Arc<RwLock<ClientState>>,
    client: reqwest::Client,
}

/// A builder for creating a [`Client`]
#[derive(Clone)]
pub struct ClientBuilder {
    auth: HeaderValue,
    ua: Option<Cow<'static, str>>,
    max_retries: u8,
    retry_delay: Duration,
}

/// Wrapper around a response and the file's parsed headers.
pub struct DownloadedFile {
    pub resp: reqwest::Response,
    pub info: models::B2FileHeaders,
}

impl ClientBuilder {
    /// Creates a new client builder with the given key ID and application key.
    pub fn new(key_id: &str, app_key: &str) -> ClientBuilder {
        ClientBuilder {
            auth: models::create_auth_header(key_id, app_key),
            ua: None,
            max_retries: 5,
            retry_delay: Duration::from_secs(1),
        }
    }

    /// Sets the `User-Agent` header to be used for requests.
    #[inline]
    pub fn user_agent(mut self, ua: impl Into<Cow<'static, str>>) -> Self {
        self.ua = Some(ua.into());
        self
    }

    /// Sets the maximum number of times to retry requests if they fail with a 401 Unauthorized error.
    #[inline]
    pub fn max_retries(mut self, max_retries: u8) -> Self {
        self.max_retries = max_retries;
        self
    }

    /// Sets the delay between authorization retries if a request fails.
    pub fn retry_delay(mut self, delay: Duration) -> Self {
        self.retry_delay = delay;
        self
    }

    /// Builds and authorizes the client for first use.
    pub async fn authorize(self) -> Result<Client, B2Error> {
        let mut builder = reqwest::ClientBuilder::new().https_only(true);

        if let Some(ref ua) = self.ua {
            builder = builder.user_agent(ua.as_ref());
        }

        let client = builder.build()?;

        Ok(Client {
            state: Arc::new(RwLock::new(Client::do_auth(&client, self).await?)),
            client,
        })
    }
}

struct DummyValue;

impl<'de> serde::Deserialize<'de> for DummyValue {
    fn deserialize<D>(_: D) -> Result<Self, D::Error>
    where
        D: serde::Deserializer<'de>,
    {
        Ok(DummyValue)
    }
}

impl Client {
    fn req(&self, method: Method, auth: &HeaderValue, url: impl AsRef<str>) -> reqwest::RequestBuilder {
        self.client.request(method, url.as_ref()).header(AUTHORIZATION, auth)
    }

    async fn json<T>(builder: reqwest::RequestBuilder) -> Result<T, B2Error>
    where
        T: serde::de::DeserializeOwned,
    {
        let resp = builder.send().await?;

        if !resp.status().is_success() {
            return Err(B2Error::B2ErrorMessage(resp.json().await?));
        }

        Ok(serde_json::from_str(&resp.text().await?)?)
    }

    async fn do_auth(client: &reqwest::Client, config: ClientBuilder) -> Result<ClientState, B2Error> {
        use failsafe::{futures::CircuitBreaker, Config, Error as FailsafeError};

        let cb = Config::new().build();
        let mut attempts = 0;

        'try_auth: loop {
            let do_auth_inner = Client::json::<models::B2Authorized>(
                client
                    .get("https://api.backblazeb2.com/b2api/v3/b2_authorize_account")
                    .header(AUTHORIZATION, &config.auth),
            );

            return match cb.call(do_auth_inner).await {
                Ok(account) => Ok(ClientState {
                    config,
                    auth: HeaderValue::from_str(&account.auth_token)
                        .expect("Unable to use auth token in header value"),
                    account,
                }),
                Err(FailsafeError::Rejected) => {
                    attempts += 1;
                    if attempts >= config.max_retries {
                        return Err(B2Error::Unauthorized);
                    }

                    tokio::time::sleep(config.retry_delay).await;

                    continue 'try_auth;
                }
                Err(FailsafeError::Inner(e)) => Err(e),
            };
        }
    }

    /// Reauthorizes the client, updating the authorization token and account information.
    async fn reauthorize(&self) -> Result<(), B2Error> {
        let new_state = Self::do_auth(&self.client, self.state.read().await.config.clone()).await?;
        *self.state.write().await = new_state;
        Ok(())
    }

    /// Runs a request, reauthorizing if necessary.
    async fn run_request_with_reauth<'a, F, R, T>(&self, f: F) -> Result<T, B2Error>
    where
        F: Fn(Self) -> R + 'a,
        R: Future<Output = Result<T, B2Error>> + 'a,
    {
        let mut retried = false;
        loop {
            return match f(self.clone()).await {
                Ok(t) => Ok(t),
                Err(B2Error::B2ErrorMessage(e)) if !retried && e.status == 401 => {
                    // box future to avoid stack bloat
                    Box::pin(self.reauthorize()).await?;

                    retried = true;
                    continue;
                }
                Err(e) => Err(e),
            };
        }
    }

    /// Uses the `b2_get_file_info` API to get information about a file by its ID.
    pub async fn get_file_info(&self, file_id: &str) -> Result<models::B2FileInfo, B2Error> {
        #[derive(Serialize)]
        #[serde(rename_all = "camelCase")]
        struct B2GetFileInfo<'a> {
            file_id: &'a str,
        }

        self.run_request_with_reauth(|b2| async move {
            let state = b2.state.read().await;

            state.check_capability("readFiles")?; // TODO: check if this is the right capability

            Client::json(b2.req(Method::GET, &state.auth, "b2_get_file_info").query(&B2GetFileInfo { file_id }))
                .await
        })
        .await
    }

    /// Downloads a file by its ID or name, returning a [`DownloadedFile`],
    /// which is a wrapper around a [`reqwest::Response`] and the file's parsed headers.
    ///
    /// The `file` parameter can be either a file ID or a file name.
    /// The `range` parameter can be used to download only a portion of the file. If `None`, the entire file will be downloaded.
    /// The `encryption` parameter is only required if the file is encrypted with server-side encryption with a customer-provided key (SSE-C).
    pub async fn download_file(
        &self,
        file: DownloadFileBy<'_>,
        range: Option<headers::Range>,
        encryption: Option<sse::ServerSideEncryptionCustomer>,
    ) -> Result<DownloadedFile, B2Error> {
        let (range, encryption) = (&range, &encryption);

        // serde_urlencoded doesn't support top-level enums,
        // so we need to use a wrapper struct
        #[derive(Serialize)]
        struct DownloadFileBy2<'a> {
            #[serde(flatten)]
            file: DownloadFileBy<'a>,
        }

        self.run_request_with_reauth(|b2| async move {
            let state = b2.state.read().await;

            state.check_capability("readFiles")?;

            let resp = b2
                .req(Method::GET, &state.auth, {
                    state.url(match file {
                        DownloadFileBy::FileId(_) => "b2_download_file_by_id",
                        DownloadFileBy::FileName(_) => "b2_download_file_by_name",
                    })
                })
                .headers({
                    let mut headers = HeaderMap::new();
                    if let Some(ref range) = range {
                        headers.typed_insert(range.clone());
                    }
                    if let Some(ref encryption) = encryption {
                        encryption.add_headers(&mut headers);
                    }
                    headers
                })
                .query(&DownloadFileBy2 { file })
                .send()
                .await?;

            Ok(DownloadedFile {
                info: models::B2FileHeaders::parse(resp.headers())?,
                resp,
            })
        })
        .await
    }

    /// Lists the names of all files in a bucket, optionally filtered by a prefix and/or delimiter.
    ///
    /// If [`ListFiles::bucket_id`] is `None`, the client's default bucket will be used. If there is no default bucket,
    /// an error will be returned.
    ///
    /// Each time you call, it returns a `nextFileName` and `nextFileId` (only if `all_versions` is true)
    /// that can be used as the starting point for the next call.
    ///
    /// NOTE: `b2_list_file_names`/`b2_list_file_versions` are Class C transactions. The maximum number of
    /// files returned per transaction is 1000. If you set maxFileCount to more than 1000 and
    /// more than 1000 are returned, the call will be billed as multiple transactions, as if you
    /// had made requests in a loop asking for 1000 at a time. For example: if you set maxFileCount
    /// to 10000 and 3123 items are returned, you will be billed for 4 Class C transactions.
    ///
    /// See the [B2 API documentation](https://www.backblaze.com/apidocs/b2-list-file-names) of this method
    /// for more information on how to use the parameters such as `prefix` and `delimiter`.
    ///
    /// # Example
    ///
    /// ```ignore
    /// let client = ClientBuilder::new(&app_id, &app_key).authorize().await?;
    ///
    /// let files = client.list_files(ListFiles::builder().all_versions(true).build()).await?;
    ///
    /// println!("{:#?}", files);
    /// ```
    pub async fn list_files(&self, mut args: ListFiles<'_>) -> Result<models::B2FileInfoList, B2Error> {
        if !args.all_versions {
            args.start_file_id = None; // not used
        }

        self.run_request_with_reauth(move |b2| async move {
            let state = b2.state.read().await;

            state.check_capability("listFiles")?;

            let mut args = ListFiles { ..args }; // redefine lifetime of `args`

            args.bucket_id = Some(state.bucket_id(args.bucket_id)?);

            let path = if args.all_versions { "b2_list_file_versions" } else { "b2_list_file_names" };

            Client::json(b2.req(Method::GET, &state.auth, state.url(path)).query(&args)).await
        })
        .await
    }

    /// Hides a file so that downloading by name will not find the file,
    /// but previous versions of the file are still stored.
    pub async fn hide_file(
        &self,
        bucket_id: Option<&str>,
        file_name: &str,
    ) -> Result<models::B2FileInfo, B2Error> {
        #[derive(Serialize)]
        #[serde(rename_all = "camelCase")]
        struct B2HideFile<'a> {
            #[serde(skip_serializing_if = "Option::is_none")]
            bucket_id: Option<&'a str>,
            file_name: &'a str,
        }

        self.run_request_with_reauth(|b2| async move {
            let state = b2.state.read().await;

            state.check_capability("writeFiles")?;
            state.check_prefix(Some(file_name))?;

            let body = B2HideFile {
                bucket_id: state.bucket_id(bucket_id).ok(),
                file_name,
            };

            Self::json(b2.req(Method::POST, &state.auth, state.url("b2_hide_file")).json(&body)).await
        })
        .await
    }

    /// Deletes one version of a file.
    ///
    /// If the version you delete is the latest version, and there are older versions,
    /// then the most recent older version will become the current version,
    /// and be the one that you'll get when downloading by name.
    ///
    /// When used on an unfinished large file, this call has the same effect as cancelling it.
    ///
    /// `bypass_governance` must be set to true if deleting a file version protected by Object Lock
    /// governance mode retention settings. Setting the value requires the
    /// `bypassGovernance` application key capability.
    pub async fn delete_file(
        &self,
        file_id: &str,
        file_name: &str,
        bypass_governance: bool,
    ) -> Result<(), B2Error> {
        #[derive(Serialize)]
        #[serde(rename_all = "camelCase")]
        struct B2DeleteFile<'a> {
            file_id: &'a str,
            file_name: &'a str,
            bypass_governance: bool,
        }

        self.run_request_with_reauth(|b2| async move {
            let state = b2.state.read().await;

            state.check_capability("deleteFiles")?;
            state.check_prefix(Some(file_name))?;

            if bypass_governance {
                // TODO: check if this is the right capability
                state.check_capability("bypassGovernance")?;
            }

            let body = B2DeleteFile {
                file_id,
                file_name,
                bypass_governance,
            };

            Self::json(b2.req(Method::POST, &state.auth, state.url("b2_delete_file_version")).json(&body))
                .await
                .map(|_: DummyValue| ())
        })
        .await
    }

    /// Modifies the Object Lock legal hold status for an existing file.
    ///
    /// Used to enable legal hold for a file in an Object Lock-enabled bucket,
    /// preventing it from being deleted, or to disable legal hold protections for a file.
    ///
    /// Backblaze B2 Cloud Storage Object Lock lets you make data immutable by preventing
    /// a file from being changed or deleted until a given date to protect data that is
    /// stored in Backblaze B2 from threats like ransomware or for regulatory compliance.
    ///
    /// `legal_hold` must be set to `true` to enable legal hold, and `false` to disable it.
    pub async fn update_legal_hold(
        &self,
        file_name: &str,
        file_id: &str,
        legal_hold: bool,
    ) -> Result<(), B2Error> {
        #[derive(Serialize)]
        #[serde(rename_all = "camelCase")]
        struct B2UpdateLegalHold<'a> {
            file_name: &'a str,
            file_id: &'a str,
            legal_hold: &'a str,
        }

        self.run_request_with_reauth(|b2| async move {
            let state = b2.state.read().await;

            state.check_capability("writeFileLegalHolds")?;
            state.check_prefix(Some(file_name))?;

            let body = B2UpdateLegalHold {
                file_name,
                file_id,
                legal_hold: if legal_hold { "on" } else { "off" },
            };

            Self::json(b2.req(Method::POST, &state.auth, state.url("b2_update_legal_hold")).json(&body))
                .await
                .map(|_: DummyValue| ())
        })
        .await
    }

    /// Modifies the Object Lock retention settings for an existing file.
    ///
    /// After enabling file retention for a file in an Object Lock-enabled bucket,
    /// any attempts to delete the file or make any changes to it before
    /// the end of the retention period will fail.
    ///
    /// File retention settings can be configured in either governance or
    /// compliance mode. For files protected in governance mode, file retention
    /// settings can be deleted or the retention period shortened only by clients
    /// with the appropriate application key capability (i.e., bypassGovernance).
    ///
    /// File retention settings for files protected in compliance mode cannot
    /// removed by any user, but their retention dates can be extended by
    /// clients with appropriate application key capabilities.
    pub async fn update_file_retention(
        &self,
        file_name: &str,
        file_id: &str,
        retention: FileRetention,
    ) -> Result<(), B2Error> {
        #[derive(Serialize)]
        #[serde(rename_all = "camelCase")]
        struct B2UpdateFileRetention<'a> {
            file_name: &'a str,
            file_id: &'a str,

            #[serde(flatten)]
            retention: FileRetention,

            bypass_governance: bool,
        }

        let body = &B2UpdateFileRetention {
            file_name,
            file_id,
            bypass_governance: retention.bypass_governance,
            retention,
        };

        self.run_request_with_reauth(|b2| async move {
            let state = b2.state.read().await;

            state.check_capability("writeFileRetentions")?;
            state.check_prefix(Some(file_name))?;

            Self::json(b2.req(Method::POST, &state.auth, state.url("b2_update_file_retention")).json(body))
                .await
                .map(|_: DummyValue| ())
        })
        .await
    }

    async fn get_b2_upload_url(
        &self,
        bucket_id: Option<&str>,
        file_id: Option<&str>,
    ) -> Result<(Option<Arc<str>>, models::B2UploadUrl), B2Error> {
        #[derive(Serialize)]
        #[serde(rename_all = "camelCase")]
        struct B2GetUploadUrlQuery<'a> {
            #[serde(skip_serializing_if = "Option::is_none")]
            bucket_id: Option<&'a str>,

            #[serde(skip_serializing_if = "Option::is_none")]
            file_id: Option<&'a str>,
        }

        self.run_request_with_reauth(|b2| async move {
            let state = b2.state.read().await;

            state.check_capability("writeFiles")?;

            let mut query = B2GetUploadUrlQuery { bucket_id, file_id };

            if query.file_id.is_some() {
                query.bucket_id = None;
            } else if query.bucket_id.is_some() {
                query.file_id = None;
            } else {
                query.bucket_id = Some(state.bucket_id(query.bucket_id)?);
            }

            let path = state.url(if file_id.is_some() { "b2_get_upload_part_url" } else { "b2_get_upload_url" });

            Ok((
                state.account.api.storage.name_prefix.clone(),
                Self::json::<models::B2UploadUrl>(b2.req(Method::GET, &state.auth, path).query(&query)).await?,
            ))
        })
        .await
    }

    async fn get_raw_upload_url(
        &self,
        bucket_id: Option<&str>,
        file_id: Option<&str>,
    ) -> Result<RawUploadUrl, B2Error> {
        let (prefix, url) = self.get_b2_upload_url(bucket_id, file_id).await?;

        Ok(RawUploadUrl {
            client: self.clone(),
            auth: url.header(),
            url,
            prefix,
        })
    }

    /// Gets a URL for uploading files using the `b2_get_upload_url` API.
    ///
    /// If `bucket_id` is `None`, the client's default bucket will be used. If there is no default bucket, an error will be returned.
    ///
    /// The returned `UploadUrl` can be used to upload files to the B2 API for 24 hours. Only one file can be uploaded to a URL at a time.
    /// You may acquire multiple URLs to upload multiple files in parallel.
    pub async fn get_upload_url(&self, bucket_id: Option<&str>) -> Result<UploadUrl, B2Error> {
        Ok(UploadUrl(self.get_raw_upload_url(bucket_id, None).await?))
    }

    /// Gets a URL for uploading parts of a large file using the `b2_get_upload_part_url` API.
    ///
    /// The returned `UploadPartUrl` can be used to upload parts of a large file to the B2 API for 24 hours.
    /// Only one part can be uploaded to a URL at a time. You may acquire multiple URLs to upload multiple parts in parallel.
    pub async fn get_upload_part_url(&self, file_id: &str) -> Result<UploadPartUrl, B2Error> {
        Ok(UploadPartUrl(self.get_raw_upload_url(None, Some(file_id)).await?))
    }

    /// Prepares parts of a large file for uploading using the `b2_start_large_file` API.
    pub async fn start_large_file(
        &self,
        bucket_id: Option<&str>,
        info: &NewLargeFileInfo,
    ) -> Result<LargeFileUpload, B2Error> {
        #[derive(Serialize)]
        #[serde(rename_all = "camelCase")]
        struct B2StartLargeFile<'a> {
            bucket_id: &'a str,
            file_name: &'a str,

            content_type: Option<&'a str>,

            #[serde(skip_serializing_if = "Option::is_none")]
            file_retention: Option<&'a FileRetention>,

            #[serde(skip_serializing_if = "Option::is_none")]
            legal_hold: Option<&'a str>,

            #[serde(skip_serializing_if = "sse::ServerSideEncryption::is_default")]
            encryption: &'a sse::ServerSideEncryption,
        }

        let info = self
            .run_request_with_reauth(|b2| async move {
                let state = b2.state.read().await;

                state.check_capability("writeFiles")?;
                state.check_prefix(Some(&info.file_name))?;

                let body = B2StartLargeFile {
                    bucket_id: state.bucket_id(bucket_id)?,
                    file_name: &info.file_name,
                    content_type: info.content_type.as_deref(),
                    file_retention: info.retention.as_ref(),
                    legal_hold: info.legal_hold.map(|lh| if lh { "on" } else { "off" }),
                    encryption: &info.encryption,
                };

                Client::json::<models::B2FileInfo>(
                    b2.req(Method::POST, &state.auth, state.url("b2_start_large_file")).json(&body),
                )
                .await
            })
            .await?;

        Ok(LargeFileUpload {
            client: self.clone(),
            info,
        })
    }
}

struct RawUploadUrl {
    client: Client,
    url: models::B2UploadUrl,
    auth: HeaderValue,
    prefix: Option<Arc<str>>,
}

/// Temporarily acquired URL for uploading single files.
///
/// This is returned by [`Client::get_upload_url`].
///
/// The URL can be used to upload a file to the B2 API for 24 hours. Only one file can be uploaded to a URL at a time.
/// This is enforced via requiring mutable references to the URL when uploading a file.
#[repr(transparent)]
pub struct UploadUrl(RawUploadUrl);

/// Temporarily acquired URL for uploading parts of a large file.
///
/// This is returned by [`Client::get_upload_part_url`].
///
/// The URL can be used to upload parts of a large file to the B2 API for 24 hours. Only one part can be uploaded to a URL at a time.
/// This is enforced via requiring mutable references to the URL when uploading a part.
#[repr(transparent)]
pub struct UploadPartUrl(RawUploadUrl);

impl RawUploadUrl {
    /// Actually performs the upload, with automatic reauthorization if necessary.
    async fn do_upload<F, T>(&mut self, f: F) -> Result<T, B2Error>
    where
        F: Fn(reqwest::RequestBuilder) -> reqwest::RequestBuilder,
        T: serde::de::DeserializeOwned,
    {
        loop {
            let res = Client::json(f(self.client.req(Method::POST, &self.auth, &self.url.upload_url)));
            return match res.await {
                Err(B2Error::B2ErrorMessage(e)) if e.status == 401 => {
                    let get_new_url =
                        self.client.get_b2_upload_url(self.url.bucket_id.as_deref(), self.url.file_id.as_deref());

                    let (prefix, url) = Box::pin(get_new_url).await?;

                    self.auth = url.header();
                    self.url = url;
                    self.prefix = prefix;

                    continue;
                }
                res => res,
            };
        }
    }

    fn check_prefix(&self, file_name: &str) -> Result<(), B2Error> {
        match self.prefix {
            Some(ref prefix) if !file_name.starts_with(prefix.as_ref()) => Err(B2Error::InvalidPrefix),
            _ => Ok(()),
        }
    }

    async fn upload_file<F, B>(&mut self, info: &NewFileInfo, file: F) -> Result<models::B2FileInfo, B2Error>
    where
        F: Fn() -> B,
        B: Into<reqwest::Body>,
    {
        self.check_prefix(&info.file_name)?;

        self.do_upload(|builder| {
            builder.body(file()).headers({
                let mut headers = HeaderMap::new();
                info.add_headers(&mut headers);
                headers
            })
        })
        .await
    }

    async fn upload_part<F, B>(&mut self, info: &NewPartInfo, body: F) -> Result<models::B2PartInfo, B2Error>
    where
        F: Fn() -> B,
        B: Into<reqwest::Body>,
    {
        self.do_upload(|builder| {
            builder.body(body()).headers({
                let mut headers = HeaderMap::new();
                info.add_headers(&mut headers);
                headers
            })
        })
        .await
    }
}

impl UploadUrl {
    /// Uploads a file to the B2 API using the URL acquired from [`Client::get_upload_url`].
    ///
    /// The `file` parameter is a closure that returns a value to be converted into a `reqwest::Body`.
    /// This method may need to retry the request if the URL or authorization token has expired, therefore
    /// it is recommended the body-creation closure be cheap to call multiple times.
    pub async fn upload_file<F, B>(&mut self, info: &NewFileInfo, file: F) -> Result<models::B2FileInfo, B2Error>
    where
        F: Fn() -> B,
        B: Into<reqwest::Body>,
    {
        self.0.upload_file(info, file).await
    }

    /// Uploads a file to the B2 API using the URL acquired from [`Client::get_upload_url`].
    ///
    /// The `bytes` parameter is a value to be converted into the body of the request.
    pub async fn upload_file_bytes(
        &mut self,
        info: &NewFileInfo,
        bytes: impl Into<bytes::Bytes>,
    ) -> Result<models::B2FileInfo, B2Error> {
        let bytes = bytes.into();
        self.upload_file(info, || bytes.clone()).await
    }
}

/// A large file that is being uploaded in parts.
///
/// Any [`UploadPartUrl`] can be used to upload a part of the file. Once all parts have been uploaded,
/// call [`LargeFileUpload::finish`] to complete the upload.
pub struct LargeFileUpload {
    client: Client,
    info: models::B2FileInfo,
}

impl LargeFileUpload {
    pub fn info(&self) -> &models::B2FileInfo {
        &self.info
    }

    /// Equivalent to [`Client::start_large_file`].
    pub async fn start(
        client: &Client,
        bucket_id: Option<&str>,
        info: &NewLargeFileInfo,
    ) -> Result<LargeFileUpload, B2Error> {
        client.start_large_file(bucket_id, info).await
    }

    /// Gets a URL for uploading a part of the large file.
    ///
    /// Equivalent to [`Client::get_upload_part_url`] with `self.info().file_id`.
    pub async fn get_upload_part_url(&self) -> Result<UploadPartUrl, B2Error> {
        self.client.get_upload_part_url(&self.info.file_id).await
    }

    /// Uploads a part of a large file to the given upload URL. Once all parts have been uploaded,
    /// call [`LargeFileUpload::finish`] to complete the upload.
    ///
    /// Parts can be uploaded in parallel, so long as each url is only used for one part at a time.
    ///
    /// The provided `url` must have been acquired from the same `LargeFileUpload` instance, as they
    /// are specific to the file being uploaded.
    ///
    /// The `part` parameter is a closure that returns a value to be converted into a [`reqwest::Body`], and
    /// may need to be called multiple times if the request needs to be retried. Therefore, it is recommended
    /// the body-creation closure be cheap to call multiple times.
    ///
    /// **NOTE**: This method does not check if the provided SHA1 hash is correct.
    pub async fn upload_part<F, B>(
        &self,
        url: &mut UploadPartUrl,
        info: &NewPartInfo,
        part: F,
    ) -> Result<models::B2PartInfo, B2Error>
    where
        F: Fn() -> B,
        B: Into<reqwest::Body>,
    {
        if url.0.url.file_id.as_deref() != Some(self.info.file_id.as_ref()) {
            return Err(B2Error::FileIdMismatch);
        }

        url.0.upload_part(info, part).await
    }

    /// Uploads a part of a large file to the given upload URL. Once all parts have been uploaded,
    /// call [`LargeFileUpload::finish`] to complete the upload.
    ///
    /// Parts can be uploaded in parallel, so long as each url is only used for one part at a time.
    ///
    /// The provided `url` must have been acquired from the same `LargeFileUpload` instance, as they
    /// are specific to the file being uploaded.
    ///
    /// The `bytes` parameter is a value to be converted into the body of the request.
    ///
    /// **NOTE**: This method does not check if the provided SHA1 hash is correct.
    pub async fn upload_part_bytes(
        &self,
        url: &mut UploadPartUrl,
        info: &NewPartInfo,
        bytes: impl Into<bytes::Bytes>,
    ) -> Result<models::B2PartInfo, B2Error> {
        let bytes = bytes.into();
        self.upload_part(url, info, || bytes.clone()).await
    }

    /// Converts the parts that have been uploaded into a single B2 file.
    ///
    /// It may be that the call to finish a large file succeeds, but you don't know it because the
    /// request timed out, or the connection was broken. In that case, retrying will result in a
    /// 400 Bad Request response because the file is already finished. If that happens, we recommend
    /// calling `b2_get_file_info`/[`Client::get_file_info`] to see if the file is there. If the file is there,
    /// you can count the upload as a success.
    ///
    /// `parts` must be sorted by `part_number`.
    pub async fn finish(self, parts: &[models::B2PartInfo]) -> Result<models::B2FileInfo, B2Error> {
        // check if parts are sorted by part_number
        if parts.windows(2).any(|w| w[0].part_number >= w[1].part_number) {
            return Err(B2Error::InvalidPartSorting);
        }

        #[derive(Serialize)]
        #[serde(rename_all = "camelCase")]
        struct B2FinishLargeFile<'a> {
            file_id: &'a str,
            part_sha1_array: Vec<&'a str>,
        }

        let body = &B2FinishLargeFile {
            file_id: &self.info.file_id,
            part_sha1_array: parts.iter().map(|part| &*part.content_sha1).collect(),
        };

        self.client
            .run_request_with_reauth(|b2| async move {
                let state = b2.state.read().await;

                Client::json(b2.req(Method::POST, &state.auth, state.url("b2_finish_large_file")).json(&body))
                    .await
            })
            .await
    }

    /// Cancels the upload of a large file, and deletes all of the parts that have been uploaded.
    ///
    /// This will return an error if there is no active upload with the given file ID.
    pub async fn cancel(self) -> Result<models::B2CancelledFileInfo, B2Error> {
        #[derive(Serialize)]
        #[serde(rename_all = "camelCase")]
        struct B2CancelLargeFile<'a> {
            file_id: &'a str,
        }

        let body = &B2CancelLargeFile {
            file_id: &self.info.file_id,
        };

        self.client
            .run_request_with_reauth(|b2| async move {
                let state = b2.state.read().await;

                Client::json(b2.req(Method::POST, &state.auth, state.url("b2_cancel_large_file")).json(&body))
                    .await
            })
            .await
    }
}

#[cfg(test)]
mod tests {
    use tokio::io::AsyncReadExt;

    use super::*;

    #[test]
    fn test_downloadby_serialization() {
        let file_id = "4_zc1234567890abcdef1234f1";
        let file_name = "example.txt";

        let file_id_json = serde_json::to_string(&DownloadFileBy::FileId(file_id)).unwrap();
        let file_name_json = serde_json::to_string(&DownloadFileBy::FileName(file_name)).unwrap();

        assert_eq!(file_id_json, format!(r#"{{"fileId":"{}"}}"#, file_id));
        assert_eq!(file_name_json, format!(r#"{{"fileName":"{}"}}"#, file_name));
    }

    #[tokio::test]
    async fn test_auth() {
        use sha1::{Digest, Sha1};

        dotenv::dotenv().ok();

        let app_id = std::env::var("APP_ID").expect("APP_ID not found in .env");
        let app_key = std::env::var("APP_KEY").expect("APP_KEY not found in .env");

        let client = ClientBuilder::new(&app_id, &app_key).authorize().await.unwrap();

        // must be mut because `upload_file` requires exclusive access to the url
        let mut upload = client.get_upload_url(None).await.unwrap();

        let mut file = tokio::fs::OpenOptions::new().read(true).open("Cargo.toml").await.unwrap();
        let meta = file.metadata().await.unwrap();

        let mut bytes = Vec::with_capacity(meta.len() as usize);
        file.read_to_end(&mut bytes).await.unwrap();

        let bytes = bytes::Bytes::from(bytes); // bytes

        let info = NewFileInfo::builder()
            .file_name("testing/Cargo.toml".to_owned())
            .content_length(meta.len())
            .content_type("text/plain".to_owned())
            .content_sha1(hex::encode(Sha1::new().chain_update(&bytes).finalize()))
            .build();

        let file_info = upload.upload_file_bytes(&info, bytes).await.unwrap();

        println!("{:#?}", client.state.read().await.account);

        let resp = client.download_file(DownloadFileBy::FileId(&file_info.file_id), None, None).await.unwrap();

        let text = resp.resp.text().await.unwrap();

        println!("OUTPUT: {text}");
    }

    #[tokio::test]
    async fn test_large_file() {
        dotenv::dotenv().ok();

        let app_id = std::env::var("APP_ID").expect("APP_ID not found in .env");
        let app_key = std::env::var("APP_KEY").expect("APP_KEY not found in .env");

        let client = ClientBuilder::new(&app_id, &app_key).authorize().await.unwrap();

        let info = NewFileFromPath::builder()
            .path(r#"./testing.webm"#.as_ref())
            .content_type("video/webm".to_owned())
            .file_name("testing.webm".to_owned())
            .build();

        let file = client.upload_from_path(info, None, None).await.unwrap();

        println!("{:?}", file);
    }

    #[tokio::test]
    async fn test_small_file() {
        dotenv::dotenv().ok();

        let app_id = std::env::var("APP_ID").expect("APP_ID not found in .env");
        let app_key = std::env::var("APP_KEY").expect("APP_KEY not found in .env");

        let client = ClientBuilder::new(&app_id, &app_key).authorize().await.unwrap();

        let info = NewFileFromPath::builder()
            .path(r#"Cargo.toml"#.as_ref())
            .content_type("test/plain".to_owned())
            .file_name("Cargo.toml".to_owned())
            .build();

        let file = client.upload_from_path(info, None, None).await.unwrap();

        println!("{:?}", file);
    }

    #[tokio::test]
    async fn test_list_files() {
        dotenv::dotenv().ok();

        let app_id = std::env::var("APP_ID").expect("APP_ID not found in .env");
        let app_key = std::env::var("APP_KEY").expect("APP_KEY not found in .env");

        let client = ClientBuilder::new(&app_id, &app_key).authorize().await.unwrap();

        let files = client.list_files(ListFiles::builder().all_versions(false).build()).await.unwrap();

        println!("{:#?}", files);
    }
}