//! The borrow contract, against a filesystem and a fake client.
//!
//! The fake stands in for the sync client's socket, not for the filesystem:
//! it renames stubs exactly as suffix-mode VFS does, so everything under test
//! is the real path resolution and the real state tracking.

use super::*;
use crate::{FolderBackend, Vfs};
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicUsize, Ordering};

/// A stand-in for a sync client, counting what it was asked to do.
struct FakeClient {
    suffix: &'static str,
    hydrations: AtomicUsize,
    dehydrations: AtomicUsize,
    /// When true, refuse to hydrate — the client is running but the server is
    /// not reachable.
    broken: bool,
}

impl FakeClient {
    fn new() -> Arc<Self> {
        Arc::new(Self {
            suffix: ".nextcloud",
            hydrations: AtomicUsize::new(0),
            dehydrations: AtomicUsize::new(0),
            broken: false,
        })
    }
    fn broken() -> Arc<Self> {
        Arc::new(Self {
            suffix: ".nextcloud",
            hydrations: AtomicUsize::new(0),
            dehydrations: AtomicUsize::new(0),
            broken: true,
        })
    }
}

impl Vfs for FakeClient {
    fn name(&self) -> &'static str {
        "fake"
    }
    fn is_placeholder(&self, on_disk: &str) -> bool {
        on_disk.ends_with(self.suffix)
    }
    fn real_name<'a>(&self, on_disk: &'a str) -> &'a str {
        on_disk.strip_suffix(self.suffix).unwrap_or(on_disk)
    }
    fn placeholder_name(&self, name: &str) -> std::borrow::Cow<'_, str> {
        std::borrow::Cow::Owned(format!("{name}{}", self.suffix))
    }
    fn can_materialise(&self) -> bool {
        true
    }
    fn materialise(&self, local: &Path) -> Result<(), RemoteError> {
        self.hydrations.fetch_add(1, Ordering::SeqCst);
        if self.broken {
            return Err(RemoteError::Network("no server".into()));
        }
        // Suffix mode renames rather than filling in place, and writes the
        // real content.
        let real = PathBuf::from(local.to_string_lossy().strip_suffix(self.suffix).unwrap());
        std::fs::write(&real, vec![9u8; 4096]).unwrap();
        std::fs::remove_file(local).unwrap();
        Ok(())
    }
    fn dematerialise(&self, local: &Path) -> Result<(), RemoteError> {
        self.dehydrations.fetch_add(1, Ordering::SeqCst);
        let stub = format!("{}{}", local.display(), self.suffix);
        std::fs::write(&stub, [0u8]).unwrap();
        std::fs::remove_file(local).unwrap();
        Ok(())
    }
}

struct Tmp(PathBuf);

impl Tmp {
    fn new(name: &str) -> Self {
        let d = std::env::temp_dir().join(format!("dr-borrow-{name}"));
        let _ = std::fs::remove_dir_all(&d);
        std::fs::create_dir_all(&d).unwrap();
        Tmp(d)
    }
    /// A dehydrated photograph.
    fn stub(&self, rel: &str) -> &Self {
        std::fs::write(self.0.join(format!("{rel}.nextcloud")), [0u8]).unwrap();
        self
    }
    /// One the user already has.
    fn real(&self, rel: &str) -> &Self {
        std::fs::write(self.0.join(rel), vec![1u8; 2048]).unwrap();
        self
    }
    fn has(&self, rel: &str) -> bool {
        self.0.join(rel).is_file()
    }
    fn backend(&self, vfs: Arc<dyn Vfs>) -> FolderBackend {
        FolderBackend::with_vfs(&self.0, vfs).unwrap()
    }
}

impl Drop for Tmp {
    fn drop(&mut self) {
        let _ = std::fs::remove_dir_all(&self.0);
    }
}

#[tokio::test]
async fn a_borrowed_placeholder_is_downloaded_and_given_back() {
    let t = Tmp::new("cycle");
    t.stub("a.CR2");
    let client = FakeClient::new();
    let b = t.backend(client.clone());
    let pool = BorrowPool::new();
    let path = RemotePath::new("a.CR2");

    {
        let held = pool.borrow(&b, &path).await.unwrap();
        assert!(held.hydrated(), "this borrow paid for it");
        assert!(t.has("a.CR2"), "content is here while borrowed");
        assert_eq!(
            b.get(&RemoteId::Path(path.clone()), None)
                .await
                .unwrap()
                .len(),
            4096
        );
    }

    let stats = pool.release_all(&b).await;
    assert_eq!(stats.released, 1);
    assert!(!t.has("a.CR2"), "given back");
    assert!(t.has("a.CR2.nextcloud"), "a placeholder is left behind");
    assert_eq!(client.dehydrations.load(Ordering::SeqCst), 1);
}

