appcore-storage 1.0.1-rc.8

// =============================================================================
//        #######
//     ###       ###     F: storage_file.rs
//    ##   ## ##   ##    P: AppCore-Runtime
//         ## ##
//                       C: 2026/06/04 12:01:22 by dnettoRaw
//    ##   ## ##   ##    U: 2026/06/04 12:01:22 by dnettoRaw
//      ###########      S: 0.6.0
// =============================================================================

use super::*;
use crate::storage::RemoteAuthStorageClient;
use appcore_security::{TokenClaims, TokenProvider};
use std::fs::{self, OpenOptions};
use std::io::{Read, Write};
use std::path::{Component, Path, PathBuf};
use std::sync::atomic::{AtomicUsize, Ordering};
use std::time::{SystemTime, UNIX_EPOCH};

/// Single-process local file provider with atomic replacement writes.
#[derive(Debug, Clone)]
pub struct FileStorageProvider {
    pub(super) storage_path: PathBuf,
    pub(super) backup_path: PathBuf,
    opened: bool,
}

impl FileStorageProvider {
    /// Creates a provider with separate data and backup roots.
    pub fn new(storage_path: impl Into<PathBuf>, backup_path: impl Into<PathBuf>) -> Self {
        Self {
            storage_path: storage_path.into(),
            backup_path: backup_path.into(),
            opened: false,
        }
    }

    /// Creates the configured data and backup directories.
    pub fn create_dirs(&self) -> StorageResult<()> {
        if let Some(parent) = self.storage_path.parent() {
            fs::create_dir_all(parent).map_err(|_| StorageError::NotAvailable)?;
        }
        fs::create_dir_all(&self.backup_path).map_err(|_| StorageError::NotAvailable)?;
        self.recover_snapshot_restore()?;
        fs::create_dir_all(&self.storage_path).map_err(|_| StorageError::NotAvailable)?;
        Ok(())
    }

    /// Atomically writes bytes below the data root.
    pub fn write_bytes(&self, path: &str, bytes: &[u8]) -> StorageResult<()> {
        self.write_bytes_atomic(path, bytes)
    }

    /// Writes through an exclusive temporary file, fsync and atomic rename.
    pub fn write_bytes_atomic(&self, path: &str, bytes: &[u8]) -> StorageResult<()> {
        let full = self.resolve_storage(path)?;
        if let Some(parent) = full.parent() {
            fs::create_dir_all(parent).map_err(|_| StorageError::NotAvailable)?;
        }
        let tmp = tmp_path_for(&full);
        self.with_storage_lock(|| {
            write_atomic_file(&tmp, &full, bytes)
                .map_err(|_| StorageError::TransactionFailed(path.to_string()))
        })?;
        Ok(())
    }

    /// Seals and atomically writes bytes using an explicit token provider.
    pub fn write_secure_bytes<P: TokenProvider>(
        &self,
        path: &str,
        bytes: &[u8],
        provider: &P,
        claims: &TokenClaims,
    ) -> StorageResult<()> {
        let sealed = provider
            .seal(bytes, claims)
            .map_err(|_| StorageError::SecurityFailed(path.to_string()))?;
        self.write_bytes_atomic(path, &sealed)
    }

    /// Writes sealed bytes and fails when authentication is not configured.
    pub fn write_auth_required_bytes<P: TokenProvider>(
        &self,
        path: &str,
        bytes: &[u8],
        provider: Option<&P>,
        claims: &TokenClaims,
    ) -> StorageResult<()> {
        let provider = require_auth_provider(path, provider)?;
        self.write_secure_bytes(path, bytes, provider, claims)
    }

    /// Seals bytes through a remote authentication service before writing.
    pub fn write_remote_auth_required_bytes(
        &self,
        path: &str,
        bytes: &[u8],
        client: Option<&RemoteAuthStorageClient>,
    ) -> StorageResult<()> {
        let client = require_remote_auth_client(path, client)?;
        let sealed = client.seal_resource(path, bytes)?;
        self.write_bytes_atomic(path, &sealed)
    }

    /// Reads bytes below the data root.
    pub fn read_bytes(&self, path: &str) -> StorageResult<Vec<u8>> {
        let full = self.resolve_storage(path)?;
        fs::read(full).map_err(|_| StorageError::RepositoryNotFound(path.to_string()))
    }

