//! Cached view of the filesystem backing the media directory.
//!
//! Free/total disk space is needed on two hot paths — the per-user storage quota
//! (checked on every upload *and* every `GET /me/quota` poll) and the admin stats
//! endpoint. Reading it means `sysinfo::Disks::new_with_refreshed_list()`, which stats
//! every mounted filesystem; doing that per request is wasteful for a number that
//! barely moves. [`DiskCache`] refreshes it at most once per [`TTL`] and serves the
//! rest from memory.

use std::collections::HashMap;
use std::path::{Path, PathBuf};
use std::sync::{Arc, RwLock};
use std::time::{Duration, Instant};

/// How long a disk reading is trusted before the next call re-stats the filesystem.
const TTL: Duration = Duration::from_secs(15);

#[derive(Clone, Copy)]
pub struct DiskInfo {
    pub total: u64,
    pub free: u64,
}

/// Cheap-to-clone cache of a filesystem's total/free bytes, keyed by the path asked about.
/// Lives in `AppState`.
///
/// ONE ENTRY PER PATH, not one entry total. This held a single slot carrying its own key, so a
/// lookup for a different path was a miss that OVERWROTE the previous reading — and the app asks
/// about two paths that are distinct mounts in production (`MEDIA_PATH=/media` and
/// `EXPORT_PATH=/exports`, docker-compose.yml). The upload gate and the per-user quota ask about
/// the media volume on every single photo; `host::get_event_status` asks about the exports volume
/// on every host-dashboard load. So while a host had the dashboard open the two evicted each
/// other and the cache hit rate collapsed to zero, putting an uncached
/// `sysinfo::Disks::new_with_refreshed_list()` — a synchronous scan of every mount, on the async
/// runtime — back on the busiest write path in the app. That is precisely the cost the comment in
/// `handlers::upload` says this cache exists to avoid, on the 2-vCPU box it says it matters on.
///
/// The key space cannot grow: both paths come from `AppConfig`, never from request input, so this
/// map holds at most as many entries as there are configured volumes.
#[derive(Clone)]
pub struct DiskCache {
    inner: Arc<RwLock<HashMap<PathBuf, (DiskInfo, Instant)>>>,
}

impl DiskCache {
    pub fn new() -> Self {
        Self {
            inner: Arc::new(RwLock::new(HashMap::new())),
        }
    }

    /// Drop every cached reading so the next `snapshot()` re-measures the filesystem.
    ///
    /// Used by the e2e TRUNCATE endpoint. Truncating deletes every uploaded file, which materially
    /// changes free space — but the cached reading survives for up to the TTL, so the next test can
    /// compute a quota from the PREVIOUS test's disk. That matters now that the quota tests steer
    /// the per-user limit off `free_disk_bytes`: a stale reading makes the limit wrong and the test
    /// flaky, for reasons that have nothing to do with the code under test.
    ///
    /// Clears ALL paths, not just the media one: truncation moves free space on every volume the
    /// app measures, and the export volume feeds the host dashboard's low-disk banner.
    pub fn invalidate(&self) {
        self.inner.write().unwrap().clear();
    }

    /// Cached `(total, free)` bytes for the filesystem backing `path`.
    ///
    /// Returns `None` when the mount can't be resolved — callers MUST treat that as
    /// "unknown", never "zero free". (The quota path in particular fails *open* on
    /// `None`: enforcing a 0-byte limit would lock every user out of uploading.)
    ///
    /// The cache is keyed BY PATH, and it RETAINS every path it has been asked about — see the
    /// struct doc for why holding one slot made the cache stop caching as soon as the app asked
    /// about its second volume.
    ///
    /// (The original single slot also carried its key, so it never returned the WRONG
    /// filesystem's numbers. What it did instead was miss and re-measure on every alternation,
    /// which is the quieter failure and the one that cost.)
    pub fn snapshot(&self, path: &Path) -> Option<DiskInfo> {
        if let Some((info, at)) = self.inner.read().unwrap().get(path)
            && at.elapsed() < TTL
        {
            return Some(*info);
        }
        let info = read_disk_for_path(path)?;
        self.inner
            .write()
            .unwrap()
            .insert(path.to_path_buf(), (info, Instant::now()));
        Some(info)
    }
}

