Implement native asset pipeline and cache (#64)

crates/libremetaverse/src/asset_cache.rs

@@ -0,0 +1,484 @@
+//! Cross-platform, bounded on-disk asset cache.
+
+#![allow(
+    clippy::cast_possible_truncation,
+    clippy::cast_sign_loss,
+    clippy::collapsible_if,
+    clippy::missing_errors_doc,
+    clippy::must_use_candidate,
+    clippy::needless_pass_by_value,
+    clippy::single_match_else,
+    clippy::type_complexity,
+    clippy::uninlined_format_args,
+    clippy::unused_async,
+    clippy::unused_self
+)]
+
+use crate::{Error, GridClient, ImageDownload};
+use libremetaverse_types::UUID;
+use libremetaverse_types::compat::{CancellationToken, Object};
+use std::fs::{self, OpenOptions};
+use std::io::Write;
+use std::path::{Path, PathBuf};
+use std::sync::Arc;
+use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
+use std::time::{SystemTime, UNIX_EPOCH};
+
+const CACHE_TARGET_PERCENT: i64 = 90;
+const ABSOLUTE_CACHE_FILE_LIMIT: u64 = crate::asset_models::MAX_ASSET_BYTES as u64;
+
+type CacheFilenameHandler = dyn Fn(String, UUID) -> Result<String, Error> + Send + Sync;
+
+#[derive(Clone)]
+pub struct AssetCacheComputeAssetCacheFilenameDelegate {
+    handler: Arc<CacheFilenameHandler>,
+}
+
+impl std::fmt::Debug for AssetCacheComputeAssetCacheFilenameDelegate {
+    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        formatter
+            .debug_struct("AssetCacheComputeAssetCacheFilenameDelegate")
+            .finish_non_exhaustive()
+    }
+}
+
+impl AssetCacheComputeAssetCacheFilenameDelegate {
+    pub fn from_handler(
+        handler: impl Fn(String, UUID) -> Result<String, Error> + Send + Sync + 'static,
+    ) -> Self {
+        Self {
+            handler: Arc::new(handler),
+        }
+    }
+
+    pub fn new(object: Object, method: isize) -> Result<Self, Error> {
+        if method == 0 {
+            return Err(Error::Argument);
+        }
+        object
+            .downcast_arc::<Self>()
+            .map(|value| (*value).clone())
+            .ok_or(Error::Argument)
+    }
+
+    pub fn invoke(&self, cache_dir: String, asset_id: UUID) -> Result<String, Error> {
+        std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
+            (self.handler)(cache_dir, asset_id)
+        }))
+        .map_err(|_| Error::InvalidOperation)?
+    }
+
+    pub fn begin_invoke(
+        &self,
+        cache_dir: String,
+        asset_id: UUID,
+        callback: Box<dyn Fn(&dyn std::any::Any) + Send + Sync>,
+        object: Object,
+    ) -> Result<Box<dyn std::any::Any + Send + Sync>, Error> {
+        let result = self.invoke(cache_dir, asset_id)?;
+        callback(&object);
+        Ok(Box::new(result))
+    }
+
+    pub fn end_invoke(
+        &self,
+        result: Box<dyn std::any::Any + Send + Sync>,
+    ) -> Result<String, Error> {
+        result
+            .downcast::<String>()
+            .map(|value| *value)
+            .map_err(|_| Error::Argument)
+    }
+}
+
+struct AssetCacheInner {
+    client: crate::client_core::ClientWeakHandle,
+    pruning: AtomicBool,
+    auto_prune: AtomicBool,
+    interval_bits: AtomicU64,
+    last_prune_ms: AtomicU64,
+}
+
+pub struct AssetCache {
+    inner: Arc<AssetCacheInner>,
+    pub compute_asset_cache_filename: Option<AssetCacheComputeAssetCacheFilenameDelegate>,
+}
+
+impl Clone for AssetCache {
+    fn clone(&self) -> Self {
+        Self {
+            inner: Arc::clone(&self.inner),
+            compute_asset_cache_filename: self.compute_asset_cache_filename.clone(),
+        }
+    }
+}
+
+impl std::fmt::Debug for AssetCache {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("AssetCache")
+            .field("auto_prune_enabled", &self.auto_prune_enabled())
+            .finish_non_exhaustive()
+    }
+}
+
+impl AssetCache {
+    fn now_ms() -> u64 {
+        u64::try_from(
+            SystemTime::now()
+                .duration_since(UNIX_EPOCH)
+                .unwrap_or_default()
+                .as_millis(),
+        )
+        .unwrap_or(u64::MAX)
+    }
+
+    pub fn new(client: GridClient) -> Result<Self, Error> {
+        Ok(Self {
+            inner: Arc::new(AssetCacheInner {
+                client: client.native_weak_handle(),
+                pruning: AtomicBool::new(false),
+                auto_prune: AtomicBool::new(true),
+                interval_bits: AtomicU64::new((300_000_f64).to_bits()),
+                last_prune_ms: AtomicU64::new(Self::now_ms()),
+            }),
+            compute_asset_cache_filename: None,
+        })
+    }
+
+    fn config(&self) -> Result<crate::AssetCacheSettings, Error> {
+        let client = self.inner.client.upgrade().ok_or(Error::InvalidOperation)?;
+        Ok(client.settings_ref().asset_cache())
+    }
+
+    fn cache_dir(&self) -> Result<PathBuf, Error> {
+        let config = self.config()?;
+        let dir = PathBuf::from(config.dir);
+        if dir.as_os_str().is_empty() {
+            return Err(Error::InvalidOperation);
+        }
+        Ok(dir)
+    }
+
+    fn path(&self, id: UUID) -> Result<PathBuf, Error> {
+        let dir = self.cache_dir()?;
+        if let Some(compute) = &self.compute_asset_cache_filename {
+            let path = PathBuf::from(compute.invoke(dir.to_string_lossy().into_owned(), id)?);
+            if path.parent() == Some(dir.as_path()) {
+                return Ok(path);
+            }
+            return Err(Error::Argument);
+        }
+        Ok(dir.join(id.to_string()))
+    }
+
+    fn operational(&self) -> Result<bool, Error> {
+        Ok(self.config()?.enabled)
+    }
+
+    fn checked_read(&self, path: &Path) -> Result<Option<Vec<u8>>, Error> {
+        let metadata = match fs::metadata(path) {
+            Ok(value) => value,
+            Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(None),
+            Err(_) => return Err(Error::InvalidOperation),
+        };
+        if !metadata.is_file() || metadata.len() == 0 || metadata.len() > ABSOLUTE_CACHE_FILE_LIMIT
+        {
+            return Err(Error::Argument);
+        }
+        let bytes = fs::read(path).map_err(|_| Error::InvalidOperation)?;
+        if bytes.len() as u64 != metadata.len() {
+            return Err(Error::InvalidOperation);
+        }
+        Ok(Some(bytes))
+    }
+
+    pub fn get_cached_asset_bytes_with_uuid(
+        &self,
+        asset_id: UUID,
+    ) -> Result<Option<Vec<u8>>, Error> {
+        if !self.operational()? {
+            return Ok(None);
+        }
+        self.checked_read(&self.path(asset_id)?)
+    }
+
+    pub async fn get_cached_asset_bytes_with_uuid_cancellation_token(
+        &self,
+        asset_id: UUID,
+        cancellation_token: Option<CancellationToken>,
+    ) -> Result<Option<Vec<u8>>, Error> {
+        if let Some(token) = &cancellation_token {
+            token.throw_if_cancellation_requested()?;
+        }
+        let result = self.get_cached_asset_bytes_with_uuid(asset_id);
+        if let Some(token) = &cancellation_token {
+            token.throw_if_cancellation_requested()?;
+        }
+        result
+    }
+
+    pub fn try_get_cached_asset_bytes_with_uuid_bytes(
+        &self,
+        asset_id: UUID,
+        data: &mut Option<Vec<u8>>,
+    ) -> bool {
+        match self.get_cached_asset_bytes_with_uuid(asset_id) {
+            Ok(Some(bytes)) => {
+                *data = Some(bytes);
+                true
+            }
+            _ => {
+                *data = None;
+                false
+            }
+        }
+    }
+
+    pub async fn try_get_cached_asset_bytes_with_uuid_cancellation_token(
+        &self,
+        asset_id: UUID,
+        cancellation_token: Option<CancellationToken>,
+    ) -> Result<(bool, Option<Vec<u8>>), Error> {
+        let data = self
+            .get_cached_asset_bytes_with_uuid_cancellation_token(asset_id, cancellation_token)
+            .await?;
+        Ok((data.is_some(), data))
+    }
+
+    pub fn save_asset_to_cache_with_uuid_bytes(
+        &self,
+        asset_id: UUID,
+        asset_data: Vec<u8>,
+    ) -> Result<bool, Error> {
+        if !self.operational()? {
+            return Ok(false);
+        }
+        if asset_data.is_empty() || asset_data.len() > crate::asset_models::MAX_ASSET_BYTES {
+            return Err(Error::Argument);
+        }
+        let path = self.path(asset_id)?;
+        if path.is_file() {
+            return Ok(true);
+        }
+        let dir = path.parent().ok_or(Error::Argument)?;
+        fs::create_dir_all(dir).map_err(|_| Error::InvalidOperation)?;
+        let nonce = UUID::random()?.to_string();
+        let temp = dir.join(format!(".{}.{}.tmp", asset_id, nonce));
+        let write_result = (|| -> Result<(), Error> {
+            let mut file = OpenOptions::new()
+                .write(true)
+                .create_new(true)
+                .open(&temp)
+                .map_err(|_| Error::InvalidOperation)?;
+            file.write_all(&asset_data)
+                .map_err(|_| Error::InvalidOperation)?;
+            file.sync_all().map_err(|_| Error::InvalidOperation)?;
+            match fs::rename(&temp, &path) {
+                Ok(()) => Ok(()),
+                Err(_) if path.is_file() => Ok(()),
+                Err(_) => Err(Error::InvalidOperation),
+            }
+        })();
+        if write_result.is_err() || temp.exists() {
+            let _ = fs::remove_file(&temp);
+        }
+        write_result.map(|()| {
+            self.maybe_begin_prune();
+            true
+        })
+    }
+
+    fn maybe_begin_prune(&self) {
+        if !self.auto_prune_enabled() {
+            return;
+        }
+        let interval = self.auto_prune_interval();
+        if !interval.is_finite() || interval <= 0.0 {
+            return;
+        }
+        let now = Self::now_ms();
+        let last = self.inner.last_prune_ms.load(Ordering::Acquire);
+        if now.saturating_sub(last) < interval as u64 {
+            return;
+        }
+        if self
+            .inner
+            .last_prune_ms
+            .compare_exchange(last, now, Ordering::AcqRel, Ordering::Acquire)
+            .is_ok()
+        {
+            let _ = self.begin_prune();
+        }
+    }
+
+    pub async fn save_asset_to_cache_with_uuid_bytes_cancellation_token(
+        &self,
+        asset_id: UUID,
+        asset_data: Vec<u8>,
+        cancellation_token: Option<CancellationToken>,
+    ) -> Result<bool, Error> {
+        if let Some(token) = &cancellation_token {
+            token.throw_if_cancellation_requested()?;
+        }
+        let result = self.save_asset_to_cache_with_uuid_bytes(asset_id, asset_data);
+        if let Some(token) = &cancellation_token {
+            token.throw_if_cancellation_requested()?;
+        }
+        result
+    }
+
+    pub fn asset_file_name(&self, asset_id: UUID) -> Result<Option<String>, Error> {
+        if !self.operational()? {
+            return Ok(None);
+        }
+        let path = self.path(asset_id)?;
+        Ok(path.is_file().then(|| path.to_string_lossy().into_owned()))
+    }
+
+    pub fn has_asset(&self, asset_id: UUID) -> Result<bool, Error> {
+        Ok(self.operational()? && self.path(asset_id)?.is_file())
+    }
+
+    fn cache_entries(&self) -> Result<Vec<(PathBuf, u64, SystemTime)>, Error> {
+        let dir = self.cache_dir()?;
+        let mut entries = Vec::new();
+        let read = match fs::read_dir(dir) {
+            Ok(value) => value,
+            Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(entries),
+            Err(_) => return Err(Error::InvalidOperation),
+        };
+        for entry in read.flatten() {
+            if let Ok(metadata) = entry.metadata() {
+                if metadata.is_file() {
+                    entries.push((
+                        entry.path(),
+                        metadata.len(),
+                        metadata
+                            .accessed()
+                            .or_else(|_| metadata.modified())
+                            .unwrap_or(UNIX_EPOCH),
+                    ));
+                }
+            }
+        }
+        Ok(entries)
+    }
+
+    pub fn clear(&self) -> Result<(), Error> {
+        for (path, _, _) in self.cache_entries()? {
+            fs::remove_file(path).map_err(|_| Error::InvalidOperation)?;
+        }
+        Ok(())
+    }
+
+    pub async fn prune(&self, cancellation_token: Option<CancellationToken>) -> Result<(), Error> {
+        if let Some(token) = &cancellation_token {
+            token.throw_if_cancellation_requested()?;
+        }
+        let max = self.config()?.max_size;
+        if max <= 0 {
+            return Err(Error::InvalidOperation);
+        }
+        let mut entries = self.cache_entries()?;
+        let mut size: u64 = entries.iter().map(|(_, size, _)| *size).sum();
+        let max = u64::try_from(max).map_err(|_| Error::InvalidOperation)?;
+        if size <= max {
+            return Ok(());
+        }
+        entries.sort_unstable_by_key(|(_, _, accessed)| *accessed);
+        let target = max.saturating_mul(CACHE_TARGET_PERCENT as u64) / 100;
+        for (path, length, _) in entries {
+            if let Some(token) = &cancellation_token {
+                token.throw_if_cancellation_requested()?;
+            }
+            fs::remove_file(path).map_err(|_| Error::InvalidOperation)?;
+            size = size.saturating_sub(length);
+            if size <= target {
+                break;
+            }
+        }
+        Ok(())
+    }
+
+    pub fn begin_prune(&self) -> Result<(), Error> {
+        if self.inner.pruning.swap(true, Ordering::AcqRel) {
+            return Ok(());
+        }
+        let cache = self.clone();
+        std::thread::Builder::new()
+            .name("asset-cache-prune".into())
+            .spawn(move || {
+                let _ = cache.prune_sync(None);
+                cache.inner.pruning.store(false, Ordering::Release);
+            })
+            .map_err(|_| {
+                self.inner.pruning.store(false, Ordering::Release);
+                Error::InvalidOperation
+            })?;
+        Ok(())
+    }
+
+    fn prune_sync(&self, cancellation_token: Option<CancellationToken>) -> Result<(), Error> {
+        if let Some(token) = &cancellation_token {
+            token.throw_if_cancellation_requested()?;
+        }
+        let max = self.config()?.max_size;
+        if max <= 0 {
+            return Err(Error::InvalidOperation);
+        }
+        let mut entries = self.cache_entries()?;
+        let mut size: u64 = entries.iter().map(|(_, size, _)| *size).sum();
+        let max = u64::try_from(max).map_err(|_| Error::InvalidOperation)?;
+        if size <= max {
+            return Ok(());
+        }
+        entries.sort_unstable_by_key(|(_, _, accessed)| *accessed);
+        let target = max.saturating_mul(CACHE_TARGET_PERCENT as u64) / 100;
+        for (path, length, _) in entries {
+            if let Some(token) = &cancellation_token {
+                token.throw_if_cancellation_requested()?;
+            }
+            fs::remove_file(path).map_err(|_| Error::InvalidOperation)?;
+            size = size.saturating_sub(length);
+            if size <= target {
+                break;
+            }
+        }
+        Ok(())
+    }
+
+    pub fn dispose(&self) -> Result<(), Error> {
+        self.inner.auto_prune.store(false, Ordering::Release);
+        Ok(())
+    }
+    pub fn get_cached_image(&self, image_id: UUID) -> Result<Option<ImageDownload>, Error> {
+        let Some(bytes) = self.get_cached_asset_bytes_with_uuid(image_id)? else {
+            return Ok(None);
+        };
+        let mut image = ImageDownload::new()?;
+        image.base.id = image_id;
+        image.base.asset_type = libremetaverse_types::AssetType::Texture;
+        image.base.size = i32::try_from(bytes.len()).map_err(|_| Error::Argument)?;
+        image.base.transferred = image.base.size;
+        image.base.success = true;
+        image.base.asset_data = bytes;
+        image.codec = crate::ImageCodec::J2C;
+        Ok(Some(image))
+    }
+    pub fn auto_prune_enabled(&self) -> bool {
+        self.inner.auto_prune.load(Ordering::Acquire)
+    }
+    pub fn set_auto_prune_enabled(&mut self, value: bool) {
+        self.inner.auto_prune.store(value, Ordering::Release);
+    }
+    pub fn auto_prune_interval(&self) -> f64 {
+        f64::from_bits(self.inner.interval_bits.load(Ordering::Acquire))
+    }
+    pub fn set_auto_prune_interval(&mut self, value: f64) {
+        if value.is_finite() && value > 0.0 {
+            self.inner
+                .interval_bits
+                .store(value.to_bits(), Ordering::Release);
+        }
+    }
+}