//! 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 std::time::SystemTime;

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

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

/// Up to this many objects, [`flush`] syncs each one rather than the whole disk.
const SYNC_EACH_UPTO: usize = 64;

/// Objects written since the last [`flush`]: on disk as far as this process
/// can tell, but maybe not yet past a power cut.
static UNFLUSHED: std::sync::Mutex<Vec<PathBuf>> = std::sync::Mutex::new(Vec::new());

/// Makes every object written so far durable, so the database never names an
/// object a power cut could take back. The database calls it before it writes
/// anything that names objects (saves, versions, changes). Each file and its
/// folder; on Linux, one `syncfs` for many. A `syncfs` also waits for
/// everything else the disk has yet to write: after an on get of a big
/// project, its gigabyte of files (coline-app's first land took 7 s in it).
pub fn flush() -> Result<()> {
    let pending = std::mem::take(&mut *UNFLUSHED.lock().unwrap_or_else(|e| e.into_inner()));
    if pending.is_empty() {
        return Ok(());
    }
    #[cfg(target_os = "linux")]
    if pending.len() > SYNC_EACH_UPTO {
        use std::os::fd::AsRawFd;
        if let Some(dir) = pending[0].parent()
            && let Ok(d) = fs::File::open(dir)
            && unsafe { libc::syncfs(d.as_raw_fd()) } == 0
        {
            return Ok(());
        }
    }
    let mut dirs = std::collections::BTreeSet::new();
    for p in &pending {
        // Windows flushes only a file opened for writing. A flush that fails
        // doesn't fail the command: the object is on disk as far as anyone
        // can tell, and on check finds and repairs one a power cut broke.
        if let Ok(f) = fs::OpenOptions::new().write(true).open(p) {
            let _ = f.sync_data();
        }
        if let Some(d) = p.parent() {
            dirs.insert(d.to_path_buf());
        }
    }
    #[cfg(unix)]
    for d in dirs {
        if let Ok(f) = fs::File::open(&d) {
            let _ = f.sync_all();
        }
    }
    Ok(())
}

/// Objects are written one file each; `on gc` moves older ones into a pack
/// (one file holding many, as they were) with a sorted index, so a big
/// project isn't tens of thousands of small files. Reads look in both.
const PACKS: &str = "packs";
const IDX_MAGIC: &[u8; 8] = b"ONSETIX1";
/// A bundle bringing at least this many new objects, and at most this many
/// bytes of them, is stored as a pack.
const PACK_FROM: usize = 256;
const PACK_UPTO: usize = 64 << 20;

/// An object as its file holds it (and a pack): kind, codec, size, then the payload.
fn stored(frame: &Frame) -> Vec<u8> {
    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);
    bytes
}

/// A pack's index: (ID, offset, length) sorted by ID.
struct PackIndex {
    pack: PathBuf,
    entries: Vec<(Id, u64, u64)>,
}

#[derive(Default)]
struct Packs {
    list: Vec<PackIndex>,
    /// The packs folder's modified time when read: another process's pack shows as a change.
    seen: Option<SystemTime>,
}

