//! Object storage for mod deliverables.
//!
//! Deliverables are patches and asset archives, far too large for PDS blob
//! storage, so a `mod.version` record carries a URL into this store rather
//! than the bytes themselves.
//!
//! The appview never handles those bytes. It signs a URL the browser uploads
//! to directly, then confirms afterwards that something arrived.

use anyhow::{anyhow, bail, Context, Result};
use s3::bucket::Bucket;
use s3::creds::Credentials;
use s3::region::Region;
use s3::BucketConfiguration;
use tokio::sync::OnceCell;

use crate::atproto::id::{Did, Rkey};
use crate::config::Config;

/// How long a browser has to start an upload before its signed URL expires.
const PRESIGN_EXPIRY_SECS: u32 = 60 * 60;

/// The largest deliverable we accept. Uploads are signed before the bytes
/// exist, so this is checked twice: against the size the browser claims, and
/// against the object that actually shows up.
pub const MAX_DELIVERABLE_BYTES: u64 = 2 * 1024 * 1024 * 1024;

pub struct Store {
    bucket: Box<Bucket>,
    public_base: String,
    /// Set once the bucket is known to exist. A failure leaves it unset, so
    /// the next upload tries again.
    bucket_ready: OnceCell<()>,
    create: BucketCreate,
}

/// What [`Store::ensure_bucket`] needs to make the bucket, kept because
/// `Bucket::create` is an associated function and cannot take the client.
struct BucketCreate {
    name: String,
    region: Region,
    credentials: Credentials,
    virtual_host: bool,
}

impl Store {
    /// Build a client for the configured bucket.
    ///
    /// Talks to nothing: the object store is only needed to publish a
    /// version, so an appview whose bucket is unreachable still serves every
    /// mod page, and download links point straight at the bucket anyway.
    pub fn new(config: &Config) -> Result<Self> {
        let (access_key, secret_key) = config.s3_credentials()?;
        let credentials = Credentials::new(Some(&access_key), Some(&secret_key), None, None, None)
            .context("bad S3 credentials")?;
        let region = Region::Custom {
            region: config.s3_region.clone(),
            endpoint: config.s3_endpoint.trim_end_matches('/').to_string(),
        };

        let bucket = Bucket::new(&config.s3_bucket, region.clone(), credentials.clone())?;
        let bucket = if config.s3_virtual_host {
            bucket
        } else {
            bucket.with_path_style()
        };

        Ok(Self {
            bucket,
            public_base: config.s3_public_base().to_string(),
            bucket_ready: OnceCell::new(),
            create: BucketCreate {
                name: config.s3_bucket.clone(),
                region,
                credentials,
                virtual_host: config.s3_virtual_host,
            },
        })
    }

    /// Create the bucket if it is not there yet, once per process.
    ///
    /// Retried rather than done only at startup, so the appview can come up
    /// before the object store does - which in the devshell it usually does.
    pub async fn ensure_bucket(&self) -> Result<()> {
        self.bucket_ready
            .get_or_try_init(|| async {
                let endpoint = match &self.create.region {
                    Region::Custom { endpoint, .. } => endpoint.as_str(),
                    _ => "the object store",
                };
                if self
                    .bucket
                    .exists()
                    .await
                    .with_context(|| format!("cannot reach the object store at {endpoint}"))?
                {
                    return Ok(());
                }

                let BucketCreate {
                    name,
                    region,
                    credentials,
                    virtual_host,
                } = &self.create;
                // Readable by anyone: a version record carries a plain URL,
                // and players downloading a mod are not signed in to anything
                // of ours. Writes still need the credentials, and the only way
                // to get one is a signed URL handed to a listing's owner.
                let settings = BucketConfiguration::public();
                if *virtual_host {
                    Bucket::create(name, region.clone(), credentials.clone(), settings).await
                } else {
                    Bucket::create_with_path_style(
                        name,
                        region.clone(),
                        credentials.clone(),
                        settings,
                    )
                    .await
                }
                .with_context(|| format!("cannot create bucket {name}"))?;

                Ok(())
            })
            .await
            .copied()
    }