#[tokio::test]
async fn a_file_the_user_already_had_is_never_taken_away() {
    // The rule the whole design rests on. Silently undoing a pin — or just a
    // file someone opened yesterday — after an indexing run is the failure
    // that would make people stop trusting this.
    let t = Tmp::new("keep");
    t.real("pinned.CR2");
    let client = FakeClient::new();
    let b = t.backend(client.clone());
    let pool = BorrowPool::new();

    {
        let held = pool
            .borrow(&b, &RemotePath::new("pinned.CR2"))
            .await
            .unwrap();
        assert!(!held.hydrated(), "nothing was downloaded");
    }
    let stats = pool.release_all(&b).await;

    assert_eq!(stats.released, 0);
    assert!(t.has("pinned.CR2"), "still here");
    assert_eq!(client.hydrations.load(Ordering::SeqCst), 0);
    assert_eq!(client.dehydrations.load(Ordering::SeqCst), 0);
}

#[tokio::test]
async fn two_lanes_wanting_one_file_download_it_once() {
    // The thumbnail pass and the face pass meet on the same RAW. Without
    // counting, the first to finish dehydrates the file the second is reading.
    let t = Tmp::new("shared");
    t.stub("a.CR2");
    let client = FakeClient::new();
    let b = t.backend(client.clone());
    let pool = BorrowPool::new();
    let path = RemotePath::new("a.CR2");

    let first = pool.borrow(&b, &path).await.unwrap();
    let second = pool.borrow(&b, &path).await.unwrap();

    assert_eq!(client.hydrations.load(Ordering::SeqCst), 1, "paid once");
    drop(first);
    assert!(t.has("a.CR2"), "still held by the second borrower");
    drop(second);

    pool.release_all(&b).await;
    assert!(!t.has("a.CR2"));
}

#[tokio::test]
async fn a_failed_download_does_not_leave_a_phantom_borrow() {
    // The client is up but the server is not. The pass must see the failure
    // and the pool must not believe it holds anything.
    let t = Tmp::new("failed");
    t.stub("a.CR2");
    let b = t.backend(FakeClient::broken());
    let pool = BorrowPool::new();

    let e = pool
        .borrow(&b, &RemotePath::new("a.CR2"))
        .await
        .unwrap_err();
    assert!(matches!(e, RemoteError::Network(_)), "{e:?}");
    assert_eq!(pool.held(), 0);
    assert!(t.has("a.CR2.nextcloud"), "left as it was found");
}

#[tokio::test]
async fn borrowing_against_a_plain_folder_does_nothing_at_all() {
    // A caller written for a VFS library must run unchanged elsewhere, or
    // every sweep grows two code paths.
    let t = Tmp::new("plain");
    t.real("a.CR2");
    let b = FolderBackend::new(&t.0).unwrap();
    let pool = BorrowPool::new();

    let held = pool.borrow(&b, &RemotePath::new("a.CR2")).await.unwrap();
    assert!(!held.hydrated());
    drop(held);
    assert_eq!(pool.release_all(&b).await.released, 0);
    assert!(t.has("a.CR2"));
}

#[tokio::test]
async fn the_backend_is_what_decides_whether_a_file_was_ours() {
    // Not the pool, and not the caller. The backend had to look before
    // deciding whether to ask, so it can answer for the cost of that same
    // `stat`; a borrower working it out separately would pay a directory
    // listing per file and could get it wrong in the direction that releases
    // a file the user pinned.
    let t = Tmp::new("who-decides");
    t.real("had.CR2").stub("wanted.CR2");
    let b = t.backend(FakeClient::new());

    assert!(
        !b.materialise(&RemoteId::Path(RemotePath::new("had.CR2")))
            .await
            .unwrap(),
        "already here, so not ours to release"
    );
    assert!(
        b.materialise(&RemoteId::Path(RemotePath::new("wanted.CR2")))
            .await
            .unwrap(),
        "this call fetched it"
    );
}

#[tokio::test]
async fn a_file_borrowed_twice_in_one_pass_is_fetched_once_and_released_once() {
    // Thumbnailing and face indexing visit the same photograph. Fetching it
    // per stage doubles the transfer over the whole library.
    let t = Tmp::new("sequential");
    t.stub("a.CR2");
    let client = FakeClient::new();
    let b = t.backend(client.clone());
    let pool = BorrowPool::new();
    let path = RemotePath::new("a.CR2");

    // Sequential borrows, as two passes over one work list would make.
    drop(pool.borrow(&b, &path).await.unwrap());
    drop(pool.borrow(&b, &path).await.unwrap());

    assert_eq!(client.hydrations.load(Ordering::SeqCst), 1, "paid once");
    let stats = pool.release_all(&b).await;
    assert_eq!(stats.released, 1, "given back once");
}