//! 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};

/// Whether `doc` has every change of `heads`.
pub(crate) fn holds_all(doc: &Document, heads: &[ChangeHash]) -> bool {
    heads
        .iter()
        .all(|head| doc.get_change_by_hash(head).is_some())
}

/// Whether the history of `theirs` (of them, those `doc` has) holds all of `ours`
/// (heads of `doc`): a peer at `theirs` has every change we have.
pub(crate) fn covers(doc: &Document, theirs: &[ChangeHash], ours: &[ChangeHash]) -> bool {
    let known: Vec<ChangeHash> = theirs
        .iter()
        .copied()
        .filter(|head| doc.get_change_by_hash(head).is_some())
        .collect();
    if known.is_empty() {
        return ours.is_empty();
    }
    let beyond: std::collections::HashSet<ChangeHash> = doc
        .get_changes_meta(&known)
        .into_iter()
        .map(|change| change.hash)
        .collect();
    ours.iter().all(|head| !beyond.contains(head))
}

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

/// The changes after `since`, walked once: their hashes; who wrote them (the authors
/// their writers set, a node's ID, `replica::authored`), each once, with the hashes of
/// their changes, and whether any has none (a snapshot, or a change of a node that set
/// none); and when they were committed, in Unix milliseconds by their writers' clocks
/// (`Replica::transact`), changes without a time (snapshots) left out.
pub(crate) struct Since {
    pub(crate) hashes: Vec<ChangeHash>,
    pub(crate) authors: Vec<(String, Vec<ChangeHash>)>,
    pub(crate) unauthored: bool,
    pub(crate) times: Vec<i64>,
}

pub(crate) fn since(doc: &Document, since: &[ChangeHash]) -> Since {
    let mut found = Since {
        hashes: Vec::new(),
        authors: Vec::new(),
        unauthored: false,
        times: Vec::new(),
    };
    for change in doc.get_changes_meta(since) {
        found.hashes.push(change.hash);
        if change.timestamp > 0 {
            found.times.push(change.timestamp);
        }
        match change
            .author
            .as_ref()
            .and_then(|author| std::str::from_utf8(author.as_bytes()).ok())
        {
            Some(author) => match found.authors.iter_mut().find(|(known, _)| known == author) {
                Some((_, hashes)) => hashes.push(change.hash),
                None => found.authors.push((author.to_owned(), vec![change.hash])),
            },
            None => found.unauthored = true,
        }
    }
    found
}

/// When the changes after `since` were committed (`Since::times`).
pub(crate) fn commit_times(doc: &Document, since: &[ChangeHash]) -> Vec<i64> {
    self::since(doc, since).times
}

/// Who wrote the changes after `since` (`Since::authors`, `Since::unauthored`).
#[cfg(test)]
pub(crate) fn authors_since(
    doc: &Document,
    since: &[ChangeHash],
) -> (Vec<(String, Vec<ChangeHash>)>, bool) {
    let found = self::since(doc, since);
    (found.authors, found.unauthored)
}

/// 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<()> {
    if let Some(misplaced) = misplaced_author(bytes) {
        bail!("Refused the downloaded changes: {misplaced}");
    }
    doc.load_incremental(bytes)
        .context("Loading the downloaded changes")?;
    Ok(())
}

/// A change that names an author although it isn't its actor's first (`misplaced_author`).
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct MisplacedAuthor {
    pub(crate) author: Vec<u8>,
    pub(crate) actor: super::ActorId,
    pub(crate) seq: u64,
}

impl std::fmt::Display for MisplacedAuthor {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(
            f,
            "change {} of actor {} names an author ({}), which only an actor's first change may",
            self.seq,
            self.actor,
            String::from_utf8_lossy(&self.author)
        )
    }
}

/// The first change among those in `data` (chunks as `load_incremental` takes them:
/// changes, compressed changes, bundles, documents) that names an author although it
/// isn't its actor's first change. Automerge records an author on an actor's first change
/// only, and applying a later one that carries an author trips an assertion in Automerge
/// 0.12 (`ChangeGraph::add_changes`) halfway through updating the document: the task
/// panics, and the document is left inconsistent. No honest node makes such a change, so
/// whatever comes from elsewhere is looked at here first, until Automerge refuses such
/// changes itself. Data that doesn't parse is left to Automerge, which refuses it.
///
/// A document chunk is loaded on its own to read its changes (no assertion on that
/// path): about as much work again as applying it, paid for documents in sync messages
/// (Automerge sends one when a peer lacks a third of the history or more).
pub(crate) fn misplaced_author(mut data: &[u8]) -> Option<MisplacedAuthor> {
    const MAGIC: [u8; 4] = [0x85, 0x6f, 0x4a, 0x83];
    // Magic bytes, checksum, chunk type, then the body's length (LEB128) and the body.
    const HEADER: usize = 9;
    while data.len() > HEADER && data[..4] == MAGIC {
        let kind = data[8];
        let (length, read) = uleb128(&data[HEADER..])?;
        let end = HEADER
            .checked_add(read)?
            .checked_add(usize::try_from(length).ok()?)?;
        let chunk = data.get(..end)?;
        data = &data[end..];
        let found = match kind {
            // A change, or a compressed one.
            1 | 2 => Change::try_from(chunk)
                .ok()
                .and_then(|change| misplaced_in(&change)),
            3 => automerge::Bundle::try_from(chunk)
                .ok()
                .and_then(|bundle| bundle.to_changes().ok())
                .and_then(|changes| changes.iter().find_map(misplaced_in)),
            0 => Document::load(chunk).ok().and_then(|doc| {
                doc.get_changes_meta(&[]).into_iter().find_map(|meta| {
                    (meta.seq != 1)
                        .then(|| author_footer(&meta.extra))
                        .flatten()
                        .map(|author| MisplacedAuthor {
                            author,
                            actor: meta.actor.into_owned(),
                            seq: meta.seq,
                        })
                })
            }),
            _ => None,
        };
        if found.is_some() {
            return found;
        }
    }
    None
}

