//! S3 backend: AWS or any S3-compatible store (Linode Object Storage,
//! MinIO, Backblaze B2, DigitalOcean Spaces, ...).
//!
//! URL form: `s3://bucket/prefix?region=...&endpoint=...&style=...`
//! where every parameter is optional:
//!
//! - `region`: falls back to `$AWS_REGION`. For non-AWS stores this is
//!   whatever region string the provider signs with.
//! - `endpoint`: falls back to `$AWS_ENDPOINT_URL`; omitted entirely
//!   means AWS proper. `https://` is assumed if no scheme is given.
//! - `style`: `path` or `vhost` addressing. Defaults to `vhost` for
//!   AWS and `path` for custom endpoints, which is the combination
//!   that works nearly everywhere; providers that require
//!   virtual-hosted buckets on custom endpoints can say `style=vhost`.
//!
//! Credentials come from the standard AWS sources (environment
//! variables, profile files, instance metadata) — the lingua franca
//! that S3-compatible providers document as well.
//!
//! Examples:
//!
//! ```text
//! s3://backups/laptop?region=eu-central-1
//! s3://backups/laptop?endpoint=eu-central-1.linodeobjects.com&region=eu-central-1
//! s3://backups/laptop?endpoint=http://localhost:9000&region=minio
//! ```
//!
//! No connection pool is needed: each operation is an independent HTTP
//! request and the client is safe to share across threads. S3 `put` is
//! atomic and last-writer-wins by nature; since objects are write-once
//! and content-addressed, overwriting is idempotent and no existence
//! pre-check or rename dance is required.

use repository::{Backend, ChunkId, Error, ObjectKey, ObjectKind};
use s3::{Bucket, Region, creds::Credentials, error::S3Error};
use url::Url;

/// How the bucket name is carried in requests: as a URL path segment or
/// as part of the host name.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum AddressingStyle {
    Path,
    VirtualHost,
}

pub struct S3Backend {
    bucket: Box<Bucket>,
    prefix: String,
}

impl S3Backend {
    pub fn from_url(url: &Url) -> Result<S3Backend, Error> {
        let name = url
            .host_str()
            .ok_or_else(|| Error::Backend(format!("{url} has no bucket name")))?
            .to_string();

        let mut region = std::env::var("AWS_REGION").ok();
        let mut endpoint = std::env::var("AWS_ENDPOINT_URL").ok();
        let mut style = None;

        for (key, value) in url.query_pairs() {
            match key.as_ref() {
                "region" => region = Some(value.into_owned()),
                "endpoint" => endpoint = Some(value.into_owned()),
                "style" => match value.as_ref() {
                    "path" => style = Some(AddressingStyle::Path),
                    "vhost" => style = Some(AddressingStyle::VirtualHost),
                    other => {
                        return Err(Error::Backend(format!(
                            "s3: style must be \"path\" or \"vhost\", not {other:?}"
                        )));
                    }
                },
                other => {
                    return Err(Error::Backend(format!(
                        "s3: unknown URL parameter {other:?}"
                    )));
                }
            }
        }

        let custom_endpoint = endpoint.is_some();

        let region = match endpoint {
            Some(endpoint) => Region::Custom {
                region: region.unwrap_or_else(|| "us-east-1".to_string()),
                endpoint: ensure_scheme(endpoint),
            },
            None => region
                .ok_or_else(|| {
                    Error::Backend(
                        "s3: no region: set $AWS_REGION or add ?region= to the URL".to_string(),
                    )
                })?
                .parse()
                .map_err(|err| Error::Backend(format!("s3: bad region: {err}")))?,
        };

        // Path-style is the near-universal default for S3-compatible
        // stores; AWS itself prefers virtual-hosted buckets
        let style = style.unwrap_or(if custom_endpoint {
            AddressingStyle::Path
        } else {
            AddressingStyle::VirtualHost
        });

        let credentials = Credentials::default()
            .map_err(|err| Error::Backend(format!("s3: credentials: {err}")))?;

        let mut bucket = Bucket::new(&name, region, credentials).map_err(s3_error)?;

        if style == AddressingStyle::Path {
            bucket = bucket.with_path_style();
        }

        Ok(S3Backend {
            bucket,
            prefix: url.path().trim_matches('/').to_string(),
        })
    }

    fn object_path(&self, key: &ObjectKey) -> String {
        let mut path = self.prefix.clone();

        for segment in key.segments() {
            if !path.is_empty() {
                path.push('/');
            }

            path.push_str(&segment);
        }

        path
    }

    fn prefix_path(&self, segments: &[&str]) -> String {
        let mut path = self.prefix.clone();

        for segment in segments {
            if !path.is_empty() {
                path.push('/');
            }

            path.push_str(segment);
        }

        path
    }
}

