lfsx-server 0.30.1

use futures_util::Stream;

use super::s3::S3Store;
use super::{Budget, CompressReport, DedupeReport, LocalStore, Object, SweepReport, VerifyReport};
use crate::error::Error;
use crate::namespace::Namespace;
#[cfg(test)]
use sha2::Digest;
#[cfg(test)]
use std::time::Duration;

// Where the objects live. A bucket decouples capacity from the machine, at the
// price of the things a filesystem gave for nothing — hard links, a directory
// walk, and a rename that is atomic. Each of those is answered here or refused
// out loud; none of them is quietly skipped.
pub enum Store {
    Local(LocalStore),
    // Even with a bucket the local store stays, because a transfer has to land
    // somewhere before anyone can tell whether it is the object it claims to be.
    // It is a write buffer, not the store.
    // Boxed because a bucket handle beside a local store makes this variant far
    // larger than the other, and every Store in the process would pay for it.
    Bucket {
        bucket: Box<S3Store>,
        staging: LocalStore,
    },
}

impl Store {
    fn staging(&self) -> &LocalStore {
        match self {
            Self::Local(store) => store,
            Self::Bucket { staging, .. } => staging,
        }
    }

    pub async fn writable(&self) -> Result<(), Error> {
        self.staging().writable().await
    }

    pub fn scans(&self) -> u64 {
        self.staging().scans()
    }

    pub async fn exists(&self, ns: &Namespace, oid: &str) -> bool {
        match self {
            Self::Local(store) => store.exists(ns, oid).await,
            Self::Bucket { bucket, .. } => bucket.exists(ns, oid).await,
        }
    }

    pub async fn open(&self, ns: &Namespace, oid: &str) -> Result<Object, Error> {
        match self {
            Self::Local(store) => store.open(ns, oid).await,
            Self::Bucket { bucket, .. } => {
                if !bucket.exists(ns, oid).await {
                    return Err(Error::NotFound);
                }

                Ok(Object::Remote {
                    bucket: (**bucket).clone(),
                    oid: oid.to_owned(),
                    size: bucket.size_of(oid).await?,
                })
            }
        }
    }

    pub async fn write<S, E>(
        &self,
        ns: &Namespace,
        oid: &str,
        expected_size: Option<u64>,
        budget: Option<Budget>,
        chunks: S,
    ) -> Result<u64, Error>
    where
        S: Stream<Item = Result<axum::body::Bytes, E>> + Unpin,
        E: std::error::Error + Send + Sync + 'static,
    {
        match self {
            Self::Local(store) => store.write(ns, oid, expected_size, budget, chunks).await,
            Self::Bucket { bucket, staging } => {
                let staged = staging
                    .stage(ns, oid, expected_size, budget, chunks)
                    .await?;
                let outcome = bucket.store(ns, oid, &staged.path).await;

                // The staging file has served its purpose either way. Leaving it
                // would be a leak the reclaimer only notices a day later.
                let _ = tokio::fs::remove_file(&staged.path).await;
                outcome?;

                Ok(staged.written)
            }
        }
    }

    // None rather than zero: a bucket has no cheap answer for what the whole
    // store holds, and building one from a full listing would cost a request per
    // object on every scrape. Zero would be read as an empty bucket by every
    // dashboard that averages it, which is the one lie this seam otherwise
    // refuses to tell — everything else it cannot do answers 501.
    pub async fn capacity(&self) -> Option<(u64, u64)> {
        match self {
            Self::Local(store) => Some(store.usage().await),
            Self::Bucket { .. } => None,
        }
    }

    pub async fn usage_of(&self, ns: &Namespace) -> (u64, u64) {
        match self {
            Self::Local(store) => store.usage_of(ns).await,
            Self::Bucket { bucket, .. } => bucket.usage_of(ns).await,
        }
    }

    pub async fn sweep(
        &self,
        ns: &Namespace,
        retained: &std::collections::HashSet<String>,
        grace: std::time::Duration,
        dry_run: bool,
    ) -> Result<SweepReport, Error> {
        match self {
            Self::Local(store) => store.sweep(ns, retained, grace, dry_run).await,
            Self::Bucket { .. } => Err(Error::Unsupported(
                "collection is not implemented for a bucket yet",
            )),
        }
    }

    pub async fn dedupe(&self, ns: &Namespace, dry_run: bool) -> Result<DedupeReport, Error> {
        match self {
            Self::Local(store) => store.dedupe(ns, dry_run).await,
            // Content addressing already gives this: two repositories pushing the
            // same object write the same key, and each holds a marker beside it.
            // There is nothing left to fold in.
            Self::Bucket { .. } => Err(Error::Unsupported(
                "a bucket stores each object once already, so there is nothing to deduplicate",
            )),
        }
    }

