//! # Repository
//!
//! The encrypted, deduplicating storage format for backups.
//!
//! A repository lives on a [`Backend`] — any store offering write-once
//! objects with get/put/list — and holds three kinds of object:
//!
//! - **`header`**: public, write-once bootstrap data: the Argon2 salt
//!   and cost parameters, the chunker parameters, and a passphrase
//!   verifier. No key material is stored anywhere, on any host: the
//!   keys are re-derived from the passphrase (via the expensive KDF,
//!   once per session) every time the repository is opened. The
//!   passphrase *is* the key: it cannot be changed without re-encrypting
//!   everything, and losing it loses the repository.
//! - **chunks**: content-defined (FastCDC) chunks of data, encrypted with
//!   ChaCha20-Poly1305 and named by an HMAC-SHA256 of their plaintext, so
//!   identical content is stored once without revealing content
//!   fingerprints to whoever holds the storage.
//! - **snapshots**: small encrypted records pointing at a manifest — a
//!   stream of [`Entry`] values describing every object in one backup —
//!   which is itself stored through the chunk store.
//!
//! Every operation is idempotent: re-storing a chunk, a manifest, or a
//! snapshot that already exists is a cheap no-op. That property is what
//! lets a suspended or crashed backup simply redo recent work when it
//! resumes.

pub mod backend;
mod cache;
mod chunker;
mod crypto;
mod manifest;
pub mod os_path;

#[cfg(test)]
mod tests;

use std::io::Read;

use crate::crypto::{HEADER_VERSION, Header, MasterKeys, RepoConfig};

pub use crate::{
    backend::{Backend, ObjectKey, ObjectKind, local::LocalBackend},
    cache::{CacheIdentity, ChunkCache},
    crypto::{ChunkId, LockId, PruneId, SnapshotId},
    manifest::{Coverage, Entry, EntryKind, ManifestEntries, Snapshot},
    manifest::{Lock, PruneRecord},
};

