libfw_server/

storage.rs

//! Local filesystem storage backend for `libfw-server`.
//!
//! Writes go to a temporary file next to the destination and are
//! atomically renamed into place on [`UploadSink::commit`], so a failed or
//! aborted upload never leaves a partial target behind.

use std::collections::HashMap;
use std::io::{Read, SeekFrom};
use std::path::{Component, Path, PathBuf};
use std::sync::Arc;
use std::time::{SystemTime, UNIX_EPOCH};

use tokio::sync::OwnedMutexGuard;

use async_trait::async_trait;
use libfw_core::metadata::{etag_from_size_mtime, ChunkRange, FileMeta};
use libfw_core::range::RangeSpec;
use libfw_core::storage::{DirEntry, StorageBackend, UploadSink, WriteMode};
use libfw_core::StorageError;

/// A [`StorageBackend`] rooted at a local directory.
///
/// Paths passed to the backend are treated as relative to `root`; any path
/// escaping the root (absolute, `..`, symlink-traversing) is rejected.
#[derive(Debug, Clone)]
pub struct FsStorage {
    root: PathBuf,
    /// Per-target locks (keyed by temp/target path). Concurrent chunk
    /// requests for *different* sessions/targets never serialize behind each
    /// other — only writers of the same target (who share a `.blocks`
    /// sidecar or an append-only resume file) take turns. Without this, a
    /// read-merge-write of the shared sidecar loses updates (making the
    /// commit coverage check reject a fully-written file).
    locks: Arc<std::sync::Mutex<HashMap<PathBuf, Arc<tokio::sync::Mutex<()>>>>>,
}

impl FsStorage {
    /// Create a backend serving files under `root`.
    pub fn new(root: impl Into<PathBuf>) -> Self {
        FsStorage {
            root: root.into(),
            locks: Arc::new(std::sync::Mutex::new(HashMap::new())),
        }
    }

    /// Get (or create) the mutex guarding `path`.
    fn lock_for(&self, path: &Path) -> Arc<tokio::sync::Mutex<()>> {
        let mut map = self.locks.lock().expect("lock registry poisoned");
        // Drop entries no sink holds anymore so the registry cannot grow
        // without bound across many short-lived sessions.
        if map.len() > 1024 {
            map.retain(|_, v| Arc::strong_count(v) > 1);
        }
        map.entry(path.to_path_buf())
            .or_insert_with(|| Arc::new(tokio::sync::Mutex::new(())))
            .clone()
    }

    /// Resolve a virtual path against the root (pure text — no I/O).
    ///
    /// Rejects absolute paths, `..` segments and `CurDir` (`.`) other than
    /// the root-relative segments accepted by `Component::Normal`. Symlink
    /// checking is async and is done by the callers via
    /// [`FsStorage::check_symlinks`] when they need it (file/dir open paths
    /// that may traverse existing directory structure).
    fn resolve(&self, path: &str) -> Result<PathBuf, StorageError> {
        let rel = Path::new(path);
        if rel.is_absolute() {
            return Err(StorageError::Unsupported("absolute paths are not allowed"));
        }
        let mut joined = self.root.clone();
        for component in rel.components() {
            match component {
                Component::Normal(seg) => {
                    joined.push(seg);
                }
                Component::CurDir => {}
                _ => {
                    return Err(StorageError::Unsupported(
                        "path must not contain '..' or special components",
                    ))
                }
            }
        }
        Ok(joined)
    }

    /// Walk every component of `full` between `self.root` and the leaf,
    /// rejecting any that is a symlink.
    ///
    /// Called asynchronously by the I/O methods that open existing paths so
    /// the symlink check never blocks a tokio worker thread. New (not yet
    /// existing) path tails are skipped — they cannot be symlinks.
    async fn check_symlinks(&self, full: &Path) -> Result<(), StorageError> {
        // Strip the root prefix to iterate only the virtual path components.
        let rel = match full.strip_prefix(&self.root) {
            Ok(r) => r,
            Err(_) => return Ok(()), // shouldn't happen; treat as safe
        };
        let mut probe = self.root.clone();
        for component in rel.components() {
            probe.push(component);
            match tokio::fs::symlink_metadata(&probe).await {
                Ok(m) if m.file_type().is_symlink() => {
                    return Err(StorageError::Unsupported(
                        "path must not traverse a symlink",
                    ))
                }
                Ok(_) => {}
                // Not-yet-existing component: fine (new upload target).
                Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}
                Err(e) => return Err(StorageError::Other(e)),
            }
        }
        Ok(())
    }
}

fn file_meta_at(rel: &str, meta: &std::fs::Metadata) -> FileMeta {
    let mtime = meta
        .modified()
        .ok()
        .and_then(|t| t.duration_since(UNIX_EPOCH).ok())
        .map(|d| d.as_secs())
        .unwrap_or(0);
    FileMeta {
        path: rel.to_string(),
        size: meta.len(),
        mtime,
        etag: etag_from_size_mtime(meta.len(), mtime),
    }
}