    pub(super) fn read_bytes_bounded(&self, path: &str, max_bytes: u64) -> StorageResult<Vec<u8>> {
        let full = self.resolve_storage(path)?;
        let file = fs::File::open(&full)
            .map_err(|_| StorageError::RepositoryNotFound(path.to_string()))?;
        let metadata = file
            .metadata()
            .map_err(|_| StorageError::RepositoryNotFound(path.to_string()))?;
        if !metadata.is_file() {
            return Err(StorageError::InvalidPath(path.to_string()));
        }
        if metadata.len() > max_bytes {
            return Err(StorageError::TransactionFailed(path.to_string()));
        }
        let capacity = usize::try_from(metadata.len())
            .map_err(|_| StorageError::TransactionFailed(path.to_string()))?;
        let read_limit = max_bytes
            .checked_add(1)
            .ok_or_else(|| StorageError::TransactionFailed(path.to_string()))?;
        let mut bytes = Vec::with_capacity(capacity);
        file.take(read_limit)
            .read_to_end(&mut bytes)
            .map_err(|_| StorageError::RepositoryNotFound(path.to_string()))?;
        if bytes.len() as u64 > max_bytes {
            return Err(StorageError::TransactionFailed(path.to_string()));
        }
        Ok(bytes)
    }

    /// Reads and opens sealed bytes using an explicit token provider.
    pub fn read_secure_bytes<P: TokenProvider>(
        &self,
        path: &str,
        provider: &P,
        claims: &TokenClaims,
    ) -> StorageResult<Vec<u8>> {
        let sealed = self.read_bytes(path)?;
        provider
            .open(&sealed, claims)
            .map_err(|_| StorageError::SecurityFailed(path.to_string()))
    }

    /// Reads sealed bytes and fails when authentication is not configured.
    pub fn read_auth_required_bytes<P: TokenProvider>(
        &self,
        path: &str,
        provider: Option<&P>,
        claims: &TokenClaims,
    ) -> StorageResult<Vec<u8>> {
        let provider = require_auth_provider(path, provider)?;
        self.read_secure_bytes(path, provider, claims)
    }

    /// Reads and opens bytes through a remote authentication service.
    pub fn read_remote_auth_required_bytes(
        &self,
        path: &str,
        client: Option<&RemoteAuthStorageClient>,
    ) -> StorageResult<Vec<u8>> {
        let client = require_remote_auth_client(path, client)?;
        let sealed = self.read_bytes(path)?;
        client.open_resource(path, &sealed)
    }

    /// Reports whether a relative data path exists.
    pub fn exists(&self, path: &str) -> StorageResult<bool> {
        let full = self.resolve_storage(path)?;
        Ok(full.exists())
    }

    /// Atomically copies a data file below the backup root.
    pub fn backup_file(&self, source: &str, backup_name: &str) -> StorageResult<()> {
        self.backup_file_atomic(source, backup_name)
    }

    /// Creates a backup through fsync and atomic replacement.
    pub fn backup_file_atomic(&self, source: &str, backup_name: &str) -> StorageResult<()> {
        let source_full = self.resolve_storage(source)?;
        if !source_full.exists() {
            return Err(StorageError::RepositoryNotFound(source.to_string()));
        }
        let backup_full = self.resolve_backup(backup_name)?;
        if let Some(parent) = backup_full.parent() {
            fs::create_dir_all(parent)
                .map_err(|_| StorageError::BackupFailed(backup_name.to_string()))?;
        }
        self.with_storage_lock(|| {
            let bytes = fs::read(source_full)
                .map_err(|_| StorageError::BackupFailed(backup_name.to_string()))?;
            let tmp = tmp_path_for(&backup_full);
            write_atomic_file(&tmp, &backup_full, &bytes)
                .map_err(|_| StorageError::BackupFailed(backup_name.to_string()))
        })?;
        Ok(())
    }

    /// Removes orphaned temporary files below data and backup roots.
    pub fn cleanup_temp_files(&self) -> StorageResult<usize> {
        let mut removed = 0usize;
        removed += cleanup_tmp_in_dir(&self.storage_path)?;
        removed += cleanup_tmp_in_dir(&self.backup_path)?;
        Ok(removed)
    }

