//! The object store on disk: one file per object, named by its ID, shared by
//! every project and workspace on the machine. Files are written once and
//! never change, so readers never need a lock.

use std::fs;
use std::io::Write;
use std::path::{Path, PathBuf};

use anyhow::{Context, Result, anyhow};
use vcs_core::bundle::{self, Frame, RAW, ZSTD};
use vcs_core::object::{Kind, Object};
use vcs_core::{Id, Source, chunk};

/// Objects at least this big are compressed on disk and on the wire.
const COMPRESS_OVER: usize = 512;

#[derive(Clone)]
pub struct Store {
    root: PathBuf,
}

impl Store {
    pub fn open(root: &Path) -> Result<Store> {
        fs::create_dir_all(root).with_context(|| format!("creating {}", root.display()))?;
        Ok(Store { root: root.to_path_buf() })
    }

    fn path(&self, id: &Id) -> PathBuf {
        let hex = id.hex();
        self.root.join(&hex[..2]).join(&hex[2..])
    }

    pub fn has(&self, id: &Id) -> bool {
        self.path(id).is_file()
    }

    /// Every object here, with its file.
    pub fn all(&self) -> Result<Vec<(Id, PathBuf)>> {
        let mut out = Vec::new();
        for dir in fs::read_dir(&self.root)?.flatten() {
            let prefix = dir.file_name().to_string_lossy().to_string();
            if prefix.len() != 2 || !dir.path().is_dir() {
                continue;
            }
            for file in fs::read_dir(dir.path())?.flatten() {
                if let Some(id) = Id::parse(&format!("{prefix}{}", file.file_name().to_string_lossy())) {
                    out.push((id, file.path()));
                }
            }
        }
        Ok(out)
    }

    /// Deletes an object (vcs purge); answers the bytes it took.
    pub fn remove(&self, id: &Id) -> u64 {
        let len = self.stored_len(id);
        if fs::remove_file(self.path(id)).is_ok() { len } else { 0 }
    }

    /// How many bytes the object takes here (0 when it's missing).
    pub fn stored_len(&self, id: &Id) -> u64 {
        fs::metadata(self.path(id)).map(|m| m.len()).unwrap_or(0)
    }

    /// Stores an object (no-op if present). Returns its ID.
    pub fn put(&self, object: &Object) -> Result<Id> {
        let id = object.id();
        if self.has(&id) {
            return Ok(id);
        }
        self.write_frame(&encode(object, id))?;
        Ok(id)
    }

    fn write_frame(&self, frame: &Frame) -> Result<()> {
        let path = self.path(&frame.id);
        if path.is_file() {
            return Ok(());
        }
        let dir = path.parent().unwrap();
        fs::create_dir_all(dir)?;
        let mut bytes = Vec::with_capacity(frame.payload.len() + 10);
        bytes.push(frame.kind as u8);
        bytes.push(frame.codec);
        bytes.extend_from_slice(&frame.raw_len.to_le_bytes());
        bytes.extend_from_slice(&frame.payload);
        let tmp = dir.join(format!(".tmp-{}-{}", std::process::id(), rand::random::<u32>()));
        {
            let mut f = fs::File::create(&tmp)?;
            f.write_all(&bytes)?;
        }
        fs::rename(&tmp, &path)?;
        Ok(())
    }

    /// The stored frame (possibly compressed), for bundles.
    pub fn frame(&self, id: &Id) -> Result<Option<Frame>> {
        let bytes = match fs::read(self.path(id)) {
            Ok(b) => b,
            Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(None),
            Err(e) => return Err(e.into()),
        };
        if bytes.len() < 10 {
            return Err(anyhow!("object {} is damaged", id.short()));
        }
        let kind = Kind::from_u8(bytes[0]).ok_or_else(|| anyhow!("object {} is damaged", id.short()))?;
        let codec = bytes[1];
        let raw_len = u64::from_le_bytes(bytes[2..10].try_into().unwrap());
        Ok(Some(Frame { id: *id, kind, codec, raw_len, payload: bytes[10..].to_vec() }))
    }

    pub fn read(&self, id: &Id) -> Result<Option<Object>> {
        match self.frame(id)? {
            Some(frame) => Ok(Some(decode(&frame)?)),
            None => Ok(None),
        }
    }

    /// Reads every object kept here and checks it against its ID, in
    /// parallel. Answers how many there are and the IDs of damaged ones.
    pub fn verify_all(&self) -> Result<(usize, Vec<Id>)> {
        use rayon::prelude::*;
        let mut ids = Vec::new();
        for dir in fs::read_dir(&self.root)? {
            let dir = dir?;
            let Some(prefix) = dir.file_name().to_str().filter(|p| p.len() == 2).map(str::to_string) else { continue };
            if !dir.file_type()?.is_dir() {
                continue;
            }
            for f in fs::read_dir(dir.path())? {
                let name = f?.file_name();
                if let Some(id) = name.to_str().and_then(|rest| Id::parse(&format!("{prefix}{rest}"))) {
                    ids.push(id);
                }
            }
        }
        let damaged: Vec<Id> = ids
            .par_iter()
            .filter(|id| match self.frame(id) {
                Ok(Some(frame)) => decode(&frame).map_or(true, |o| bundle::verify(&frame, &o.data).is_err()),
                _ => true,
            })
            .copied()
            .collect();
        Ok((ids.len(), damaged))
    }