#[async_trait]
impl StorageBackend for FsStorage {
    async fn file_meta(&self, path: &str) -> Result<Option<FileMeta>, StorageError> {
        let full = self.resolve(path)?;
        self.check_symlinks(&full).await?;
        let meta = match tokio::fs::metadata(&full).await {
            Ok(m) => m,
            Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(None),
            Err(e) => return Err(StorageError::Other(e)),
        };
        if meta.is_dir() {
            return Err(StorageError::Unsupported("path is a directory"));
        }
        Ok(Some(file_meta_at(path, &meta)))
    }

    async fn read_stream(
        &self,
        path: &str,
        range: RangeSpec,
    ) -> Result<Box<dyn Read + Send>, StorageError> {
        let full = self.resolve(path)?;
        self.check_symlinks(&full).await?;
        let mut file = tokio::fs::File::open(&full)
            .await
            .map_err(|e| StorageError::Other(e))?;
        if range.start > 0 {
            tokio::io::AsyncSeekExt::seek(&mut file, SeekFrom::Start(range.start))
                .await
                .map_err(|e| StorageError::Other(e))?;
        }
        let std_file = file
            .try_into_std()
            .map_err(|e| StorageError::Other(std::io::Error::other(format!("{e:?}"))))?;
        // Restrict to exactly the requested range.
        let limited = std_file.take(range.len());
        Ok(Box::new(limited))
    }

    async fn write_stream(
        &self,
        path: &str,
        mode: WriteMode,
    ) -> Result<Box<dyn UploadSink>, StorageError> {
        let full = self.resolve(path)?;
        self.check_symlinks(&full).await?;
        if let Some(parent) = full.parent() {
            tokio::fs::create_dir_all(parent)
                .await
                .map_err(|e| StorageError::Other(e))?;
        }

        match mode {
            WriteMode::Create | WriteMode::Overwrite => {
                if mode == WriteMode::Create
                    && tokio::fs::try_exists(&full)
                        .await
                        .map_err(|e| StorageError::Other(e))?
                {
                    return Err(StorageError::AlreadyExists(path.to_string()));
                }
                // Write to a temp file, rename on commit.
                let tmp = temp_path_for(&full);
                let file = tokio::fs::File::create(&tmp)
                    .await
                    .map_err(|e| StorageError::Other(e))?;
                Ok(Box::new(FsSink {
                    file,
                    tmp: Some(tmp),
                    target: full,
                    rel: path.to_string(),
                    mode,
                    written: 0,
                    blocks_path: None,
                    ranges: Vec::new(),
                    target_guard: None,
                }))
            }
            WriteMode::Resume { offset } => {
                // Concurrent `Resume` writers of the same target would both
                // observe the same on-disk length and interleave their
                // appends. Hold the per-target lock for the sink's whole
                // lifetime so they take turns (the loser re-checks the
                // length only after the winner's commit/rejection).
                let target_guard = Some(self.lock_for(&full).lock_owned().await);
                let file = tokio::fs::OpenOptions::new()
                    .write(true)
                    .append(true)
                    .open(&full)
                    .await
                    .map_err(|e| StorageError::Other(e))?;
                let current = file
                    .metadata()
                    .await
                    .map_err(|e| StorageError::Other(e))?
                    .len();
                if current != offset {
                    return Err(StorageError::write_failed(
                        offset,
                        std::io::Error::other(format!(
                            "existing file is {current} bytes, expected {offset}"
                        )),
                    ));
                }
                Ok(Box::new(FsSink {
                    file,
                    tmp: None,
                    target: full,
                    rel: path.to_string(),
                    mode,
                    written: offset,
                    blocks_path: None,
                    ranges: Vec::new(),
                    target_guard,
                }))
            }
        }
    }