#[derive(Clone)]
pub struct Store {
    root: PathBuf,
    packs: std::sync::Arc<std::sync::RwLock<Packs>>,
}

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(), packs: Default::default() })
    }

    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() || self.packed(id).is_some()
    }

    /// Reads the packs' indexes again if the packs folder changed since.
    fn load_packs(&self, force: bool) {
        let dir = self.root.join(PACKS);
        let modified = fs::metadata(&dir).and_then(|m| m.modified()).ok();
        if !force && self.packs.read().unwrap_or_else(|e| e.into_inner()).seen == modified {
            return;
        }
        let mut list = Vec::new();
        for f in fs::read_dir(&dir).into_iter().flatten().flatten() {
            let p = f.path();
            if p.extension().and_then(|e| e.to_str()) != Some("idx") {
                continue;
            }
            let Ok(bytes) = fs::read(&p) else { continue };
            if bytes.len() < 16 || &bytes[..8] != IDX_MAGIC {
                continue;
            }
            let n = u64::from_le_bytes(bytes[8..16].try_into().unwrap()) as usize;
            if bytes.len() != 16 + n * 48 {
                continue;
            }
            let entries = (0..n)
                .map(|i| {
                    let at = 16 + i * 48;
                    let id = Id(bytes[at..at + 32].try_into().unwrap());
                    (id, u64::from_le_bytes(bytes[at + 32..at + 40].try_into().unwrap()), u64::from_le_bytes(bytes[at + 40..at + 48].try_into().unwrap()))
                })
                .collect();
            list.push(PackIndex { pack: p.with_extension("pack"), entries });
        }
        *self.packs.write().unwrap_or_else(|e| e.into_inner()) = Packs { list, seen: modified };
    }

    /// Where a packed object is: (pack, offset, length).
    fn packed(&self, id: &Id) -> Option<(PathBuf, u64, u64)> {
        self.load_packs(false);
        let packs = self.packs.read().unwrap_or_else(|e| e.into_inner());
        packs.list.iter().find_map(|p| p.entries.binary_search_by(|e| e.0.cmp(id)).ok().map(|i| (p.pack.clone(), p.entries[i].1, p.entries[i].2)))
    }

    fn read_packed(&self, id: &Id) -> Result<Option<Vec<u8>>> {
        use std::io::{Read, Seek, SeekFrom};
        for attempt in 0..2 {
            let Some((pack, off, len)) = self.packed(id) else { return Ok(None) };
            let read = (|| -> std::io::Result<Vec<u8>> {
                let mut f = fs::File::open(&pack)?;
                f.seek(SeekFrom::Start(off))?;
                let mut buf = vec![0; len as usize];
                f.read_exact(&mut buf)?;
                Ok(buf)
            })();
            match read {
                Ok(buf) => return Ok(Some(buf)),
                // Repacked by another process meanwhile: read the indexes again.
                Err(_) if attempt == 0 => self.load_packs(true),
                Err(e) => return Err(e.into()),
            }
        }
        Ok(None)
    }

    /// Every packed object's ID.
    pub fn packed_ids(&self) -> Vec<Id> {
        self.load_packs(false);
        self.packs.read().unwrap_or_else(|e| e.into_inner()).list.iter().flat_map(|p| p.entries.iter().map(|e| e.0)).collect()
    }

    /// Moves loose objects not modified for `older` into a new pack.
    /// Answers how many it moved.
    pub fn pack_loose(&self, older: std::time::Duration) -> Result<usize> {
        let now = SystemTime::now();
        let mut picked: Vec<(Id, PathBuf)> = self
            .all()?
            .into_iter()
            .filter(|(_, p)| fs::metadata(p).and_then(|m| m.modified()).ok().and_then(|m| now.duration_since(m).ok()).is_some_and(|age| age >= older))
            .collect();
        if picked.is_empty() {
            return Ok(0);
        }
        picked.sort_by(|a, b| a.0.cmp(&b.0));
        let entries: Vec<(Id, Vec<u8>)> = picked.iter().filter_map(|(id, p)| fs::read(p).ok().map(|b| (*id, b))).collect();
        self.write_pack(&entries)?;
        for (_, p) in &picked {
            let _ = fs::remove_file(p);
        }
        Ok(entries.len())
    }

    /// Writes a pack and its index (the index last: an index means a whole pack).
    fn write_pack(&self, entries: &[(Id, Vec<u8>)]) -> Result<()> {
        let dir = self.root.join(PACKS);
        fs::create_dir_all(&dir)?;
        let name = format!("{:016x}{:08x}", rand::random::<u64>(), std::process::id());
        let (pack, idx) = (dir.join(format!("{name}.pack")), dir.join(format!("{name}.idx")));
        let mut body = Vec::new();
        let mut index = Vec::with_capacity(16 + entries.len() * 48);
        index.extend_from_slice(IDX_MAGIC);
        index.extend_from_slice(&(entries.len() as u64).to_le_bytes());
        for (id, bytes) in entries {
            index.extend_from_slice(&id.0);
            index.extend_from_slice(&(body.len() as u64).to_le_bytes());
            index.extend_from_slice(&(bytes.len() as u64).to_le_bytes());
            body.extend_from_slice(bytes);
        }
        for (path, data) in [(&pack, &body), (&idx, &index)] {
            let tmp = path.with_extension("tmp");
            let mut f = fs::File::create(&tmp)?;
            f.write_all(data)?;
            f.sync_all()?;
            fs::rename(&tmp, path)?;
        }
        #[cfg(unix)]
        if let Ok(d) = fs::File::open(&dir) {
            let _ = d.sync_all();
        }
        self.load_packs(true);
        Ok(())
    }

    /// Takes objects out of the packs that hold them (each such pack is
    /// written again without them, so their bytes are gone). Answers the bytes freed.
    pub fn remove_packed(&self, ids: &std::collections::HashSet<Id>) -> Result<u64> {
        self.load_packs(true);
        let packs: Vec<(PathBuf, Vec<(Id, u64, u64)>)> = self.packs.read().unwrap_or_else(|e| e.into_inner()).list.iter().map(|p| (p.pack.clone(), p.entries.clone())).collect();
        let mut freed = 0;
        for (pack, entries) in packs {
            if !entries.iter().any(|e| ids.contains(&e.0)) {
                continue;
            }
            let data = fs::read(&pack)?;
            let keep: Vec<(Id, Vec<u8>)> = entries.iter().filter(|e| !ids.contains(&e.0)).map(|e| (e.0, data[e.1 as usize..(e.1 + e.2) as usize].to_vec())).collect();
            freed += entries.iter().filter(|e| ids.contains(&e.0)).map(|e| e.2).sum::<u64>();
            if !keep.is_empty() {
                self.write_pack(&keep)?;
            }
            fs::remove_file(pack.with_extension("idx"))?;
            fs::remove_file(&pack)?;
        }
        self.load_packs(true);
        Ok(freed)
    }

    /// 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 (on purge); answers the bytes it took. A packed
    /// one's pack is written again without it, so its bytes are gone too.
    pub fn remove(&self, id: &Id) -> u64 {
        let loose = fs::metadata(self.path(id)).map(|m| m.len()).ok();
        let mut freed = 0;
        if let Some(len) = loose
            && fs::remove_file(self.path(id)).is_ok()
        {
            freed += len;
        }
        if self.packed(id).is_some() {
            freed += self.remove_packed(&std::iter::once(*id).collect()).unwrap_or(0);
        }
        freed
    }

    /// 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()).ok().or_else(|| self.packed(id).map(|p| p.2)).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() || self.packed(&frame.id).is_some() {
            return Ok(());
        }
        let dir = path.parent().unwrap();
        fs::create_dir_all(dir)?;
        let bytes = stored(frame);
        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)?;
        UNFLUSHED.lock().unwrap_or_else(|e| e.into_inner()).push(path);
        Ok(())
    }

    /// Deletes a damaged object, so it can be fetched again.
    pub fn forget(&self, id: &Id) {
        let _ = fs::remove_file(self.path(id));
        if self.packed(id).is_some() {
            let _ = self.remove_packed(&std::iter::once(*id).collect());
        }
    }

    /// 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 => match self.read_packed(id)? {
                Some(b) => b,
                None => 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);
                }
            }
        }
        ids.extend(self.packed_ids());
        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(onset_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),
            // Not loose: packed (read whole below), or not here.
            Err(e) if e.kind() == std::io::ErrorKind::NotFound && self.packed(id).is_some() => {}
            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(onset_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)
    }

    /// `bundle`, with each object that replaced one the host has (`bases`:
    /// object → the one it replaced) as a delta on it when that's under half
    /// its size: a folder with one entry changed goes as tens of bytes.
    pub fn bundle_based(&self, ids: &[Id], bases: &std::collections::HashMap<Id, Id>) -> Result<Vec<u8>> {
        let mut out = Vec::new();
        bundle::write_header(&mut out);
        for id in ids {
            let delta = match bases.get(id) {
                Some(base) => match (self.get(id)?, self.get(base)?) {
                    (Some(target), Some(was)) if target.kind == was.kind && target.kind != onset_core::Kind::Chunks => {
                        onset_core::delta::encode(&was.data, &target.data).map(|d| bundle::Frame { id: *id, kind: target.kind, codec: bundle::DELTA, raw_len: target.data.len() as u64, payload: [&base.0[..], &d[..]].concat() })
                    }
                    _ => None,
                },
                None => None,
            };
            if let Some(frame) = delta {
                bundle::write_frame(&mut out, &frame);
            } else 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.
    /// Many new ones (a project's first get) go into one pack, not a file
    /// each: the files were half of what a get wrote.
    pub fn ingest(&self, bytes: &[u8]) -> Result<usize> {
        let frames = bundle::read(bytes)?;
        let n = frames.len();
        let mut new = Vec::new();
        for frame in frames {
            if self.has(&frame.id) {
                continue;
            }
            let object = decode(&frame)?;
            bundle::verify(&frame, &object.data)?;
            new.push(frame);
        }
        // A big bundle (a big project's files) stays a file each: written as
        // one pack, coline-app's gigabyte took longer to get, not shorter.
        let bytes: usize = new.iter().map(|f| f.payload.len()).sum();
        if new.len() < PACK_FROM || bytes > PACK_UPTO {
            for frame in &new {
                self.write_frame(frame)?;
            }
            return Ok(n);
        }
        let mut entries: Vec<(Id, Vec<u8>)> = new.iter().map(|f| (f.id, stored(f))).collect();
        entries.sort_by(|a, b| a.0.cmp(&b.0));
        entries.dedup_by(|a, b| a.0 == b.0);
        self.write_pack(&entries)?;
        Ok(n)
    }

    /// `ingest` for an upload, which may hold deltas on objects stored here
    /// or earlier in it (see `bundle_based`): each is made whole, checked
    /// against its ID and stored whole. Together they make at most 64 MiB.
    pub fn ingest_uploaded(&self, bytes: &[u8]) -> Result<usize> {
        let frames = bundle::read_stored(bytes)?;
        let n = frames.len();
        let mut made = 0usize;
        for frame in frames {
            if self.has(&frame.id) {
                continue;
            }
            let object = if frame.codec == bundle::DELTA {
                let base_id = Id::from_slice(&frame.payload[..32]).ok_or_else(|| anyhow::anyhow!("bad delta"))?;
                let base = self.read(&base_id)?.ok_or_else(|| anyhow::anyhow!("object {} is a delta on {}, which isn't here; send it whole", frame.id.short(), base_id.short()))?;
                made += frame.raw_len as usize;
                if made > 64 << 20 {
                    anyhow::bail!("an upload's deltas make at most 64 MiB; send the objects whole");
                }
                Object::new(frame.kind, onset_core::git::apply_delta(&base.data, &frame.payload[32..], frame.raw_len as usize)?)
            } else {
                decode(&frame)?
            };
            bundle::verify(&frame, &object.data)?;
            self.write_frame(&encode(&object, frame.id))?;
        }
        Ok(n)
    }
}

impl Source for Store {
    fn get(&self, id: &Id) -> onset_core::Result<Option<Object>> {
        self.read(id).map_err(|e| onset_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::new(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);
    }
}