//! A local record of chunks the repository is known to hold.
//!
//! Storing a chunk that already exists is harmless — writes are
//! idempotent and deduplication is the whole point — but *finding out*
//! costs a round trip to the backend for every chunk considered. Over a
//! remote link that dominates: re-reading an interrupted twenty-gigabyte
//! file asks about twenty thousand chunks before it gets back to where
//! it stopped, and a second backup of an unchanged tree asks about every
//! chunk in it.
//!
//! The cache answers those questions locally. It is advisory in one
//! direction only: not knowing about a chunk simply means asking the
//! backend as before, so a truncated, half-written, or deleted cache
//! costs speed and nothing else. Claiming a chunk the repository does
//! not have would be another matter, which is why entries are only ever
//! added after the backend has confirmed the chunk — by storing it, or
//! by saying it already had it.

use std::{
    collections::HashSet,
    fs::{File, OpenOptions},
    io::{BufReader, BufWriter, Write as _},
    path::{Path, PathBuf},
    sync::Mutex,
};

use crate::{ChunkId, Error};

/// Chunk ids are 32 bytes and the body is a plain concatenation of them,
/// so a torn write can only ever leave a trailing partial id, which
/// loading discards.
const ID_BYTES: usize = 32;

const MAGIC: &[u8; 8] = b"beepchnk";
const FORMAT_VERSION: u8 = 1;

/// Magic, version, repository fingerprint, epoch.
const HEADER_BYTES: usize = 8 + 1 + 16 + 16;

/// What a cache file must agree with before a single id in it is
/// trusted: which repository it belongs to, and what that repository had
/// removed when it was written.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct CacheIdentity {
    /// [`crate::Repository::fingerprint`] — the repository *and*
    /// passphrase this cache describes.
    pub fingerprint: [u8; 16],
    /// [`crate::Repository::epoch`] — a digest of everything the
    /// repository has ever had pruned out of it.
    pub epoch: [u8; 16],
}

impl CacheIdentity {
    fn encode(&self) -> Vec<u8> {
        let mut header = Vec::with_capacity(HEADER_BYTES);

        header.extend_from_slice(MAGIC);
        header.push(FORMAT_VERSION);
        header.extend_from_slice(&self.fingerprint);
        header.extend_from_slice(&self.epoch);

        header
    }

    /// Reads a header, or `None` if this is not a cache file of ours.
    fn decode(bytes: &[u8]) -> Option<CacheIdentity> {
        if bytes.len() < HEADER_BYTES || &bytes[..8] != MAGIC || bytes[8] != FORMAT_VERSION {
            return None;
        }

        Some(CacheIdentity {
            fingerprint: bytes[9..25].try_into().ok()?,
            epoch: bytes[25..41].try_into().ok()?,
        })
    }
}

pub struct ChunkCache {
    known: Mutex<HashSet<ChunkId>>,
    /// Appended to as chunks are confirmed. Never fsynced: losing recent
    /// additions costs a few round trips next time, and nothing else.
    log: Mutex<BufWriter<File>>,
    path: PathBuf,
    /// Which repository this cache describes. Fixed for the life of the
    /// file, unlike the epoch, which every prune moves on.
    fingerprint: [u8; 16],
}

impl ChunkCache {
    /// Opens the cache at `path` for the repository described by
    /// `identity`, keeping what it holds only if it agrees.
    ///
    /// A file belonging to another repository, written by another format
    /// version, or written before a prune is discarded and started
    /// again — never read. Starting empty is always correct here: it
    /// costs round trips and nothing else, whereas trusting the wrong
    /// file would have a backup skip storing chunks the repository does
    /// not have.
    pub fn open(path: impl Into<PathBuf>, identity: CacheIdentity) -> Result<ChunkCache, Error> {
        let path = path.into();

        if let Some(parent) = path.parent() {
            std::fs::create_dir_all(parent)?;
        }

        let known = match File::open(&path) {
            Ok(file) => Self::load(file, identity)?,
            Err(err) if err.kind() == std::io::ErrorKind::NotFound => None,
            Err(err) => return Err(err.into()),
        };

        let known = match known {
            Some(known) => known,
            None => {
                // Start the file again under this identity
                let mut fresh = File::create(&path)?;
                fresh.write_all(&identity.encode())?;

                HashSet::new()
            }
        };

        let log = OpenOptions::new().append(true).open(&path)?;

        Ok(ChunkCache {
            known: Mutex::new(known),
            log: Mutex::new(BufWriter::new(log)),
            path,
            fingerprint: identity.fingerprint,
        })
    }