    async fn write_stream_session(
        &self,
        path: &str,
        session: &str,
        owner: &str,
        mode: WriteMode,
    ) -> Result<Box<dyn UploadSink>, StorageError> {
        let full = self.resolve(path)?;
        self.check_symlinks(&full).await?;
        if let Some(parent) = full.parent() {
            tokio::fs::create_dir_all(parent)
                .await
                .map_err(|e| StorageError::Other(e))?;
        }
        // Both the session id and owner tag are embedded in the temp filename,
        // so neither must inject path separators or `..`. Restrict to safe
        // chars (alphanumeric, `-`, `_`).
        let sanitize = |s: &str| -> String {
            s.chars()
                .map(|c| {
                    if c.is_ascii_alphanumeric() || c == '-' || c == '_' {
                        c
                    } else {
                        '_'
                    }
                })
                .collect()
        };
        let safe_session = sanitize(session);
        // Embed the owner tag so that two authenticated users with the same
        // file path and an identical (guessed/observed) session id cannot
        // collide on the same temp file.  A short prefix of the owner hash
        // is enough for discrimination without making filenames unwieldy.
        let safe_owner = sanitize(&owner.chars().take(16).collect::<String>());
        let name = full
            .file_name()
            .map(|n| n.to_string_lossy().to_string())
            .unwrap_or_else(|| "upload".to_string());
        let tmp = full.with_file_name(format!(".libfw-sess-{safe_owner}-{safe_session}-{name}"));

        let exists = tokio::fs::try_exists(&tmp)
            .await
            .map_err(|e| StorageError::Other(e))?;
        let file = if exists {
            // Subsequent chunk / resume of an in-flight session: open the
            // shared temp for positional (seek + write) access. `mode` is
            // ignored; the already-received ranges are reloaded from the
            // sidecar so the client can resume only the missing parts.
            tokio::fs::OpenOptions::new()
                .read(true)
                .write(true)
                .open(&tmp)
                .await
                .map_err(|e| StorageError::Other(e))?
        } else {
            // First request for this session: create the shared temp.
            match mode {
                WriteMode::Create | WriteMode::Overwrite => {
                    if mode == WriteMode::Create
                        && tokio::fs::try_exists(&full)
                            .await
                            .map_err(|e| StorageError::Other(e))?
                    {
                        return Err(StorageError::AlreadyExists(path.to_string()));
                    }
                    tokio::fs::OpenOptions::new()
                        .write(true)
                        .create_new(true)
                        .open(&tmp)
                        .await
                        .map_err(|e| StorageError::Other(e))?
                }
                // Resumable sessions are driven by the per-block probe (the
                // client asks "what ranges do you have?" and sends only the
                // gaps), not by a contiguous offset append, so a
                // `WriteMode::Resume` request has no meaning here.
                WriteMode::Resume { .. } => {
                    return Err(StorageError::Unsupported(
                        "session upload does not support contiguous resume; use the block probe".into(),
                    ))
                }
            }
        };
        // Chunks of the same session share the temp + sidecar, so they
        // serialize on the temp's per-target lock (different sessions stay
        // fully parallel). The guard is held for the sink's whole lifetime:
        // concurrent first requests would otherwise both load the (missing)
        // sidecar, then each chunk write's read-merge-write would clobber the
        // other's received-range bookkeeping.
        let lock = self.lock_for(&tmp);
        let target_guard = Some(lock.clone().lock_owned().await);

        // Load any already-received byte ranges from the sidecar so a pause /
        // resume only re-sends the missing blocks.
        let blocks_path = blocks_path_for(&tmp);
        // A freshly-created temp means nothing has been received yet: a stale
        // `.blocks` sidecar left behind by an earlier aborted attempt must not
        // make the probe report phantom ranges (the client would then skip
        // chunks and commit an empty / zero-filled file).
        if !exists {
            let _ = tokio::fs::remove_file(&blocks_path).await;
        }
        let ranges = if exists {
            read_ranges(&blocks_path).await?
        } else {
            Vec::new()
        };
        Ok(Box::new(FsSink {
            file,
            tmp: Some(tmp),
            target: full,
            rel: path.to_string(),
            mode,
            written: 0,
            blocks_path: Some(blocks_path),
            ranges,
            target_guard,
        }))
    }

    async fn list_dir(&self, path: &str) -> Result<Vec<DirEntry>, StorageError> {
        let full = if path.is_empty() {
            self.root.clone()
        } else {
            self.resolve(path)?
        };
        self.check_symlinks(&full).await?;
        let mut entries = Vec::new();
        let mut read_dir = tokio::fs::read_dir(&full)
            .await
            .map_err(|e| {
                if e.kind() == std::io::ErrorKind::NotFound {
                    StorageError::NotFound(path.to_string())
                } else {
                    StorageError::Other(e)
                }
            })?;
        while let Some(entry) = read_dir
            .next_entry()
            .await
            .map_err(|e| StorageError::Other(e))?
        {
            let name = entry.file_name().to_string_lossy().to_string();
            let rel = if path.is_empty() {
                name.clone()
            } else {
                format!("{path}/{name}")
            };
            let meta = entry
                .metadata()
                .await
                .map_err(|e| StorageError::Other(e))?;
            let mtime = meta
                .modified()
                .ok()
                .and_then(|t| t.duration_since(UNIX_EPOCH).ok())
                .map(|d| d.as_secs())
                .unwrap_or(0);
            entries.push(DirEntry {
                path: rel,
                is_dir: meta.is_dir(),
                size: if meta.is_dir() { 0 } else { meta.len() },
                mtime,
            });
        }
        entries.sort_by(|a, b| a.path.cmp(&b.path));
        Ok(entries)
    }