#[derive(Debug, thiserror::Error)]
pub enum Error {
    #[error("the backend already contains a repository")]
    AlreadyInitialized,
    #[error("the backend does not contain a repository")]
    NotInitialized,
    #[error("chunk {0} is not present in the repository")]
    MissingChunk(ChunkId),
    #[error("chunk {0} did not decrypt to content matching its id")]
    CorruptChunk(ChunkId),
    #[error("snapshot {0} is not present in the repository")]
    MissingSnapshot(SnapshotId),
    #[error("decryption failed: wrong key or corrupted data")]
    Crypto,
    #[error("wrong passphrase for this repository")]
    WrongPassword,
    #[error("the repository header is malformed")]
    InvalidHeader,
    #[error("the repository was created by a newer, incompatible version")]
    UnsupportedVersion,
    #[error("{0} is not a valid object name")]
    InvalidObjectName(String),
    #[error("backend error: {0}")]
    Backend(String),
    #[error("the backend is at capacity: {0}")]
    BackendBusy(String),
    #[error("chunking failed: {0}")]
    Chunking(String),
    #[error(transparent)]
    CiboriumEncode(#[from] ciborium::ser::Error<std::io::Error>),
    #[error(transparent)]
    CiboriumDecode(#[from] ciborium::de::Error<std::io::Error>),
    #[error(transparent)]
    IO(#[from] std::io::Error),
}

/// A chunk already hashed and encrypted, awaiting its backend write.
///
/// Produced by [`Repository::prepare_chunk`], consumed by
/// [`Repository::store_prepared`].
pub struct PreparedChunk {
    id: ChunkId,
    sealed: Vec<u8>,
}

impl PreparedChunk {
    pub fn id(&self) -> ChunkId {
        self.id
    }

    /// The size of the encrypted object as it will be stored.
    pub fn sealed_len(&self) -> usize {
        self.sealed.len()
    }
}

/// An open repository: a [`Backend`] plus the unlocked master keys.
///
/// All methods take `&self` and the type is `Send + Sync`, so pipeline
/// threads share one repository behind an `Arc`.
pub struct Repository {
    backend: Box<dyn Backend>,
    keys: MasterKeys,
    config: RepoConfig,
    /// A local record of chunks this repository is known to hold, saving
    /// a round trip each. Optional, and advisory: without it every chunk
    /// is asked about as before.
    cache: Option<ChunkCache>,
    /// Sealed bytes handed to the backend through this handle; see
    /// [`Repository::bytes_written`].
    written: std::sync::atomic::AtomicU64,
}

impl Repository {
    /// Initializes a new repository on an empty backend: chooses a salt
    /// and parameters, derives the keys from `password`, and stores the
    /// public header. Nothing secret is written — creating a repository
    /// with the same passphrase and header always yields the same keys.
    pub fn create(
        backend: Box<dyn Backend>,
        password: impl AsRef<str>,
    ) -> Result<Repository, Error> {
        if backend.contains(&ObjectKey::Header)? {
            return Err(Error::AlreadyInitialized);
        }

        let mut header = Header::generate();
        let keys = MasterKeys::derive(password.as_ref(), &header)?;
        header.verifier = keys.verifier(&header);

        let mut bytes = Vec::new();
        ciborium::into_writer(&header, &mut bytes)?;
        backend.put(&ObjectKey::Header, &bytes)?;

        Ok(Repository {
            backend,
            keys,
            config: header.config,
            cache: None,
            written: std::sync::atomic::AtomicU64::new(0),
        })
    }

    /// Opens an existing repository, re-deriving its keys from
    /// `password` and the stored header. This is the only point where
    /// the expensive key derivation runs; a wrong passphrase is caught
    /// here by the header's verifier and fails with
    /// [`Error::WrongPassword`] before anything can be written under
    /// mismatched keys.
    pub fn open(backend: Box<dyn Backend>, password: impl AsRef<str>) -> Result<Repository, Error> {
        let Some(bytes) = backend.get(&ObjectKey::Header)? else {
            return Err(Error::NotInitialized);
        };

        let header: Header =
            ciborium::from_reader(bytes.as_slice()).map_err(|_| Error::InvalidHeader)?;

        if header.version != HEADER_VERSION {
            return Err(Error::UnsupportedVersion);
        }

        let keys = MasterKeys::derive(password.as_ref(), &header)?;

        if keys.verifier(&header) != header.verifier {
            return Err(Error::WrongPassword);
        }

        Ok(Repository {
            backend,
            keys,
            config: header.config,
            cache: None,
            written: std::sync::atomic::AtomicU64::new(0),
        })
    }

    /// Hashes one chunk of plaintext and, unless the repository already
    /// holds it, seals it for storage. Splitting this CPU-bound half
    /// from [`store_prepared`](Repository::store_prepared) lets a
    /// pipeline do the sealing on one thread pool and the (possibly
    /// slow, remote) write on another.
    pub fn prepare_chunk(
        &self,
        plaintext: &[u8],
    ) -> Result<(ChunkId, Option<PreparedChunk>), Error> {
        let id = ChunkId::compute(self.keys.id_key(), plaintext);

        if let Some(cache) = &self.cache
            && cache.holds(id)
        {
            return Ok((id, None));
        }

        if self.backend.contains(&ObjectKey::Chunk(id))? {
            self.remember(id);

            return Ok((id, None));
        }

        let sealed = crypto::seal(&self.keys, &chunk_aad(id), plaintext)?;

        Ok((id, Some(PreparedChunk { id, sealed })))
    }

    /// Notes that the backend has confirmed a chunk, so the next run
    /// need not ask about it.
    fn remember(&self, id: ChunkId) {
        if let Some(cache) = &self.cache {
            cache.remember(id);
        }
    }

    /// What a local cache for this repository must agree with before it
    /// may be trusted: which repository it belongs to, and what that
    /// repository had pruned when the cache was written.
    ///
    /// Costs one listing of the prune records, which is why it is taken
    /// once, when the cache is attached.
    pub fn cache_identity(&self) -> Result<CacheIdentity, Error> {
        Ok(CacheIdentity {
            fingerprint: self.fingerprint(),
            epoch: self.epoch()?,
        })
    }

    /// Attaches a local chunk cache, which lives at a path of the
    /// caller's choosing — the repository does not know where the
    /// machine keeps such things.
    ///
    /// A cache belongs to exactly one repository: use
    /// [`Repository::fingerprint`] in its name. Handing a repository the
    /// cache of another one would have it skip storing chunks this one
    /// does not hold.
    pub fn attach_chunk_cache(&mut self, cache: ChunkCache) {
        self.cache = Some(cache);
    }

    /// The attached cache, if any.
    pub fn chunk_cache(&self) -> Option<&ChunkCache> {
        self.cache.as_ref()
    }

    /// A stable name for this repository-and-passphrase pair, for
    /// naming per-repository local files. Derived from the keys, so it
    /// reveals nothing about the contents and cannot collide with
    /// another repository's.
    pub fn fingerprint(&self) -> [u8; 16] {
        let digest = crypto::keyed_digest(self.keys.id_key(), &[b"repository-fingerprint"]);

        digest[..16].try_into().expect("digest is 32 bytes")
    }

    /// Writes a prepared chunk to the backend. Idempotent, like every
    /// repository write.
    pub fn store_prepared(&self, chunk: &PreparedChunk) -> Result<(), Error> {
        self.put(&ObjectKey::Chunk(chunk.id), &chunk.sealed)?;
        self.remember(chunk.id);

        Ok(())
    }

    /// Stores one chunk of plaintext, returning its id and whether it was
    /// new. Content already in the repository is not written again, which
    /// is both the deduplication mechanism and what makes redoing work
    /// after a resume harmless.
    pub fn store_chunk(&self, plaintext: &[u8]) -> Result<(ChunkId, bool), Error> {
        match self.prepare_chunk(plaintext)? {
            (id, Some(chunk)) => {
                self.store_prepared(&chunk)?;

                Ok((id, true))
            }
            (id, None) => Ok((id, false)),
        }
    }

    /// Fetches and decrypts a chunk, verifying that its content still
    /// matches its id.
    pub fn load_chunk(&self, id: ChunkId) -> Result<Vec<u8>, Error> {
        let sealed = self
            .backend
            .get(&ObjectKey::Chunk(id))?
            .ok_or(Error::MissingChunk(id))?;

        let plaintext = crypto::open(&self.keys, &chunk_aad(id), &sealed)?;

        if ChunkId::compute(self.keys.id_key(), &plaintext) != id {
            return Err(Error::CorruptChunk(id));
        }

        Ok(plaintext)
    }

    /// True if the repository already holds the chunk with this id.
    pub fn contains_chunk(&self, id: ChunkId) -> Result<bool, Error> {
        self.backend.contains(&ObjectKey::Chunk(id))
    }

    /// Splits `source` into content-defined chunks using the repository's
    /// fixed chunking parameters.
    pub fn chunk_stream<'r>(
        &'r self,
        source: impl Read + 'r,
    ) -> impl Iterator<Item = Result<Vec<u8>, Error>> + 'r {
        chunker::chunks(source, &self.config)
    }

    /// Serializes a stream of manifest entries, chunks it, and stores the
    /// chunks, returning the chunk list for a [`Snapshot`] record.
    pub fn store_manifest(
        &self,
        entries: impl IntoIterator<Item = Entry>,
    ) -> Result<Vec<ChunkId>, Error> {
        let stream = manifest::EntryStream::new(entries.into_iter());
        let mut ids = Vec::new();

        for chunk in chunker::chunks(stream, &self.config) {
            let (id, _) = self.store_chunk(&chunk?)?;
            ids.push(id);
        }

        Ok(ids)
    }

    /// Iterates over the entries of a stored manifest, fetching its
    /// chunks on demand.
    pub fn manifest_entries(&self, manifest: Vec<ChunkId>) -> ManifestEntries<'_> {
        ManifestEntries::new(self, manifest)
    }