    /// Reads a cache file, returning its contents if its header agrees
    /// with `identity` and `None` if the file is not this repository's
    /// to trust. A trailing partial id — all a crash can leave — is
    /// dropped.
    fn load(file: File, identity: CacheIdentity) -> Result<Option<HashSet<ChunkId>>, Error> {
        let mut reader = BufReader::new(file);
        let mut header = [0u8; HEADER_BYTES];

        if read_full(&mut reader, &mut header)? != HEADER_BYTES {
            return Ok(None);
        }

        if CacheIdentity::decode(&header) != Some(identity) {
            return Ok(None);
        }

        let mut known = HashSet::new();
        let mut buffer = [0u8; ID_BYTES];

        while read_full(&mut reader, &mut buffer)? == ID_BYTES {
            known.insert(ChunkId::from_bytes(buffer));
        }

        Ok(Some(known))
    }

    /// Drops the given chunks and adopts a new epoch: what a prune does
    /// to the cache on the machine it ran from, so that the rest of the
    /// cache stays useful instead of being thrown away wholesale.
    ///
    /// Rewritten in place through a temporary file, so an interrupted
    /// rewrite leaves either the old cache (whose epoch no longer
    /// matches, and which is therefore discarded on the next open) or
    /// the new one.
    pub fn forget(&self, swept: &HashSet<ChunkId>, epoch: [u8; 16]) -> Result<(), Error> {
        let mut known = self.known.lock().unwrap();
        let mut log = self.log.lock().unwrap();

        known.retain(|id| !swept.contains(id));

        let identity = CacheIdentity {
            fingerprint: self.fingerprint,
            epoch,
        };

        let temporary = self.path.with_extension("rewriting");
        let mut fresh = BufWriter::new(File::create(&temporary)?);

        fresh.write_all(&identity.encode())?;

        for id in known.iter() {
            fresh.write_all(id.as_bytes())?;
        }

        fresh.flush()?;
        drop(fresh);

        std::fs::rename(&temporary, &self.path)?;
        *log = BufWriter::new(OpenOptions::new().append(true).open(&self.path)?);

        Ok(())
    }

    /// Where the cache lives, for diagnostics.
    pub fn path(&self) -> &Path {
        &self.path
    }

    /// How many chunks it knows about.
    pub fn len(&self) -> usize {
        self.known.lock().unwrap().len()
    }

    pub fn is_empty(&self) -> bool {
        self.len() == 0
    }

    /// Whether the repository is known to hold this chunk. `false` means
    /// "ask the backend", never "the chunk is absent".
    pub fn holds(&self, id: ChunkId) -> bool {
        self.known.lock().unwrap().contains(&id)
    }

    /// Records a chunk the backend has confirmed — because it was just
    /// stored, or because it said it already had it.
    ///
    /// Write failures are deliberately ignored: this is a cache, and a
    /// backup must not fail because one could not be written.
    pub fn remember(&self, id: ChunkId) {
        if !self.known.lock().unwrap().insert(id) {
            return;
        }

        let mut log = self.log.lock().unwrap();
        let _ = log.write_all(id.as_bytes());
    }

    /// Pushes buffered additions to the file. Called at the points a
    /// pipeline makes its own state durable; missing them loses only
    /// speed.
    pub fn flush(&self) {
        let _ = self.log.lock().unwrap().flush();
    }
}

impl Drop for ChunkCache {
    fn drop(&mut self) {
        self.flush();
    }
}