    async fn mkdir_all(&self, path: &str) -> Result<(), StorageError> {
        let full = self.resolve(path)?;
        self.check_symlinks(&full).await?;
        tokio::fs::create_dir_all(&full)
            .await
            .map_err(|e| StorageError::Other(e))?;
        Ok(())
    }

    async fn remove(&self, path: &str) -> Result<(), StorageError> {
        let full = self.resolve(path)?;
        self.check_symlinks(&full).await?;
        // Use `symlink_metadata` (not `metadata`) so a symlink that appeared
        // since `resolve` checked is never followed — we refuse to remove
        // *through* it. (The check-then-use race cannot be fully closed on
        // all platforms without openat/O_NOFOLLOW; this narrows it for the
        // destructive operation.)
        let meta = tokio::fs::symlink_metadata(&full)
            .await
            .map_err(|e| StorageError::Other(e))?;
        if meta.file_type().is_symlink() {
            return Err(StorageError::Unsupported(
                "refusing to remove through a symlink",
            ));
        }
        if meta.is_dir() {
            tokio::fs::remove_dir_all(&full)
                .await
                .map_err(|e| StorageError::Other(e))
        } else {
            tokio::fs::remove_file(&full)
                .await
                .map_err(|e| StorageError::Other(e))
        }
    }

    async fn cleanup_stale_sessions(
        &self,
        max_age: std::time::Duration,
    ) -> Result<usize, StorageError> {
        // tus-style expiry: a client that vanishes mid-upload leaves its
        // `.libfw-sess-<id>-<name>` temp (and `.blocks` sidecar) behind. Walk
        // the root, removing the ones whose last write is older than
        // `max_age`. Only session temps are touched — never committed user
        // files — and symlinked directories are never followed.
        let deadline = SystemTime::now()
            .checked_sub(max_age)
            .unwrap_or(UNIX_EPOCH);
        let mut removed = 0usize;
        let mut stack = vec![self.root.clone()];
        while let Some(dir) = stack.pop() {
            let mut rd = match tokio::fs::read_dir(&dir).await {
                Ok(rd) => rd,
                Err(_) => continue,
            };
            while let Ok(Some(entry)) = rd.next_entry().await {
                let ft = match entry.file_type().await {
                    Ok(ft) => ft,
                    Err(_) => continue,
                };
                // On Linux/macOS `is_dir()` and `is_symlink()` are mutually
                // exclusive: a symlink to a directory reports
                // `is_symlink()=true`, `is_dir()=false`. The symlink check
                // below therefore rules out every symlinked directory
                // before the walk descends into it, and the two tests can
                // stay independent.
                if ft.is_symlink() {
                    // Never follow symlinks into directories.
                    continue;
                }
                if ft.is_dir() {
                    stack.push(entry.path());
                    continue;
                }
                let name = entry.file_name().to_string_lossy().to_string();
                if !name.starts_with(".libfw-sess-") {
                    continue;
                }
                let modified = entry
                    .metadata()
                    .await
                    .ok()
                    .and_then(|m| m.modified().ok())
                    .and_then(|t| t.duration_since(UNIX_EPOCH).ok())
                    .map(|d| UNIX_EPOCH + d)
                    .unwrap_or(UNIX_EPOCH);
                if modified < deadline {
                    // A sidecar is itself a `.libfw-sess-*` entry, so the walk
                    // reaches it too: it goes with its temp, and an orphaned
                    // one is still swept on its own pass. Only a temp counts,
                    // keeping the result a number of sessions.
                    let is_sidecar = name.ends_with(".blocks");
                    let mut sidecar = entry.path().as_os_str().to_owned();
                    sidecar.push(".blocks");
                    let _ = tokio::fs::remove_file(PathBuf::from(sidecar)).await;
                    if tokio::fs::remove_file(entry.path()).await.is_ok() && !is_sidecar {
                        removed += 1;
                    }
                }
            }
        }
        Ok(removed)
    }
}

/// A temporary path for a target, unique per attempt.
///
/// Uniqueness combines a monotonic counter with a timestamp so two
/// concurrent Create-mode uploads to the same target never collide on the
/// same temp file (which would otherwise interleave/truncate each other).
static TMP_COUNTER: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);

fn temp_path_for(target: &Path) -> PathBuf {
    let nanos = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map(|d| d.as_nanos())
        .unwrap_or(0);
    let counter = TMP_COUNTER.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
    let file_name = target
        .file_name()
        .map(|n| n.to_string_lossy().to_string())
        .unwrap_or_else(|| "upload".to_string());
    let tmp_name = format!(".libfw-tmp-{file_name}-{nanos}-{counter}");
    target.with_file_name(tmp_name)
}