    /// Derives the snapshot id for a backup of `root` whose manifest is
    /// `manifest` and which covers what `coverage` says. The id is a
    /// keyed digest rather than a random value, so a backup that is
    /// redone after an interruption converges on the same snapshot
    /// object instead of creating a duplicate.
    ///
    /// Coverage belongs in the derivation because objects are
    /// write-once: a partial snapshot and the complete one that
    /// supersedes it can name exactly the same manifest — the last
    /// checkpoint of a run having drained every entry — and two records
    /// making different claims must not be one object, or the second
    /// one would silently not be written.
    pub fn snapshot_id(
        &self,
        root: &std::path::Path,
        manifest: &[ChunkId],
        coverage: Coverage,
    ) -> SnapshotId {
        let mut parts: Vec<&[u8]> = vec![
            b"snapshot-id",
            coverage.tag(),
            root.as_os_str().as_encoded_bytes(),
        ];
        parts.extend(manifest.iter().map(|id| id.as_bytes().as_slice()));

        let digest = crypto::keyed_digest(self.keys.id_key(), &parts);

        SnapshotId::from_bytes(digest[..16].try_into().expect("digest is 32 bytes"))
    }

    /// Derives the id for a snapshot rewritten from another over a
    /// narrowed manifest, as an excision makes.
    ///
    /// What it is derived from is the snapshot it replaces, rather than
    /// what it now holds. Two backups of one tree can differ only in
    /// what an excision took out of them, and snapshots that end up
    /// sharing a manifest must stay separate records rather than
    /// collapsing into one write-once object and taking each other's
    /// place. Deriving from the original also keeps the useful half of
    /// content addressing: rewriting the same snapshot the same way
    /// lands on the same id, so an interrupted excision converges
    /// instead of leaving duplicates behind.
    pub fn rewritten_snapshot_id(&self, original: &Snapshot, manifest: &[ChunkId]) -> SnapshotId {
        let mut parts: Vec<&[u8]> = vec![b"rewritten-snapshot-id", original.id.as_bytes()];
        parts.extend(manifest.iter().map(|id| id.as_bytes().as_slice()));

        let digest = crypto::keyed_digest(self.keys.id_key(), &parts);

        SnapshotId::from_bytes(digest[..16].try_into().expect("digest is 32 bytes"))
    }

