//! 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()
}

/// Who wrote the changes after `since`: the authors their writers set (a node's ID,
/// `replica::authored`), each once; and whether any change has none (a snapshot, or a
/// change of a node that set none).
pub(crate) fn authors_since(doc: &Document, since: &[ChangeHash]) -> (Vec<String>, bool) {
    let mut authors: Vec<String> = Vec::new();
    let mut unknown = false;
    for change in doc.get_changes_meta(since) {
        match change
            .author
            .as_ref()
            .and_then(|author| std::str::from_utf8(author.as_bytes()).ok())
        {
            Some(author) if !authors.iter().any(|known| known == author) => {
                authors.push(author.to_owned());
            }
            Some(_) => {}
            None => unknown = true,
        }
    }
    (authors, unknown)
}

/// 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` and `from`, whatever
/// else it has. None if `doc` lacks part of either: without one of `from`, the delta
/// would carry that one's history too, where `to` shares it, and differ from another
/// node's (a download takes chunks from several, checked against one's manifest).
pub(crate) fn delta_bytes(
    doc: &Document,
    from: &[ChangeHash],
    to: &[ChangeHash],
) -> Option<Vec<u8>> {
    if !missing(doc, to).is_empty() || !missing(doc, from).is_empty() {
        return None;
    }
    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());
        let heads = doc.get_heads();
        assert!(delta_bytes(&doc, &[], &[ChangeHash([7; 32])]).is_none());
        assert!(
            delta_bytes(&doc, &[ChangeHash([7; 32])], &heads).is_none(),
            "nor without where it starts"
        );
    }

    /// 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());
    }

    /// The authors of the changes after some heads: each node that wrote some, once,
    /// however their changes were merged; a change without an author is told apart.
    #[test]
    fn authors_of_the_changes_since() {
        use crate::clock::UnixMillis;
        use crate::document::replica::{Writer, transact};
        let write = |doc: &mut Document, node: &str, key: &str| {
            let writer = Writer { node, user: None };
            transact(doc, UnixMillis(1), &writer, |doc| doc.put(ROOT, key, 1)).unwrap();
        };
        let mut base = Document::new();
        write(&mut base, "node-a", "base");
        let since = base.get_heads();
        assert_eq!(authors_since(&base, &since), (Vec::new(), false));

        let mut a = base.fork();
        write(&mut a, "node-a", "a1");
        write(&mut a, "node-a", "a2");
        let mut b = base.fork();
        write(&mut b, "node-b", "b1");
        let mut merged = a.clone();
        merged.merge(&mut b).unwrap();
        let (mut authors, unknown) = authors_since(&merged, &since);
        authors.sort();
        assert_eq!(authors, ["node-a", "node-b"]);
        assert!(!unknown);

        // A change no node authored (a snapshot's, say).
        let mut snapshot = Document::new();
        snapshot.merge(&mut merged.clone()).unwrap();
        change(&mut snapshot, |doc| doc.put(ROOT, "s", 1).unwrap());
        let (_, unknown) = authors_since(&snapshot, &merged.get_heads());
        assert!(unknown);
    }
}