/// Streaming write handle for the filesystem backend.
pub struct FsSink {
    file: tokio::fs::File,
    tmp: Option<PathBuf>,
    target: PathBuf,
    /// The full virtual path (used for the committed `FileMeta`).
    rel: String,
    /// The mode the sink was opened with (used for the Create atomicity
    /// check at commit time).
    mode: WriteMode,
    written: u64,
    /// Optional sidecar path tracking received byte ranges for a resumable
    /// "session" upload (parallel to the session temp file). `None` for
    /// ordinary (non-session) sinks.
    blocks_path: Option<PathBuf>,
    /// In-memory copy of the received ranges (kept in sync with
    /// `blocks_path`).
    ranges: Vec<ChunkRange>,
    /// Per-target guard held for the sink's whole lifetime. `Some` only for
    /// session sinks (serializing sidecar read-merge-write per target) and
    /// for `WriteMode::Resume` (serializing concurrent appends to the same
    /// target). `None` for plain Create/Overwrite sinks, which write to a
    /// unique temp and rename on commit — no shared state to guard.
    target_guard: Option<OwnedMutexGuard<()>>,
}

/// Merge `new` (a `[start, end)` half-open range) into a sorted, disjoint
/// list of ranges, coalescing overlaps/adjacencies. Returns the updated list.
fn merge_range(ranges: &mut Vec<ChunkRange>, new: ChunkRange) {
    if new.is_empty() {
        return;
    }
    ranges.push(new);
    ranges.sort_by_key(|r| r.start);
    let mut merged: Vec<ChunkRange> = Vec::with_capacity(ranges.len());
    for r in ranges.drain(..) {
        if let Some(last) = merged.last_mut() {
            // Overlapping or adjacent ranges coalesce.
            if r.start <= last.end {
                if r.end > last.end {
                    last.end = r.end;
                }
                continue;
            }
        }
        merged.push(r);
    }
    *ranges = merged;
}

#[async_trait]
impl UploadSink for FsSink {
    async fn write(&mut self, buf: &[u8]) -> Result<(), StorageError> {
        use tokio::io::AsyncWriteExt;
        self.file
            .write_all(buf)
            .await
            .map_err(|e| StorageError::write_failed(self.written, e))?;
        self.written += buf.len() as u64;
        Ok(())
    }

    async fn write_at(&mut self, offset: u64, buf: &[u8]) -> Result<(), StorageError> {
        use tokio::io::{AsyncSeekExt, AsyncWriteExt};
        self.file
            .seek(SeekFrom::Start(offset))
            .await
            .map_err(|e| StorageError::write_failed(offset, e))?;
        self.file
            .write_all(buf)
            .await
            .map_err(|e| StorageError::write_failed(offset, e))?;
        // Track the highest extent written (used by `len`-style bookkeeping;
        // the real size for commit validation comes from file metadata).
        self.written = self.written.max(offset.saturating_add(buf.len() as u64));
        // For a resumable session sink, record the received byte range in the
        // sidecar so a later probe / resume knows this part is already on
        // disk and only missing gaps need to be re-sent.
        //
        // The read-merge-write is serialized by the per-target guard this
        // sink holds for its whole lifetime: concurrent chunks of the SAME
        // session each hold a *stale* in-memory `ranges` copy loaded when
        // their sink was opened, so a bare `persist_ranges` would clobber
        // other chunks' ranges (lost update) and make the commit coverage
        // check reject a fully-written file. Re-reading under the guard
        // keeps every writer's merge based on the latest on-disk state.
        if let Some(blocks) = self.blocks_path.clone() {
            let _guard = self.target_guard.as_ref().expect("session sink holds a guard");
            let mut current = read_ranges(&blocks).await?;
            merge_range(&mut current, ChunkRange {
                start: offset,
                end: offset.saturating_add(buf.len() as u64),
            });
            persist_ranges(&blocks, &current).await?;
            self.ranges = current;
        }
        Ok(())
    }

    async fn received_ranges(&mut self) -> Result<Vec<ChunkRange>, StorageError> {
        Ok(self.ranges.clone())
    }

    async fn len(&self) -> Result<u64, StorageError> {
        self.file
            .metadata()
            .await
            .map(|m| m.len())
            .map_err(|e| StorageError::Other(e))
    }

    async fn commit(self: Box<Self>) -> Result<FileMeta, StorageError> {
        use tokio::io::AsyncWriteExt;
        let FsSink {
            mut file,
            tmp,
            target,
            rel,
            mode,
            blocks_path,
            ..
        } = *self;
        let _ = remove_blocks_sidecar(blocks_path.as_deref()).await;
        file.flush().await.map_err(|e| StorageError::Other(e))?;
        file.sync_all().await.map_err(|e| StorageError::Other(e))?;
        drop(file);
        if let Some(tmp) = tmp {
            // Create mode must never clobber a target that appeared while
            // we were streaming (closes the check-then-rename TOCTOU for
            // the common non-concurrent-writer case).
            if matches!(mode, WriteMode::Create)
                && tokio::fs::try_exists(&target)
                    .await
                    .map_err(|e| StorageError::Other(e))?
            {
                let _ = tokio::fs::remove_file(&tmp).await;
                return Err(StorageError::AlreadyExists(rel));
            }
            tokio::fs::rename(&tmp, &target)
                .await
                .map_err(|e| StorageError::Other(e))?;
            // Durability: fsync the parent directory so the rename itself
            // survives a crash.
            if let Some(parent) = target.parent() {
                if let Ok(d) = tokio::fs::File::open(parent).await {
                    let _ = d.sync_all().await;
                }
            }
        }
        let meta = tokio::fs::metadata(&target)
            .await
            .map_err(|e| StorageError::Other(e))?;
        Ok(file_meta_at(&rel, &meta))
    }