    pub async fn compress(&self, ns: &Namespace, dry_run: bool) -> Result<CompressReport, Error> {
        match self {
            Self::Local(store) => store.compress(ns, dry_run).await,
            Self::Bucket { .. } => Err(Error::Unsupported(
                "compression is not implemented for a bucket yet",
            )),
        }
    }

    pub async fn verify(&self, ns: &Namespace) -> Result<VerifyReport, Error> {
        match self {
            Self::Local(store) => store.verify(ns).await,
            Self::Bucket { .. } => Err(Error::Unsupported(
                "verification is not implemented for a bucket yet",
            )),
        }
    }
}

#[cfg(test)]
mod tests {
    use futures_util::StreamExt;

    use super::*;
    use crate::storage::s3::tests::{bucket, store};

    fn namespace() -> Namespace {
        Namespace::new("FerrLabs", "Blastlands").unwrap()
    }

    async fn bucket_store(root: &tempfile::TempDir, endpoint: &str) -> Store {
        Store::Bucket {
            bucket: Box::new(store(endpoint)),
            staging: LocalStore::new(root.path()),
        }
    }

    #[tokio::test]
    async fn an_upload_lands_in_the_bucket_and_reads_back_through_the_same_seam() {
        let root = tempfile::tempdir().unwrap();
        let (endpoint, _objects) = bucket().await;
        let store = bucket_store(&root, &endpoint).await;
        let payload = b"an asset that never touches this disk for long".repeat(32);
        let oid = hex::encode(sha2::Sha256::digest(&payload));

        let written = store
            .write(
                &namespace(),
                &oid,
                Some(payload.len() as u64),
                None,
                futures_util::stream::iter([Ok::<_, std::io::Error>(axum::body::Bytes::from(
                    payload.clone(),
                ))]),
            )
            .await
            .unwrap();

        assert_eq!(written, payload.len() as u64);
        assert!(store.exists(&namespace(), &oid).await);

        let object = store.open(&namespace(), &oid).await.unwrap();
        let size = object.size();
        let mut chunks = object.stream(0, size).await.unwrap();
        let mut out = Vec::new();
        while let Some(chunk) = chunks.next().await {
            out.extend_from_slice(&chunk.unwrap());
        }

        assert_eq!(out, payload);
    }

    #[tokio::test]
    async fn the_staging_file_does_not_outlive_the_upload() {
        let root = tempfile::tempdir().unwrap();
        let (endpoint, _objects) = bucket().await;
        let store = bucket_store(&root, &endpoint).await;
        let payload = b"an asset passing through".to_vec();
        let oid = hex::encode(sha2::Sha256::digest(&payload));

        store
            .write(
                &namespace(),
                &oid,
                None,
                None,
                futures_util::stream::iter([Ok::<_, std::io::Error>(axum::body::Bytes::from(
                    payload,
                ))]),
            )
            .await
            .unwrap();

        let leftovers = crate::storage::tests::staging_files(root.path());
        assert!(
            leftovers.is_empty(),
            "local disk is a write buffer here, and one that is never emptied is a disk that \
             fills: {leftovers:?}"
        );
    }

    #[tokio::test]
    async fn a_bucket_reports_no_capacity_rather_than_an_empty_one() {
        let root = tempfile::tempdir().unwrap();
        let (endpoint, _objects) = bucket().await;

        assert!(
            bucket_store(&root, &endpoint)
                .await
                .capacity()
                .await
                .is_none(),
            "zero would be read as an empty store by every dashboard that averages it"
        );
        assert!(
            Store::Local(LocalStore::new(root.path()))
                .capacity()
                .await
                .is_some()
        );
    }

    #[tokio::test]
    async fn the_maintenance_commands_say_they_do_not_apply_rather_than_lying() {
        let root = tempfile::tempdir().unwrap();
        let (endpoint, _objects) = bucket().await;
        let store = bucket_store(&root, &endpoint).await;
        let ns = namespace();

        for outcome in [
            store.dedupe(&ns, true).await.err(),
            store.compress(&ns, true).await.err(),
            store.verify(&ns).await.err(),
            store
                .sweep(&ns, &std::collections::HashSet::new(), Duration::ZERO, true)
                .await
                .err(),
        ] {
            assert!(
                matches!(outcome, Some(Error::Unsupported(_))),
                "an operator running collection against a bucket has to be told it did nothing, \
                 not handed an empty report that reads like success: {outcome:?}"
            );
        }
    }
}