    /// Stores a file's content (chunked when big) and returns its ID.
    pub fn put_file(&self, content: &[u8]) -> Result<Id> {
        let objects = chunk::file_objects(content);
        let id = objects[0].id();
        if self.has(&id) {
            return Ok(id);
        }
        // Chunks first, so the list never points at missing chunks.
        for o in objects.iter().skip(1) {
            self.put(o)?;
        }
        self.put(&objects[0])?;
        Ok(id)
    }

    /// A file's content: a sealed folder's file opened when this machine has its key.
    pub fn read_file(&self, id: &Id) -> Result<Vec<u8>> {
        Ok(crate::seal::open(chunk::read_file(self, id)?))
    }

    /// Every object a file ID needs (itself plus chunks).
    /// A file's size (a big file's whole size), read from its frame without
    /// decompressing a blob; None when it isn't here.
    pub fn file_size(&self, id: &Id) -> Result<Option<u64>> {
        let Some(frame) = self.frame(id)? else { return Ok(None) };
        if frame.kind != Kind::Chunks {
            return Ok(Some(frame.raw_len));
        }
        Ok(Some(vcs_core::object::decode_chunks(&decode(&frame)?.data)?.iter().map(|(_, n)| n).sum()))
    }

    pub fn file_parts(&self, id: &Id) -> Result<Vec<Id>> {
        let mut out = vec![*id];
        // The first byte says whether it's a chunk list; blobs aren't read.
        let mut kind = [0u8; 1];
        match fs::File::open(self.path(id)).and_then(|mut f| std::io::Read::read_exact(&mut f, &mut kind)) {
            Ok(()) if kind[0] == Kind::Chunks as u8 => {}
            Ok(()) => return Ok(out),
            Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(out),
            Err(e) => return Err(e.into()),
        }
        if let Some(frame) = self.frame(id)?
            && frame.kind == Kind::Chunks
        {
            let o = decode(&frame)?;
            out.extend(vcs_core::object::decode_chunks(&o.data)?.into_iter().map(|(c, _)| c));
        }
        Ok(out)
    }

    /// A bundle of the given objects (those present).
    pub fn bundle(&self, ids: &[Id]) -> Result<Vec<u8>> {
        let mut out = Vec::new();
        bundle::write_header(&mut out);
        for id in ids {
            if let Some(frame) = self.frame(id)? {
                bundle::write_frame(&mut out, &frame);
            }
        }
        bundle::write_end(&mut out);
        Ok(out)
    }

    /// Stores every object in a bundle after checking it. Returns how many.
    pub fn ingest(&self, bytes: &[u8]) -> Result<usize> {
        let frames = bundle::read(bytes)?;
        let n = frames.len();
        for frame in frames {
            if self.has(&frame.id) {
                continue;
            }
            let object = decode(&frame)?;
            bundle::verify(&frame, &object.data)?;
            self.write_frame(&frame)?;
        }
        Ok(n)
    }
}

impl Source for Store {
    fn get(&self, id: &Id) -> vcs_core::Result<Option<Object>> {
        self.read(id).map_err(|e| vcs_core::Error::Corrupt(format!("{e:#}")))
    }
}

pub fn encode(object: &Object, id: Id) -> Frame {
    if object.data.len() >= COMPRESS_OVER
        && let Ok(packed) = zstd::bulk::compress(&object.data, 3)
        && packed.len() < object.data.len()
    {
        return Frame { id, kind: object.kind, codec: ZSTD, raw_len: object.data.len() as u64, payload: packed };
    }
    Frame { id, kind: object.kind, codec: RAW, raw_len: object.data.len() as u64, payload: object.data.clone() }
}

pub fn decode(frame: &Frame) -> Result<Object> {
    let data = match frame.codec {
        RAW => frame.payload.clone(),
        ZSTD => {
            // A frame's claimed size isn't trusted: reading stops just past it.
            use std::io::Read;
            let mut out = Vec::with_capacity((frame.raw_len as usize).min(frame.payload.len().saturating_mul(4)).min(1 << 20));
            zstd::stream::read::Decoder::new(&frame.payload[..])?.take(frame.raw_len + 1).read_to_end(&mut out).context("damaged object")?;
            out
        }
        _ => return Err(anyhow!("unknown codec")),
    };
    Ok(Object { kind: frame.kind, data })
}

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

    #[test]
    fn put_read_bundle() {
        let dir = tempfile::tempdir().unwrap();
        let store = Store::open(dir.path()).unwrap();
        let big = "x".repeat(10_000).into_bytes();
        let id = store.put_file(&big).unwrap();
        assert_eq!(store.read_file(&id).unwrap(), big);
        let bundle = store.bundle(&[id]).unwrap();
        let other = tempfile::tempdir().unwrap();
        let store2 = Store::open(other.path()).unwrap();
        assert_eq!(store2.ingest(&bundle).unwrap(), 1);
        assert_eq!(store2.read_file(&id).unwrap(), big);
    }
}