    async fn abort(self: Box<Self>) -> Result<(), StorageError> {
        let FsSink {
            file,
            tmp,
            blocks_path,
            ..
        } = *self;
        drop(file);
        if let Some(tmp) = tmp {
            let _ = tokio::fs::remove_file(&tmp).await;
        }
        let _ = remove_blocks_sidecar(blocks_path.as_deref()).await;
        Ok(())
    }
}

/// Sidecar filename for a session temp (e.g. `<temp>.blocks`).
fn blocks_path_for(tmp: &Path) -> PathBuf {
    let mut name = tmp.as_os_str().to_owned();
    name.push(".blocks");
    PathBuf::from(name)
}

/// Read the persisted received ranges for a session temp sidecar.
///
/// An invalid or tampered sidecar is a protocol error: the server must refuse
/// to continue instead of silently treating the upload as if it had no
/// received data and accidentally re-sending/overwriting valid chunks.
async fn read_ranges(blocks: &Path) -> Result<Vec<ChunkRange>, StorageError> {
    match tokio::fs::read_to_string(blocks).await {
        Ok(text) => serde_json::from_str::<Vec<ChunkRange>>(&text).map_err(|e| {
            // This message reaches clients as a 500 body, so it must not
            // disclose server-side filesystem paths.
            StorageError::Other(std::io::Error::other(format!(
                "corrupt session state: {e}"
            )))
        }),
        Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(Vec::new()),
        Err(e) => Err(StorageError::Other(e)),
    }
}

/// Persist the received byte ranges to a session temp sidecar.
///
/// Write-then-rename: [`read_ranges`] treats a malformed sidecar as fatal, so
/// a torn write (crash or error mid-update) must never be observable.
/// Same-directory rename is atomic on POSIX and replaces on Windows; the
/// `.tmp` sibling is itself a `.libfw-sess-*` file, so the stale-session
/// sweeper collects any leftovers.
async fn persist_ranges(blocks: &Path, ranges: &[ChunkRange]) -> Result<(), StorageError> {
    let text = serde_json::to_string(ranges).unwrap_or_else(|_| "[]".to_string());
    let mut tmp = blocks.as_os_str().to_owned();
    tmp.push(".tmp");
    let tmp = PathBuf::from(tmp);
    tokio::fs::write(&tmp, text)
        .await
        .map_err(StorageError::Other)?;
    tokio::fs::rename(&tmp, blocks)
        .await
        .map_err(StorageError::Other)?;
    Ok(())
}