    /// A URL the browser can `PUT` a deliverable to, and the key it lands at.
    ///
    /// The key is ours to choose, never the caller's: it is what stops one
    /// author's upload from overwriting another's object.
    pub async fn presign_upload(
        &self,
        author: &Did,
        listing: &Rkey,
        filename: &str,
    ) -> Result<(String, String)> {
        let key = format!(
            "{}/{}/{}/{}",
            author,
            listing,
            Rkey::new(),
            sanitize_filename(filename)
        );
        let url = self
            .bucket
            .presign_put(&key, PRESIGN_EXPIRY_SECS, None, None)
            .await?;

        Ok((url, key))
    }

    /// How many bytes an author is storing, across every mod they have.
    ///
    /// Summed from the bucket rather than tracked in the database, so it
    /// counts what is actually costing us storage - including an upload that
    /// no record ended up pointing at.
    pub async fn usage(&self, author: &Did) -> Result<u64> {
        let pages = self.bucket.list(format!("{author}/"), None).await?;
        Ok(pages
            .iter()
            .flat_map(|page| &page.contents)
            .map(|object| object.size)
            .sum())
    }

    /// The size of an uploaded object, or `None` if nothing is there.
    pub async fn size(&self, key: &str) -> Result<Option<u64>> {
        let (head, status) = self.bucket.head_object(key).await?;
        match status {
            200 => Ok(Some(head.content_length.unwrap_or(0).max(0) as u64)),
            404 => Ok(None),
            other => bail!("object store answered {other} for a HEAD of {key}"),
        }
    }

    pub async fn delete(&self, key: &str) -> Result<()> {
        self.bucket.delete_object(key).await?;
        Ok(())
    }

    /// Where a browser downloads `key` from.
    pub fn public_url(&self, key: &str) -> String {
        format!("{}/{key}", self.public_base)
    }
}

/// Reduce an uploaded file's name to something safe to put in a key.
///
/// Keys are path-like and end up in a URL, so anything that could climb out
/// of the prefix we chose, or that needs escaping to survive the trip, is
/// collapsed to an underscore.
fn sanitize_filename(filename: &str) -> String {
    let name = filename.rsplit(['/', '\\']).next().unwrap_or(filename);
    let cleaned: String = name
        .chars()
        .map(|c| {
            if c.is_ascii_alphanumeric() || c == '.' || c == '-' || c == '_' {
                c
            } else {
                '_'
            }
        })
        .take(128)
        .collect();

    // Leading dots would make the object hidden-by-convention and, when the
    // whole name is dots, produce a key segment naming a directory.
    let cleaned = cleaned.trim_start_matches('.').to_string();
    if cleaned.is_empty() {
        "deliverable".to_string()
    } else {
        cleaned
    }
}

/// Read the access key and secret from a JSON credentials file.
pub fn read_credentials_file(path: &std::path::Path) -> Result<(String, String)> {
    #[derive(serde::Deserialize)]
    #[serde(rename_all = "camelCase")]
    struct File {
        access_key: String,
        secret_key: String,
    }

    let contents =
        std::fs::read_to_string(path).with_context(|| format!("cannot read {}", path.display()))?;
    let file: File = serde_json::from_str(&contents)
        .map_err(|e| anyhow!("{} is not {{accessKey, secretKey}}: {e}", path.display()))?;

    Ok((file.access_key, file.secret_key))
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn filenames_cannot_climb_out_of_their_key_prefix() {
        assert_eq!(sanitize_filename("../../secrets"), "secrets");
        assert_eq!(sanitize_filename("/etc/passwd"), "passwd");
        assert_eq!(sanitize_filename("C:\\Users\\mod.bps"), "mod.bps");
        assert_eq!(sanitize_filename(".."), "deliverable");
        assert_eq!(sanitize_filename(""), "deliverable");
    }

    #[test]
    fn ordinary_filenames_survive_intact() {
        assert_eq!(
            sanitize_filename("master-quest_v1.0.bps"),
            "master-quest_v1.0.bps"
        );
    }

    #[test]
    fn spaces_and_quotes_do_not_reach_the_key() {
        assert_eq!(
            sanitize_filename("my mod \"final\".zip"),
            "my_mod__final_.zip"
        );
    }
}