Mangalord/backend/src/storage/mod.rs

//! Pluggable blob storage.
//!
//! Handlers depend on the `Storage` trait, never on a concrete backend.
//! Add new backends (S3, GCS, …) as new impls in this module and wire
//! them up in `app::build` based on config.

mod local;

use std::io;
use std::pin::Pin;

use std::path::Path;

use async_trait::async_trait;
use bytes::Bytes;
use futures_core::Stream;

pub use local::LocalStorage;

#[derive(thiserror::Error, Debug)]
pub enum StorageError {
    #[error(transparent)]
    Io(#[from] io::Error),
    #[error("not found")]
    NotFound,
    #[error("invalid storage key")]
    BadKey,
}

/// Boxed byte stream returned by `Storage::get_stream` so the trait stays
/// object-safe regardless of the concrete reader behind it.
pub type ByteStream = Pin<Box<dyn Stream<Item = io::Result<Bytes>> + Send>>;

/// Boxed byte stream accepted by `Storage::put_stream`. The item type
/// is fallible so a producer (e.g. an HTTP body) can surface a transport
/// error mid-stream without breaking the trait shape; the storage impl
/// is responsible for not installing a partial blob on such an error.
pub type PutByteStream<'a> =
    Pin<Box<dyn Stream<Item = Result<Bytes, StorageError>> + Send + 'a>>;

pub struct StreamingFile {
    pub stream: ByteStream,
    pub size_bytes: u64,
}

#[async_trait]
pub trait Storage: Send + Sync {
    async fn put(&self, key: &str, bytes: &[u8]) -> Result<(), StorageError>;
    /// Stream a blob to storage without holding the entire body in
    /// memory. The chapter-content download path uses this so peak
    /// memory stays at one chunk per concurrent dispatch (not one full
    /// page image). The contract is atomic: a stream that errors mid-way
    /// must leave nothing visible at `key` — implementations should
    /// write to a temp location and rename only on the successful
    /// drain. Returns the total bytes written on success.
    ///
    /// The default implementation buffers the stream into memory and
    /// calls `put`, so backends without a native streaming write still
    /// satisfy the contract (at the cost of peak memory). LocalStorage
    /// overrides this to do a temp-file rename; a future S3Storage
    /// would override with a multi-part upload.
    async fn put_stream(
        &self,
        key: &str,
        mut stream: PutByteStream<'_>,
    ) -> Result<u64, StorageError> {
        use futures_util::StreamExt as _;
        let mut buf: Vec<u8> = Vec::new();
        while let Some(chunk) = stream.next().await {
            let chunk = chunk?;
            buf.extend_from_slice(&chunk);
        }
        let len = buf.len() as u64;
        self.put(key, &buf).await?;
        Ok(len)
    }
    /// Reads the entire blob into memory. Convenient for small assets
    /// (covers, thumbnails). For pages and other large blobs, use
    /// `get_stream` so axum can pipe bytes straight to the client.
    async fn get(&self, key: &str) -> Result<Vec<u8>, StorageError>;
    async fn get_stream(&self, key: &str) -> Result<StreamingFile, StorageError>;
    async fn delete(&self, key: &str) -> Result<(), StorageError>;
    async fn exists(&self, key: &str) -> Result<bool, StorageError>;

    /// Move a blob from `from` to `to`, overwriting any existing blob at
    /// `to`. Returns `NotFound` if `from` doesn't exist. The chapter
    /// upload path uses this to promote a staged page to its final,
    /// chapter-scoped key once the chapter row (and thus its id) exists —
    /// so pages can be streamed to storage as their multipart parts
    /// arrive, without buffering the whole chapter in memory.
    ///
    /// The default implementation streams `from` to `to` and deletes the
    /// source, so backends without a native move still satisfy the
    /// contract. LocalStorage overrides it with a filesystem rename (an
    /// atomic metadata op within a mount); a future S3Storage would
    /// override with a server-side copy + delete.
    async fn rename(&self, from: &str, to: &str) -> Result<(), StorageError> {
        use futures_util::StreamExt as _;
        let StreamingFile { stream, .. } = self.get_stream(from).await?;
        let mapped: PutByteStream<'_> = Box::pin(stream.map(|r| r.map_err(StorageError::Io)));
        self.put_stream(to, mapped).await?;
        self.delete(from).await?;
        Ok(())
    }

    /// Size in bytes of the blob at `key`, `NotFound` when it doesn't
    /// exist. Cheap metadata lookup (local: `fs::metadata`; a future
    /// `S3Storage`: HEAD object). Used by the cover-capture path and the
    /// one-shot storage-size backfill — never on the hot read path, which
    /// reads the stored `size_bytes` columns. Required (no default) so a
    /// new backend can't silently skip it.
    async fn size(&self, key: &str) -> Result<u64, StorageError>;

    /// Filesystem path the backend is rooted at, when introspectable.
    /// Returns `None` for backends that aren't a local filesystem (e.g.
    /// a future `S3Storage`). The admin system endpoint uses this to
    /// statvfs the data dir; backends that return `None` get a `disk:
    /// null` payload instead of fabricated numbers.
    fn local_root(&self) -> Option<&Path> {
        None
    }
}