impl Default for DiskCache {
    fn default() -> Self {
        Self::new()
    }
}

/// UNCACHED free-space reading for the filesystem backing `path`.
///
/// Deliberately bypasses [`DiskCache`]. The cache exists for the quota poll, where a 15s-stale
/// number is fine because it is only ever advisory. The export preflight is the opposite case: it
/// decides whether to start writing a multi-GB archive, and the sibling export worker running
/// concurrently can move free space by tens of gigabytes well inside the TTL. A stale reading there
/// would authorise exactly the write that fills the disk.
pub fn free_bytes(path: &Path) -> Option<u64> {
    read_disk_for_path(path).map(|d| d.free)
}

/// Resolve the filesystem backing `media_path` and read its total/free bytes.
///
/// Snapshots the mount table via `sysinfo`, then delegates the selection to the pure
/// [`select_disk`] so the (fiddly, edge-case-prone) matching logic is unit-testable
/// without touching the real filesystem.
fn read_disk_for_path(media_path: &Path) -> Option<DiskInfo> {
    let disks = sysinfo::Disks::new_with_refreshed_list();
    let mounts: Vec<(String, u64, u64)> = disks
        .iter()
        .map(|d| {
            (
                d.mount_point().to_string_lossy().to_string(),
                d.total_space(),
                d.available_space(),
            )
        })
        .collect();
    select_disk(&mounts, &media_path.to_string_lossy())
}

/// Pick the filesystem for `media_path` from a `(mount_point, total, free)` table.
///
/// Chooses the **longest** mount point that is a prefix of `media_path` (the most
/// specific filesystem) rather than the first match — otherwise the root `/` mount,
/// which prefixes every absolute path, could shadow a dedicated `/media` volume. Falls
/// back to `/` when nothing prefixes the path (e.g. a relative media path), and to
/// `None` when even that is absent — the caller treats `None` as "unknown" and fails
/// open on quota.
fn select_disk(mounts: &[(String, u64, u64)], media_path: &str) -> Option<DiskInfo> {
    mounts
        .iter()
        .filter(|(mp, _, _)| media_path.starts_with(mp.as_str()))
        .max_by_key(|(mp, _, _)| mp.len())
        .or_else(|| mounts.iter().find(|(mp, _, _)| mp == "/"))
        .map(|(_, total, free)| DiskInfo {
            total: *total,
            free: *free,
        })
}

#[cfg(test)]
mod tests {
    use super::{DiskCache, select_disk};

    /// THE REGRESSION: two paths must not evict each other.
    ///
    /// `MEDIA_PATH=/media` and `EXPORT_PATH=/exports` are separate mounts in production. With a
    /// single-slot cache, every host-dashboard load (exports) threw away the reading the upload
    /// gate and the per-user quota depend on (media), and the next upload threw the exports one
    /// back — so during exactly the window a host is watching the disk, every photo paid for a
    /// full `sysinfo` mount scan on the async runtime.
    ///
    /// Asserted on the map's occupancy rather than on timing, so it fails deterministically.
    #[test]
    fn asking_about_a_second_volume_does_not_evict_the_first() {
        let cache = DiskCache::new();
        // Two paths that exist on any host this runs on. They may well resolve to the SAME
        // filesystem — irrelevant here: what is under test is that each keeps its own entry.
        let media = std::path::Path::new("/");
        let exports = std::env::temp_dir();

        assert!(cache.snapshot(media).is_some(), "root must be resolvable");
        assert!(
            cache.snapshot(&exports).is_some(),
            "the temp dir must be resolvable"
        );

        let held = cache.inner.read().unwrap();
        assert!(
            held.contains_key(media),
            "the media reading was evicted by a lookup for another volume — the upload gate is \
             back to an uncached mount scan per photo"
        );
        assert!(held.contains_key(exports.as_path()));
    }