/// `change`, if it names an author but isn't its actor's first (`misplaced_author`).
fn misplaced_in(change: &Change) -> Option<MisplacedAuthor> {
    let author = change.author()?;
    (change.seq() != 1).then(|| MisplacedAuthor {
        author: author.as_bytes().to_vec(),
        actor: change.actor_id().clone(),
        seq: change.seq(),
    })
}

/// The author in a change's extra bytes: footers, each its kind (LEB128; 1 is the
/// author), its length (LEB128) and its bytes.
fn author_footer(mut extra: &[u8]) -> Option<Vec<u8>> {
    while !extra.is_empty() {
        let (kind, read) = uleb128(extra)?;
        let (length, more) = uleb128(&extra[read..])?;
        let start = read + more;
        let end = start.checked_add(usize::try_from(length).ok()?)?;
        let bytes = extra.get(start..end)?;
        if kind == 1 {
            return Some(bytes.to_vec());
        }
        extra = &extra[end..];
    }
    None
}

/// An unsigned LEB128 number at the start of `bytes`, and how many bytes it took.
fn uleb128(bytes: &[u8]) -> Option<(u64, usize)> {
    let mut value = 0u64;
    for (i, byte) in bytes.iter().enumerate().take(10) {
        value |= u64::from(byte & 0x7f) << (7 * i);
        if byte & 0x80 == 0 {
            return Some((value, i + 1));
        }
    }
    None
}

#[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));
        });
    }

    /// Heads cover ours when their history holds all of ours: as far as ours, or
    /// further; not when behind or beside; heads we don't know count for nothing.
    #[test]
    fn heads_that_cover_ours() {
        let mut doc = Document::new();
        change(&mut doc, |doc| doc.put(ROOT, "a", 1_i64).unwrap());
        let early = doc.get_heads();
        change(&mut doc, |doc| doc.put(ROOT, "b", 2_i64).unwrap());
        let ours = doc.get_heads();
        let mut further = doc.fork();
        change(&mut further, |doc| doc.put(ROOT, "c", 3_i64).unwrap());
        doc.merge(&mut further.clone()).unwrap();
        let unknown = ChangeHash([7; 32]);

        assert!(covers(&doc, &ours, &ours));
        assert!(covers(&doc, &further.get_heads(), &ours), "further");
        assert!(!covers(&doc, &early, &ours), "behind");
        assert!(!covers(&doc, &[unknown], &ours), "unknown");
        assert!(
            covers(&doc, &[unknown, ours[0]], &ours),
            "the known one does"
        );
    }

    #[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();
        let counts: Vec<(&str, usize)> = authors
            .iter()
            .map(|(author, hashes)| (author.as_str(), hashes.len()))
            .collect();
        assert_eq!(counts, [("node-a", 2), ("node-b", 1)]);
        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);
    }

    /// An actor without an author, and its two changes: the second as made, and with an
    /// author footer it mustn't carry.
    fn misplaced() -> (Vec<u8>, Vec<u8>, Vec<u8>) {
        let mut doc = Document::new().with_actor(ActorId::from(b"writer".to_vec()));
        change(&mut doc, |tx| tx.put(ROOT, "x", 1).unwrap());
        change(&mut doc, |tx| tx.put(ROOT, "x", 2).unwrap());
        let changes = doc.get_changes(&[]);
        let (first, second) = (
            changes[0].raw_bytes().to_vec(),
            changes[1].raw_bytes().to_vec(),
        );
        let crafted = crate::document::testing::with_author_footer(&second, b"mallory");
        (first, second, crafted)
    }

    #[test]
    fn an_author_on_a_later_change_is_found() {
        let (first, second, crafted) = misplaced();
        assert_eq!(misplaced_author(&[first.clone(), second].concat()), None);
        let found = misplaced_author(&[first.clone(), crafted.clone()].concat()).unwrap();
        assert_eq!(found.seq, 2);
        assert_eq!(found.author, b"mallory");
        assert_eq!(found.actor, ActorId::from(b"writer".to_vec()));
        // Honest authors (on first changes) and whole documents pass.
        let mut authored = Document::new();
        authored.set_author(Some(automerge::Author::from(b"node".to_vec())));
        change(&mut authored, |tx| tx.put(ROOT, "a", 1).unwrap());
        change(&mut authored, |tx| tx.put(ROOT, "a", 2).unwrap());
        assert_eq!(misplaced_author(&authored.save()), None);
        let bytes: Vec<u8> = authored
            .get_changes(&[])
            .iter()
            .flat_map(|change| change.raw_bytes().to_vec())
            .collect();
        assert_eq!(misplaced_author(&bytes), None);
        // What isn't chunks is left to Automerge.
        assert_eq!(misplaced_author(b"not a chunk"), None);
    }

    #[test]
    fn a_delta_with_a_misplaced_author_is_refused_and_the_document_kept() {
        let (first, _, crafted) = misplaced();
        let mut doc = Document::new();
        load_delta(&mut doc, &first).unwrap();
        let heads = doc.get_heads();
        assert!(load_delta(&mut doc, &crafted).is_err());
        assert_eq!(doc.get_heads(), heads);
        // Still whole: it saves and loads.
        crate::document::load(&doc.save()).unwrap();
    }
}