/// Remove a session temp sidecar (best-effort; missing file is fine).
async fn remove_blocks_sidecar(blocks: Option<&Path>) -> Result<(), StorageError> {
    if let Some(blocks) = blocks {
        let _ = tokio::fs::remove_file(blocks).await;
    }
    Ok(())
}

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

    /// Force a file's mtime (used to age a fake session temp).
    fn filetime_set(path: &Path, time: std::time::SystemTime) -> std::io::Result<()> {
        // Open writable: on Windows, setting file times through a read-only
        // handle is refused.
        let file = std::fs::OpenOptions::new().write(true).open(path)?;
        file.set_times(std::fs::FileTimes::new().set_modified(time))
    }

    #[tokio::test]
    async fn write_read_roundtrip() {
        let dir = tempfile::tempdir().unwrap();
        let storage = FsStorage::new(dir.path());

        let sink = storage
            .write_stream("a/b.txt", WriteMode::Create)
            .await
            .unwrap();
        let mut sink = sink;
        sink.write(b"hello ").await.unwrap();
        sink.write(b"world").await.unwrap();
        let meta = sink.commit().await.unwrap();
        assert_eq!(meta.size, 11);

        let got = storage.file_meta("a/b.txt").await.unwrap().unwrap();
        assert_eq!(got.size, 11);

        let mut reader = storage
            .read_stream("a/b.txt", RangeSpec::new(0, 11))
            .await
            .unwrap();
        let mut buf = Vec::new();
        reader.read_to_end(&mut buf).unwrap();
        assert_eq!(buf, b"hello world");
    }

    #[tokio::test]
    async fn create_rejects_existing() {
        let dir = tempfile::tempdir().unwrap();
        let storage = FsStorage::new(dir.path());
        let sink = storage.write_stream("f.txt", WriteMode::Create).await.unwrap();
        let mut sink = sink;
        sink.write(b"x").await.unwrap();
        sink.commit().await.unwrap();

        let res = storage.write_stream("f.txt", WriteMode::Create).await;
        assert!(matches!(res, Err(StorageError::AlreadyExists(_))));
    }

    #[tokio::test]
    async fn resume_writes_at_offset() {
        let dir = tempfile::tempdir().unwrap();
        let storage = FsStorage::new(dir.path());

        let sink = storage.write_stream("f.txt", WriteMode::Create).await.unwrap();
        let mut sink = sink;
        sink.write(b"ABCD").await.unwrap();
        sink.commit().await.unwrap();

        let mut sink = storage
            .write_stream("f.txt", WriteMode::Resume { offset: 4 })
            .await
            .unwrap();
        sink.write(b"EF").await.unwrap();
        sink.commit().await.unwrap();

        let mut reader = storage
            .read_stream("f.txt", RangeSpec::full(6))
            .await
            .unwrap();
        let mut buf = Vec::new();
        reader.read_to_end(&mut buf).unwrap();
        assert_eq!(buf, b"ABCDEF");
    }

    #[tokio::test]
    async fn resume_offset_mismatch_fails() {
        let dir = tempfile::tempdir().unwrap();
        let storage = FsStorage::new(dir.path());
        let sink = storage.write_stream("f.txt", WriteMode::Create).await.unwrap();
        let mut sink = sink;
        sink.write(b"AB").await.unwrap();
        sink.commit().await.unwrap();

        let res = storage.write_stream("f.txt", WriteMode::Resume { offset: 9 }).await;
        assert!(res.is_err());
    }

    #[tokio::test]
    async fn abort_leaves_no_partial_file() {
        let dir = tempfile::tempdir().unwrap();
        let storage = FsStorage::new(dir.path());
        let sink = storage.write_stream("f.txt", WriteMode::Create).await.unwrap();
        let mut sink = sink;
        sink.write(b"partial").await.unwrap();
        sink.abort().await.unwrap();
        assert!(!dir.path().join("f.txt").exists());
    }

    #[tokio::test]
    async fn list_dir_and_remove() {
        let dir = tempfile::tempdir().unwrap();
        let storage = FsStorage::new(dir.path());
        std::fs::create_dir_all(dir.path().join("sub")).unwrap();
        std::fs::write(dir.path().join("sub/x.txt"), b"1").unwrap();
        std::fs::write(dir.path().join("a.txt"), b"22").unwrap();

        let entries = storage.list_dir("").await.unwrap();
        assert_eq!(entries.len(), 2);
        let names: Vec<_> = entries.iter().map(|e| e.path.as_str()).collect();
        assert_eq!(names, vec!["a.txt", "sub"]);

        storage.remove("sub").await.unwrap();
        assert!(!dir.path().join("sub").exists());
    }

    #[tokio::test]
    async fn rejects_path_traversal() {
        let dir = tempfile::tempdir().unwrap();
        let storage = FsStorage::new(dir.path());
        assert!(storage.file_meta("../etc/passwd").await.is_err());
        assert!(storage.file_meta("/etc/passwd").await.is_err());
    }

    #[tokio::test]
    async fn corrupt_session_sidecar_is_rejected() {
        let dir = tempfile::tempdir().unwrap();
        let storage = FsStorage::new(dir.path());
        let tmp = dir.path().join(".libfw-sess-owner-session-f.txt");
        std::fs::write(&tmp, b"partial").unwrap();
        std::fs::write(tmp.with_extension("txt.blocks"), b"{not-json}").unwrap();

        let res = storage
            .write_stream_session("f.txt", "session", "owner", WriteMode::Create)
            .await;
        let err = match res {
            Ok(_) => panic!("malformed sidecar should not be accepted"),
            Err(e) => e,
        };
        // The message is surfaced to clients as a 500 body — it must not leak
        // the server-side storage root.
        let msg = err.to_string();
        assert!(msg.contains("corrupt session state"), "{msg}");
        assert!(
            !msg.contains(&dir.path().display().to_string()),
            "error leaked the server path: {msg}"
        );
    }

    #[tokio::test]
    async fn session_sidecar_is_written_atomically() {
        let dir = tempfile::tempdir().unwrap();
        let storage = FsStorage::new(dir.path());
        let mut sink = storage
            .write_stream_session("f.txt", "session", "owner", WriteMode::Create)
            .await
            .unwrap();

        sink.write_at(4, b"abcd").await.unwrap();
        assert_eq!(
            sink.received_ranges().await.unwrap(),
            vec![ChunkRange { start: 4, end: 8 }]
        );
        sink.write_at(8, b"ef").await.unwrap();

        // The sidecar always holds complete, parseable JSON…
        let blocks = dir.path().join(".libfw-sess-owner-session-f.txt.blocks");
        let text = std::fs::read_to_string(&blocks).unwrap();
        let ranges: Vec<ChunkRange> = serde_json::from_str(&text).unwrap();
        assert_eq!(ranges, vec![ChunkRange { start: 4, end: 10 }]);
        // …and the write-then-rename leaves no temporary sibling behind.
        assert!(!dir
            .path()
            .join(".libfw-sess-owner-session-f.txt.blocks.tmp")
            .exists());
        sink.abort().await.unwrap();
    }

    #[tokio::test]
    async fn cleanup_stale_sessions_removes_old_temps_only() {
        let dir = tempfile::tempdir().unwrap();
        let storage = FsStorage::new(dir.path());

        // A stale session temp (very old mtime) + its range sidecar.
        let stale = dir.path().join(".libfw-sess-oldid-a.bin");
        std::fs::write(&stale, b"partial").unwrap();
        std::fs::write(dir.path().join(".libfw-sess-oldid-a.bin.blocks"), b"[[0,7]]").unwrap();
        // Age it well beyond the 1-hour TTL (7 days old). Windows/FAT can
        // clamp very old timestamps, so use a recent-but-stale date.
        let old = std::time::SystemTime::now()
            .checked_sub(std::time::Duration::from_secs(7 * 24 * 3600))
            .unwrap();
        assert!(
            filetime_set(&stale, old).is_ok(),
            "failed to age the stale session temp"
        );

        // A fresh session temp (now) must be kept.
        let fresh = dir.path().join(".libfw-sess-newid-a.bin");
        std::fs::write(&fresh, b"partial").unwrap();

        // A committed user file must never be touched.
        std::fs::write(dir.path().join("real.txt"), b"real").unwrap();

        let removed = storage
            .cleanup_stale_sessions(std::time::Duration::from_secs(3600))
            .await
            .unwrap();
        assert_eq!(removed, 1, "only the stale temp is removed");
        assert!(!stale.exists());
        assert!(!dir.path().join(".libfw-sess-oldid-a.bin.blocks").exists());
        assert!(fresh.exists(), "fresh temp survives");
        assert!(dir.path().join("real.txt").exists(), "user file survives");
    }

    #[tokio::test]
    async fn cleanup_stale_sessions_sweeps_orphaned_sidecars_without_counting_them() {
        let dir = tempfile::tempdir().unwrap();
        let storage = FsStorage::new(dir.path());

        // A sidecar whose temp is already gone (a crash between the two
        // writes): it must still be swept, or it leaks for the process
        // lifetime, but it is not a session in its own right.
        let orphan = dir.path().join(".libfw-sess-goneid-a.bin.blocks");
        std::fs::write(&orphan, b"[[0,7]]").unwrap();
        let old = std::time::SystemTime::now()
            .checked_sub(std::time::Duration::from_secs(7 * 24 * 3600))
            .unwrap();
        assert!(filetime_set(&orphan, old).is_ok(), "failed to age the orphan");

        // A fresh orphan must survive until it goes stale.
        let fresh_orphan = dir.path().join(".libfw-sess-newid-a.bin.blocks");
        std::fs::write(&fresh_orphan, b"[[0,7]]").unwrap();

        let removed = storage
            .cleanup_stale_sessions(std::time::Duration::from_secs(3600))
            .await
            .unwrap();
        assert_eq!(removed, 0, "an orphaned sidecar is not a session");
        assert!(!orphan.exists(), "the stale orphan is swept");
        assert!(fresh_orphan.exists(), "fresh orphan survives");
    }

    #[tokio::test]
    async fn committed_meta_has_full_relative_path() {
        let dir = tempfile::tempdir().unwrap();
        let storage = FsStorage::new(dir.path());
        let sink = storage
            .write_stream("deep/nested/f.txt", WriteMode::Create)
            .await
            .unwrap();
        let mut sink = sink;
        sink.write(b"hi").await.unwrap();
        let meta = sink.commit().await.unwrap();
        assert_eq!(meta.path, "deep/nested/f.txt");
    }

    #[cfg(unix)]
    #[tokio::test]
    async fn rejects_symlink_traversal() {
        use std::os::unix::fs::symlink;

        let dir = tempfile::tempdir().unwrap();
        let storage = FsStorage::new(dir.path());

        // A symlink planted inside the root pointing outside it must not let
        // reads/writes escape the mount root.
        let outside = tempfile::tempdir().unwrap();
        std::fs::write(outside.path().join("secret.txt"), b"top-secret").unwrap();
        symlink(outside.path(), dir.path().join("evil")).unwrap();

        assert!(storage.read_stream("evil/secret.txt", RangeSpec::full(10)).await.is_err());
        assert!(storage.file_meta("evil/secret.txt").await.is_err());
        assert!(storage.write_stream("evil/new.txt", WriteMode::Create).await.is_err());
    }
}