impl Backend for S3Backend {
    fn put(&self, key: &ObjectKey, data: &[u8]) -> Result<(), Error> {
        let response = self
            .bucket
            .put_object(self.object_path(key), data)
            .map_err(s3_error)?;

        if (200..300).contains(&response.status_code()) {
            Ok(())
        } else {
            Err(Error::Backend(format!(
                "s3: put returned status {}",
                response.status_code()
            )))
        }
    }

    fn get(&self, key: &ObjectKey) -> Result<Option<Vec<u8>>, Error> {
        match self.bucket.get_object(self.object_path(key)) {
            Ok(response) if response.status_code() == 404 => Ok(None),
            Ok(response) if (200..300).contains(&response.status_code()) => {
                Ok(Some(response.into_bytes().to_vec()))
            }
            Ok(response) => Err(Error::Backend(format!(
                "s3: get returned status {}",
                response.status_code()
            ))),
            Err(err) if is_not_found(&err) => Ok(None),
            Err(err) => Err(s3_error(err)),
        }
    }

    fn contains(&self, key: &ObjectKey) -> Result<bool, Error> {
        match self.bucket.head_object(self.object_path(key)) {
            Ok((_, 404)) => Ok(false),
            Ok((_, code)) if (200..300).contains(&code) => Ok(true),
            Ok((_, code)) => Err(Error::Backend(format!("s3: head returned status {code}"))),
            Err(err) if is_not_found(&err) => Ok(false),
            Err(err) => Err(s3_error(err)),
        }
    }

    fn list(
        &self,
        kind: ObjectKind,
        visit: &mut dyn FnMut(ObjectKey) -> Result<(), Error>,
    ) -> Result<(), Error> {
        match kind {
            ObjectKind::Header => {
                if self.contains(&ObjectKey::Header)? {
                    visit(ObjectKey::Header)?;
                }
            }
            // The flat kinds: one prefix of hex-named objects each
            ObjectKind::Snapshot | ObjectKind::Prune | ObjectKind::Lock => {
                self.visit_names(&self.prefix_path(&[kind.directory()]), &mut |name| {
                    if let Some(key) = kind.key_for(name) {
                        return visit(key);
                    }

                    Ok(())
                })?;
            }
            ObjectKind::Chunk => {
                self.visit_names(&self.prefix_path(&["chunks"]), &mut |name| {
                    if let Ok(id) = ChunkId::from_hex(name) {
                        return visit(ObjectKey::Chunk(id));
                    }

                    Ok(())
                })?;
            }
        }

        Ok(())
    }

    /// Adds up every object under the repository prefix, a thousand
    /// keys to a request.
    ///
    /// An object store has no size of its own to speak of, but a
    /// backup told to keep its repository under a certain size is
    /// asking about the repository, and that is answerable.
    fn used_space(&self) -> Result<Option<u64>, Error> {
        // A repository at the bucket root has no prefix to append a
        // separator to, and "/" would match nothing
        let prefix = match self.prefix.is_empty() {
            true => String::new(),
            false => format!("{}/", self.prefix),
        };

        let pages = self.bucket.list(prefix, None).map_err(s3_error)?;

        let total = pages
            .iter()
            .flat_map(|page| page.contents.iter())
            .map(|object| object.size)
            .sum();

        Ok(Some(total))
    }

    fn delete(&self, key: &ObjectKey) -> Result<(), Error> {
        match self.bucket.delete_object(self.object_path(key)) {
            Ok(_) => Ok(()),
            Err(err) if is_not_found(&err) => Ok(()),
            Err(err) => Err(s3_error(err)),
        }
    }
}

impl S3Backend {
    /// Visits the final name component of every object under `prefix`.
    /// S3 listing is flat, so chunk fan-out directories need no special
    /// handling here.
    fn visit_names(
        &self,
        prefix: &str,
        visit: &mut dyn FnMut(&str) -> Result<(), Error>,
    ) -> Result<(), Error> {
        let pages = self
            .bucket
            .list(format!("{prefix}/"), None)
            .map_err(s3_error)?;

        for page in pages {
            for object in page.contents {
                let name = object
                    .key
                    .rsplit('/')
                    .next()
                    .expect("rsplit always yields at least one part");

                visit(name)?;
            }
        }

        Ok(())
    }
}

/// Endpoints are commonly written without a scheme ("HTTPS is implied");
/// the client library wants a full URL.
fn ensure_scheme(endpoint: String) -> String {
    if endpoint.contains("://") {
        endpoint
    } else {
        format!("https://{endpoint}")
    }
}

fn is_not_found(err: &S3Error) -> bool {
    matches!(err, S3Error::HttpFailWithBody(404, _))
}

fn s3_error(err: S3Error) -> Error {
    Error::Backend(format!("s3: {err}"))
}