    /// A STALE entry must be re-measured rather than deadlock the caller.
    ///
    /// This is the one path the test above does not reach, and the only one where the locking
    /// could be wrong: the hit check holds a READ guard, and an expired entry then falls through
    /// to `read_disk_for_path` and a WRITE guard on the same `RwLock` — which is not reentrant.
    /// Whether the read guard is still alive at that point depends on when the `if let`
    /// scrutinee's temporary is dropped (edition 2024 drops it before the fall-through; the 2021
    /// rules did not). That is far too subtle to leave to a reading of the edition, so it is
    /// pinned here: this test HANGS rather than fails if the guards ever overlap.
    #[test]
    fn a_stale_entry_is_refreshed_without_deadlocking() {
        let cache = DiskCache::new();
        let path = std::path::Path::new("/");
        assert!(cache.snapshot(path).is_some());

        // Back-date the entry past the TTL — the same trick `sse_tickets` uses to age an entry
        // without sleeping through the window.
        {
            let mut map = cache.inner.write().unwrap();
            let (info, _) = *map.get(path).expect("just inserted");
            let stale = std::time::Instant::now()
                .checked_sub(super::TTL + std::time::Duration::from_secs(1))
                .expect("host uptime should exceed the disk TTL");
            map.insert(path.to_path_buf(), (info, stale));
        }

        // Must take the miss path — read guard released, filesystem re-measured, write guard
        // taken — and come back with a fresh reading.
        assert!(
            cache.snapshot(path).is_some(),
            "a stale entry must be re-measured, not dropped"
        );
        assert!(
            cache.inner.read().unwrap()[path].1.elapsed() < super::TTL,
            "the refreshed entry must carry a NEW timestamp, or every later call re-measures too"
        );
    }

    /// `invalidate` must clear EVERY volume: the e2e TRUNCATE changes free space on all of them,
    /// and a surviving entry would let the next test compute against the previous one's disk.
    #[test]
    fn invalidate_clears_every_volume() {
        let cache = DiskCache::new();
        cache.snapshot(std::path::Path::new("/"));
        cache.snapshot(&std::env::temp_dir());
        cache.invalidate();
        assert!(
            cache.inner.read().unwrap().is_empty(),
            "invalidate must not leave a volume behind"
        );
    }

    #[test]
    fn picks_longest_matching_mount() {
        // Both "/" and "/media" prefix the path; the dedicated volume must win.
        let mounts = vec![("/".to_string(), 100, 40), ("/media".to_string(), 200, 150)];
        let d = select_disk(&mounts, "/media/originals/x.jpg").unwrap();
        assert_eq!((d.total, d.free), (200, 150));
    }

    #[test]
    fn falls_back_to_root_when_no_specific_mount_matches() {
        let mounts = vec![("/".to_string(), 100, 40), ("/media".to_string(), 200, 150)];
        // "/var/lib" is only prefixed by "/".
        let d = select_disk(&mounts, "/var/lib/data").unwrap();
        assert_eq!((d.total, d.free), (100, 40));
    }

    #[test]
    fn relative_path_uses_root_fallback() {
        let mounts = vec![("/".to_string(), 100, 40)];
        // A relative path prefixes nothing, so the explicit "/" fallback applies.
        let d = select_disk(&mounts, "media/originals").unwrap();
        assert_eq!((d.total, d.free), (100, 40));
    }

    #[test]
    fn none_when_no_mount_matches_and_no_root() {
        // No "/" present and nothing prefixes the relative path → unknown (fail-open).
        let mounts = vec![("/data".to_string(), 100, 40)];
        assert!(select_disk(&mounts, "relative/path").is_none());
    }

    #[test]
    fn none_on_empty_mount_table() {
        assert!(select_disk(&[], "/media/x").is_none());
    }
}