    /// Stores a snapshot record. This is the final step of a backup: the
    /// snapshot only becomes visible once fully written.
    pub fn store_snapshot(&self, snapshot: &Snapshot) -> Result<(), Error> {
        let mut plaintext = Vec::new();
        ciborium::into_writer(snapshot, &mut plaintext)?;

        let sealed = crypto::seal(&self.keys, &snapshot_aad(snapshot.id), &plaintext)?;
        self.put(&ObjectKey::Snapshot(snapshot.id), &sealed)?;

        Ok(())
    }

    /// Fetches and decrypts a snapshot record.
    pub fn load_snapshot(&self, id: SnapshotId) -> Result<Snapshot, Error> {
        let sealed = self
            .backend
            .get(&ObjectKey::Snapshot(id))?
            .ok_or(Error::MissingSnapshot(id))?;

        let plaintext = crypto::open(&self.keys, &snapshot_aad(id), &sealed)?;

        Ok(ciborium::from_reader(plaintext.as_slice())?)
    }

    /// Removes a snapshot record, leaving its chunks alone.
    ///
    /// Used to retire the partial snapshots a suspended backup
    /// publishes, once a newer one or the finished backup has replaced
    /// them. Deleting an absent record is not an error, so a crash
    /// between publishing and recording the fact costs nothing.
    pub fn delete_snapshot(&self, id: SnapshotId) -> Result<(), Error> {
        self.backend.delete(&ObjectKey::Snapshot(id))
    }

    /// Records that a prune has happened, which is what tells every
    /// local chunk cache that its contents may no longer be true.
    ///
    /// Written *before* any chunk is swept: a prune interrupted halfway
    /// must still invalidate the caches, or one of them could go on
    /// vouching for a chunk that is already gone.
    pub fn record_prune(&self, record: &PruneRecord) -> Result<(), Error> {
        let mut plaintext = Vec::new();
        ciborium::into_writer(record, &mut plaintext)?;

        let sealed = crypto::seal(&self.keys, &prune_aad(record.id), &plaintext)?;
        self.put(&ObjectKey::Prune(record.id), &sealed)?;

        Ok(())
    }

    /// Every prune this repository has recorded, newest last.
    pub fn prunes(&self) -> Result<Vec<PruneRecord>, Error> {
        let mut ids = Vec::new();

        self.backend.list(ObjectKind::Prune, &mut |key| {
            if let ObjectKey::Prune(id) = key {
                ids.push(id);
            }

            Ok(())
        })?;

        let mut records = Vec::new();

        for id in ids {
            // A record deleted between listing and reading is simply not
            // part of the history any more
            let Some(sealed) = self.backend.get(&ObjectKey::Prune(id))? else {
                continue;
            };

            let plaintext = crypto::open(&self.keys, &prune_aad(id), &sealed)?;
            records.push(ciborium::from_reader(plaintext.as_slice())?);
        }

        records.sort_by_key(|record: &PruneRecord| record.started);

        Ok(records)
    }

    /// A digest of everything ever removed from this repository.
    ///
    /// Two readings that agree mean nothing has been swept in between,
    /// which is precisely what a local chunk cache needs to know before
    /// it may be trusted. Derived from the set of prune records rather
    /// than a counter, so it needs no ordering, no read-modify-write,
    /// and nothing but the write-once puts every backend offers.
    pub fn epoch(&self) -> Result<[u8; 16], Error> {
        let mut ids = Vec::new();

        self.backend.list(ObjectKind::Prune, &mut |key| {
            if let ObjectKey::Prune(id) = key {
                ids.push(*id.as_bytes());
            }

            Ok(())
        })?;

        ids.sort_unstable();

        let parts: Vec<&[u8]> = ids.iter().map(|id| id.as_slice()).collect();
        let digest = crypto::keyed_digest(self.keys.id_key(), &parts);

        Ok(digest[..16].try_into().expect("digest is 32 bytes"))
    }

