//! Changes as bytes, to send and to load (`swarm::blob` moves them): one change by its
//! hash, checked against it when it arrives, and a delta, the changes between two sets
//! of heads, the same bytes on every node holding them.

use std::collections::{HashMap, HashSet};

use anyhow::{Context, Result, anyhow, bail};

use super::{Change, ChangeHash, Document, ReadDoc};

/// Those of `heads` that `doc` doesn't have.
pub(crate) fn missing(doc: &impl ReadDoc, heads: &[ChangeHash]) -> Vec<ChangeHash> {
    heads
        .iter()
        .copied()
        .filter(|head| doc.get_change_by_hash(head).is_none())
        .collect()
}

/// When the changes after `since` were committed, in Unix milliseconds by their
/// writers' clocks (`Replica::transact`); changes without a time (snapshots) left out.
pub(crate) fn commit_times(doc: &Document, since: &[ChangeHash]) -> Vec<i64> {
    doc.get_changes(since)
        .iter()
        .map(|change| change.timestamp())
        .filter(|&time| time > 0)
        .collect()
}

/// The change `hash` as the document encodes it; None if `doc` doesn't have it.
pub(crate) fn change_bytes(doc: &Document, hash: &ChangeHash) -> Option<Vec<u8>> {
    doc.get_change_by_hash(hash)
        .map(|change| change.raw_bytes().to_vec())
}

/// The change in `bytes`, if it is the change `hash`.
pub(crate) fn change_from_bytes(bytes: Vec<u8>, hash: ChangeHash) -> Result<Change> {
    let change = Change::from_bytes(bytes)
        .map_err(|error| anyhow!("change {hash} doesn't load: {error}"))?;
    if change.hash() != hash {
        bail!("change has hash {}, not {hash}", change.hash());
    }
    Ok(change)
}

/// The bytes of the delta from `from` to `to` in `doc`: the changes in the history of
/// `to` but not of `from`, each as encoded, sorted by hash and concatenated
/// (`load_delta` reads that). The same on every node holding `to`, whatever else it
/// has. None if `doc` lacks part of `to`.
pub(crate) fn delta_bytes(
    doc: &Document,
    from: &[ChangeHash],
    to: &[ChangeHash],
) -> Option<Vec<u8>> {
    if !missing(doc, to).is_empty() {
        return None;
    }
    // Heads of `from` we don't have can't hide anything we have.
    let from: Vec<ChangeHash> = from
        .iter()
        .copied()
        .filter(|head| doc.get_change_by_hash(head).is_some())
        .collect();
    let since: HashMap<ChangeHash, (Vec<ChangeHash>, Vec<u8>)> = doc
        .get_changes(&from)
        .iter()
        .map(|change| {
            (
                change.hash(),
                (change.deps().to_vec(), change.raw_bytes().to_vec()),
            )
        })
        .collect();
    // Of those, the ones `to` depends on: what we have beyond `to` stays out.
    let mut included: HashSet<ChangeHash> = HashSet::new();
    let mut pending: Vec<ChangeHash> = to.to_vec();
    while let Some(hash) = pending.pop() {
        if let Some((deps, _)) = since.get(&hash)
            && included.insert(hash)
        {
            pending.extend(deps.iter().copied());
        }
    }
    let mut hashes: Vec<ChangeHash> = included.into_iter().collect();
    hashes.sort_unstable();
    let mut bytes = Vec::new();
    for hash in hashes {
        bytes.extend_from_slice(&since[&hash].1);
    }
    Some(bytes)
}

/// Loads a delta (`delta_bytes`) into `doc`.
pub(crate) fn load_delta(doc: &mut Document, bytes: &[u8]) -> Result<()> {
    doc.load_incremental(bytes)
        .context("Loading the downloaded changes")?;
    Ok(())
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::document::testing::change;
    use automerge::{ActorId, ROOT, transaction::Transactable};

    fn random_write(rng: &mut crate::testrand::Rng, doc: &mut Document) {
        let key = format!("k{}", rng.below(6));
        let value = rng.below(100) as i64;
        change(doc, |doc| doc.put(ROOT, key, value).unwrap());
    }

    fn sorted(mut heads: Vec<ChangeHash>) -> Vec<ChangeHash> {
        heads.sort_unstable();
        heads
    }

    /// Over random histories: two nodes that reached the same heads through different
    /// merge orders, one with more on top, produce the same bytes for a delta to those
    /// heads; a document at `from` that loads them is at `to`.
    #[test]
    fn delta_bytes_are_the_same_everywhere_and_complete() {
        crate::testrand::for_seeds(200, |rng| {
            let actor = |i: u8| ActorId::from(&[i; 16][..]);
            let mut base = Document::new().with_actor(actor(0));
            for _ in 0..rng.between(0, 5) {
                random_write(rng, &mut base);
            }
            let from = base.get_heads();
            let mut writers: Vec<Document> = (1..=rng.between(1, 3))
                .map(|i| base.fork().with_actor(actor(i as u8)))
                .collect();
            for _ in 0..rng.between(1, 15) {
                let writer = rng.below(writers.len());
                random_write(rng, &mut writers[writer]);
            }
            let mut one = base.fork().with_actor(actor(10));
            let mut two = base.fork().with_actor(actor(11));
            for writer in writers.iter_mut() {
                one.merge(writer).unwrap();
            }
            for writer in writers.iter_mut().rev() {
                two.merge(writer).unwrap();
            }
            let to = one.get_heads();
            assert_eq!(sorted(to.clone()), sorted(two.get_heads()));
            // Beyond `to`: must stay out of the delta.
            random_write(rng, &mut two);

            let bytes = delta_bytes(&one, &from, &to).unwrap();
            assert_eq!(delta_bytes(&two, &from, &to).unwrap(), bytes);

            let mut behind = base.fork();
            load_delta(&mut behind, &bytes).unwrap();
            assert_eq!(sorted(behind.get_heads()), sorted(to));
        });
    }

    #[test]
    fn no_delta_without_its_heads() {
        let mut doc = Document::new();
        change(&mut doc, |doc| doc.put(ROOT, "k", 1_i64).unwrap());
        assert!(delta_bytes(&doc, &[], &[ChangeHash([7; 32])]).is_none());
    }

    /// A change loads from its bytes only as the change it was asked for.
    #[test]
    fn a_change_is_checked_against_its_hash() {
        let mut doc = Document::new();
        change(&mut doc, |doc| doc.put(ROOT, "k", 1_i64).unwrap());
        let hash = doc.get_heads()[0];
        let bytes = change_bytes(&doc, &hash).unwrap();
        assert_eq!(change_from_bytes(bytes.clone(), hash).unwrap().hash(), hash);
        assert!(change_from_bytes(bytes, ChangeHash([7; 32])).is_err());
        assert!(change_bytes(&doc, &ChangeHash([7; 32])).is_none());
    }
}