    fn resolve_storage(&self, relative: &str) -> StorageResult<PathBuf> {
        resolve_under_root(&self.storage_path, relative)
    }

    fn resolve_backup(&self, relative: &str) -> StorageResult<PathBuf> {
        resolve_under_root(&self.backup_path, relative)
    }
}

pub(super) fn resolve_under_root(root: &Path, relative: &str) -> StorageResult<PathBuf> {
    let rel = Path::new(relative);
    if rel.as_os_str().is_empty() || rel.is_absolute() {
        return Err(StorageError::InvalidPath(relative.to_string()));
    }
    let mut current = root.to_path_buf();
    for component in rel.components() {
        if matches!(
            component,
            Component::ParentDir | Component::RootDir | Component::Prefix(_)
        ) {
            return Err(StorageError::InvalidPath(relative.to_string()));
        }
        if let Component::Normal(part) = component {
            current.push(part);
            if fs::symlink_metadata(&current)
                .map(|metadata| metadata.file_type().is_symlink())
                .unwrap_or(false)
            {
                return Err(StorageError::InvalidPath(relative.to_string()));
            }
        }
    }
    Ok(current)
}

static TMP_COUNTER: AtomicUsize = AtomicUsize::new(0);

pub(crate) fn tmp_path_for(path: &Path) -> PathBuf {
    let pid = std::process::id();
    let thread_id = format!("{:?}", std::thread::current().id());
    let clean_thread_id: String = thread_id
        .chars()
        .filter(|c| c.is_ascii_alphanumeric())
        .collect();
    let counter = TMP_COUNTER.fetch_add(1, Ordering::SeqCst);

    let mut tmp_name = path
        .file_name()
        .map(|n| n.to_os_string())
        .unwrap_or_else(|| "tmp".into());
    tmp_name.push(format!(".{}_{}_{}.tmp", pid, clean_thread_id, counter));
    path.with_file_name(tmp_name)
}

#[cfg(unix)]
pub(super) fn fsync_parent(path: &Path) -> std::io::Result<()> {
    if let Some(parent) = path.parent() {
        fs::File::open(parent)?.sync_all()?;
    }
    Ok(())
}

#[cfg(not(unix))]
pub(super) fn fsync_parent(_path: &Path) -> std::io::Result<()> {
    Ok(())
}

pub(super) fn write_atomic_file(
    tmp: &Path,
    final_path: &Path,
    bytes: &[u8],
) -> std::io::Result<()> {
    write_atomic_file_inner(tmp, final_path, bytes, None)
}

#[cfg_attr(not(test), allow(dead_code))]
#[derive(Debug, Clone, Copy)]
pub(crate) enum AtomicWriteFault {
    DiskFull,
    PermissionDenied,
    AfterPartialWrite,
    AfterFileSync,
    BeforeRename,
}

#[cfg(test)]
pub(crate) fn write_atomic_file_with_fault(
    tmp: &Path,
    final_path: &Path,
    bytes: &[u8],
    fault: AtomicWriteFault,
) -> std::io::Result<()> {
    write_atomic_file_inner(tmp, final_path, bytes, Some(fault))
}

fn write_atomic_file_inner(
    tmp: &Path,
    final_path: &Path,
    bytes: &[u8],
    #[cfg_attr(not(test), allow(unused_variables))] fault: Option<AtomicWriteFault>,
) -> std::io::Result<()> {
    #[cfg(test)]
    match fault {
        Some(AtomicWriteFault::DiskFull) => {
            return Err(std::io::Error::from_raw_os_error(28));
        }
        Some(AtomicWriteFault::PermissionDenied) => {
            return Err(std::io::Error::new(
                std::io::ErrorKind::PermissionDenied,
                "injected permission failure",
            ));
        }
        _ => {}
    }
    let mut file = OpenOptions::new().create_new(true).write(true).open(tmp)?;
    #[cfg(test)]
    if matches!(fault, Some(AtomicWriteFault::AfterPartialWrite)) {
        file.write_all(&bytes[..bytes.len() / 2])?;
        file.sync_all()?;
        return Err(std::io::Error::new(
            std::io::ErrorKind::WriteZero,
            "injected partial write",
        ));
    }
    file.write_all(bytes)?;
    file.sync_all()?;
    #[cfg(test)]
    if matches!(
        fault,
        Some(AtomicWriteFault::AfterFileSync | AtomicWriteFault::BeforeRename)
    ) {
        return Err(std::io::Error::other("injected pre-rename failure"));
    }
    drop(file);
    fs::rename(tmp, final_path)?;
    fsync_parent(final_path)?;
    Ok(())
}