    /// Claims the repository for a running backup, which holds chunks
    /// that no snapshot names yet.
    pub fn take_lock(&self, lock: &Lock) -> Result<(), Error> {
        let mut plaintext = Vec::new();
        ciborium::into_writer(lock, &mut plaintext)?;

        let sealed = crypto::seal(&self.keys, &lock_aad(lock.id), &plaintext)?;
        self.put(&ObjectKey::Lock(lock.id), &sealed)?;

        Ok(())
    }

    /// Gives a lock back. Idempotent, like every delete.
    pub fn release_lock(&self, id: LockId) -> Result<(), Error> {
        self.backend.delete(&ObjectKey::Lock(id))
    }

    /// Every lock currently held on the repository.
    pub fn locks(&self) -> Result<Vec<Lock>, Error> {
        let mut ids = Vec::new();

        self.backend.list(ObjectKind::Lock, &mut |key| {
            if let ObjectKey::Lock(id) = key {
                ids.push(id);
            }

            Ok(())
        })?;

        let mut locks = Vec::new();

        for id in ids {
            // A lock released while we were listing is simply gone
            let Some(sealed) = self.backend.get(&ObjectKey::Lock(id))? else {
                continue;
            };

            let plaintext = crypto::open(&self.keys, &lock_aad(id), &sealed)?;
            locks.push(ciborium::from_reader(plaintext.as_slice())?);
        }

        Ok(locks)
    }

    /// Removes a chunk. Only [`crate::prune`]-style sweeping should call
    /// this, and only for chunks no snapshot references.
    pub fn delete_chunk(&self, id: ChunkId) -> Result<(), Error> {
        self.backend.delete(&ObjectKey::Chunk(id))
    }

    /// The ids of every chunk the repository holds.
    pub fn chunks(&self) -> Result<Vec<ChunkId>, Error> {
        let mut ids = Vec::new();

        self.backend.list(ObjectKind::Chunk, &mut |key| {
            if let ObjectKey::Chunk(id) = key {
                ids.push(id);
            }

            Ok(())
        })?;

        Ok(ids)
    }

    /// The ids of every snapshot in the repository, in no particular
    /// order.
    pub fn snapshots(&self) -> Result<Vec<SnapshotId>, Error> {
        let mut ids = Vec::new();

        self.backend.list(ObjectKind::Snapshot, &mut |key| {
            if let ObjectKey::Snapshot(id) = key {
                ids.push(id);
            }

            Ok(())
        })?;

        Ok(ids)
    }

    /// Direct access to the underlying backend, for maintenance
    /// operations that work below the repository abstraction.
    pub fn backend(&self) -> &dyn Backend {
        &*self.backend
    }

    /// Sealed bytes this handle has handed to the backend since it was
    /// opened.
    ///
    /// An upper bound on what it added to the store, and the cheap way
    /// to know how a store has grown under a run without asking the
    /// store again: objects are write-once, so a put of something
    /// already there is counted here and adds nothing over there.
    pub fn bytes_written(&self) -> u64 {
        self.written.load(std::sync::atomic::Ordering::Acquire)
    }

    /// Writes an object, counting what it hands over. Every write a
    /// repository makes goes through here, so that count is the whole
    /// of what this handle has put into the store.
    fn put(&self, key: &ObjectKey, sealed: &[u8]) -> Result<(), Error> {
        self.backend.put(key, sealed)?;
        self.written
            .fetch_add(sealed.len() as u64, std::sync::atomic::Ordering::AcqRel);

        Ok(())
    }
}

fn chunk_aad(id: ChunkId) -> Vec<u8> {
    let mut aad = b"chunk:".to_vec();
    aad.extend_from_slice(id.as_bytes());

    aad
}

fn snapshot_aad(id: SnapshotId) -> Vec<u8> {
    let mut aad = b"snapshot:".to_vec();
    aad.extend_from_slice(id.as_bytes());

    aad
}

fn prune_aad(id: PruneId) -> Vec<u8> {
    let mut aad = b"prune:".to_vec();
    aad.extend_from_slice(id.as_bytes());

    aad
}

fn lock_aad(id: LockId) -> Vec<u8> {
    let mut aad = b"lock:".to_vec();
    aad.extend_from_slice(id.as_bytes());

    aad
}