/// Reads until the buffer is full or the file ends, returning how much
/// was read. `Read::read` is free to return less than asked for, and a
/// short read at the end is exactly what a torn write looks like.
fn read_full(reader: &mut impl std::io::Read, buffer: &mut [u8]) -> Result<usize, Error> {
    let mut filled = 0;

    while filled < buffer.len() {
        match reader.read(&mut buffer[filled..])? {
            0 => break,
            count => filled += count,
        }
    }

    Ok(filled)
}

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

    fn id(seed: u8) -> ChunkId {
        ChunkId::from_bytes([seed; ID_BYTES])
    }

    fn identity(fingerprint: u8, epoch: u8) -> CacheIdentity {
        CacheIdentity {
            fingerprint: [fingerprint; 16],
            epoch: [epoch; 16],
        }
    }

    #[test]
    fn remembers_across_reopening() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("chunks");

        {
            let cache = ChunkCache::open(&path, identity(1, 0)).unwrap();
            assert!(!cache.holds(id(1)));

            cache.remember(id(1));
            cache.remember(id(2));
            cache.remember(id(1)); // already known, not written twice

            assert!(cache.holds(id(1)));
            assert_eq!(cache.len(), 2);
        }

        let reopened = ChunkCache::open(&path, identity(1, 0)).unwrap();
        assert_eq!(reopened.len(), 2);
        assert!(reopened.holds(id(1)));
        assert!(reopened.holds(id(2)));
        assert!(!reopened.holds(id(3)));

        assert_eq!(
            std::fs::metadata(&path).unwrap().len() as usize,
            HEADER_BYTES + 2 * ID_BYTES,
            "a repeated chunk is not appended again"
        );
    }

    #[test]
    fn a_torn_write_costs_only_its_own_record() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("chunks");

        {
            let cache = ChunkCache::open(&path, identity(1, 0)).unwrap();
            cache.remember(id(1));
            cache.remember(id(2));
        }

        // Half of the second id survives the crash
        let mut truncated = std::fs::read(&path).unwrap();
        truncated.truncate(HEADER_BYTES + ID_BYTES + ID_BYTES / 2);
        std::fs::write(&path, truncated).unwrap();

        let reopened = ChunkCache::open(&path, identity(1, 0)).unwrap();
        assert!(reopened.holds(id(1)), "whole records still load");
        assert!(!reopened.holds(id(2)), "the partial one is dropped");

        // And it goes on being usable
        reopened.remember(id(3));
        assert!(reopened.holds(id(3)));
    }

    #[test]
    fn another_repositorys_cache_is_never_read() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("chunks");

        {
            let cache = ChunkCache::open(&path, identity(1, 0)).unwrap();
            cache.remember(id(1));
        }

        let foreign = ChunkCache::open(&path, identity(2, 0)).unwrap();
        assert!(
            !foreign.holds(id(1)),
            "a cache belonging to another repository must not be trusted"
        );

        // ...and it is now this repository's cache, not the other's
        foreign.remember(id(7));
        drop(foreign);

        let reopened = ChunkCache::open(&path, identity(2, 0)).unwrap();
        assert!(reopened.holds(id(7)));
        assert!(!reopened.holds(id(1)));
    }

    #[test]
    fn a_prune_since_the_cache_was_written_discards_it() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("chunks");

        {
            let cache = ChunkCache::open(&path, identity(1, 0)).unwrap();
            cache.remember(id(1));
            cache.remember(id(2));
        }

        // Something was pruned elsewhere: every id in the file may name
        // a chunk that is gone
        let after = ChunkCache::open(&path, identity(1, 9)).unwrap();
        assert!(after.is_empty());
    }

    #[test]
    fn forgetting_keeps_the_rest_of_the_cache_warm() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("chunks");

        let cache = ChunkCache::open(&path, identity(1, 0)).unwrap();
        for seed in 1..=5 {
            cache.remember(id(seed));
        }

        let swept: HashSet<ChunkId> = [id(2), id(4)].into_iter().collect();
        cache.forget(&swept, [9u8; 16]).unwrap();

        assert!(cache.holds(id(1)) && cache.holds(id(3)) && cache.holds(id(5)));
        assert!(!cache.holds(id(2)) && !cache.holds(id(4)));

        // It goes on being appendable, and reopens under the new epoch
        cache.remember(id(6));
        drop(cache);

        let reopened = ChunkCache::open(&path, identity(1, 9)).unwrap();
        assert_eq!(reopened.len(), 4);
        assert!(reopened.holds(id(6)) && !reopened.holds(id(2)));
    }

    #[test]
    fn a_missing_cache_simply_starts_empty() {
        let dir = tempfile::tempdir().unwrap();
        let cache =
            ChunkCache::open(dir.path().join("nested/deeper/chunks"), identity(1, 0)).unwrap();

        assert!(cache.is_empty());
        cache.remember(id(9));
        assert!(cache.holds(id(9)));
    }
}