fn require_auth_provider<'a, P: TokenProvider>(
    path: &str,
    provider: Option<&'a P>,
) -> StorageResult<&'a P> {
    provider.ok_or_else(|| StorageError::AuthUnavailable(path.to_string()))
}

fn require_remote_auth_client<'a>(
    path: &str,
    client: Option<&'a RemoteAuthStorageClient>,
) -> StorageResult<&'a RemoteAuthStorageClient> {
    client.ok_or_else(|| StorageError::AuthUnavailable(path.to_string()))
}

fn cleanup_tmp_in_dir(root: &Path) -> StorageResult<usize> {
    if !root.exists() {
        return Ok(0);
    }
    let mut removed = 0usize;
    let entries = fs::read_dir(root).map_err(|_| StorageError::NotAvailable)?;
    for entry in entries.flatten() {
        let path = entry.path();
        if path.is_dir() {
            removed += cleanup_tmp_in_dir(&path)?;
            continue;
        }
        if path
            .file_name()
            .map(|n| n.to_string_lossy().ends_with(".tmp"))
            .unwrap_or(false)
        {
            fs::remove_file(path).map_err(|_| StorageError::NotAvailable)?;
            removed += 1;
        }
    }
    Ok(removed)
}

fn count_tmp_in_dir(root: &Path) -> usize {
    if !root.exists() {
        return 0;
    }
    let mut count = 0usize;
    let entries = match fs::read_dir(root) {
        Ok(entries) => entries,
        Err(_) => return 0,
    };
    for entry in entries.flatten() {
        let path = entry.path();
        if path.is_dir() {
            count += count_tmp_in_dir(&path);
            continue;
        }
        if path
            .file_name()
            .map(|n| n.to_string_lossy().ends_with(".tmp"))
            .unwrap_or(false)
        {
            count += 1;
        }
    }
    count
}

impl StorageProvider for FileStorageProvider {
    fn status(&self) -> StorageStatus {
        if self.storage_path.exists() && self.backup_path.exists() {
            StorageStatus::Online
        } else {
            StorageStatus::Offline
        }
    }

    fn health(&self) -> StorageHealth {
        let orphan_tmp = count_tmp_in_dir(&self.storage_path) + count_tmp_in_dir(&self.backup_path);
        if self.storage_path.exists() && self.backup_path.exists() {
            if orphan_tmp > 0 {
                return StorageHealth {
                    status: StorageStatus::Degraded,
                    message: Some(format!("found {orphan_tmp} orphan temp files")),
                };
            }
            return StorageHealth {
                status: StorageStatus::Online,
                message: None,
            };
        }
        StorageHealth {
            status: StorageStatus::Degraded,
            message: Some("storage or backup path is missing".to_string()),
        }
    }

    fn open(&mut self) -> StorageResult<()> {
        self.create_dirs()?;
        self.opened = true;
        Ok(())
    }

    fn close(&mut self) -> StorageResult<()> {
        self.opened = false;
        Ok(())
    }

    fn begin_transaction(&mut self) -> StorageResult<Box<dyn Transaction>> {
        if !self.opened {
            return Err(StorageError::NotAvailable);
        }
        Err(StorageError::TransactionsUnsupported)
    }

    fn list_backups(&self) -> Vec<BackupDescriptor> {
        let entries = match fs::read_dir(&self.backup_path) {
            Ok(entries) => entries,
            Err(_) => return Vec::new(),
        };
        let mut list = Vec::new();
        for entry in entries.flatten() {
            let name = entry.file_name().to_string_lossy().to_string();
            let created_at_ms = SystemTime::now()
                .duration_since(UNIX_EPOCH)
                .map(|d| d.as_millis() as u64)
                .unwrap_or(0);
            list.push(BackupDescriptor {
                name,
                created_at_ms,
            });
        }
        list
    }
}