//! A model of two neighbours' sync sessions over a channel that loses, duplicates and
//! reorders, with writes on both sides: the session as `swarm::sync` drives it (pushes
//! and messages taken as delivered where level, answers awaited otherwise, reply
//! timeouts, a STATUS now and then that makes the one behind pull), without the swarm.
//! Seeded (`testrand::for_seeds`): a failure names its seed.

use std::cmp::Reverse;
use std::collections::{BTreeMap, BinaryHeap, HashMap};
use std::time::Duration;

use automerge::{ROOT, ReadDoc, transaction::Transactable};
use tokio::time::Instant;

use super::{Received, SyncSession, decode};
use crate::backoff::Backoff;
use crate::document::testing::try_change;
use crate::document::{ChangeHash, Document};
use crate::swarm::throughput::SyncBytes;
use crate::testrand::{Rng, for_seeds};
use crate::transport::SyncPayload;

/// One step of model time.
const TICK: Duration = Duration::from_millis(10);
/// Writes happen in the first ticks, over a lossy channel; then it turns fair.
const WRITING: u32 = 300;
/// How long after that it may take to be quiet (nothing in the channel, nothing to
/// say, nothing awaited), in ticks: a generous bound for the reply timeouts' backoff.
const QUIET_WITHIN: u32 = 20_000;
/// How often each side says its heads (the keepalive STATUS), in ticks.
const STATUS_EVERY: u32 = 50;

#[derive(Clone)]
enum Envelope {
    Message {
        payload: bytes::Bytes,
        delivered: bool,
    },
    Changes {
        heads: Vec<ChangeHash>,
        changes: Vec<Vec<u8>>,
    },
    Status {
        heads: Vec<ChangeHash>,
    },
}

struct Side {
    doc: Document,
    session: SyncSession,
    /// Who brought each change first: `true` if the other side (else written here).
    from_peer: HashMap<ChangeHash, bool>,
}

struct Channel {
    queue: BinaryHeap<Reverse<(u32, u64, usize)>>,
    envelopes: HashMap<u64, Envelope>,
    seq: u64,
    loss: f64,
    dup: f64,
}

impl Channel {
    fn send(&mut self, rng: &mut Rng, now: u32, to: usize, envelope: Envelope) {
        let copies = if rng.chance(self.loss) {
            0
        } else if rng.chance(self.dup) {
            2
        } else {
            1
        };
        for _ in 0..copies {
            self.seq += 1;
            // 1 to 8 ticks: later messages overtake earlier ones.
            let at = now + rng.between(1, 8) as u32;
            self.envelopes.insert(self.seq, envelope.clone());
            self.queue.push(Reverse((at, self.seq, to)));
        }
    }

    fn due(&mut self, now: u32) -> Vec<(usize, Envelope)> {
        let mut due = Vec::new();
        while let Some(Reverse((at, seq, to))) = self.queue.peek().copied() {
            if at > now {
                break;
            }
            self.queue.pop();
            if let Some(envelope) = self.envelopes.remove(&seq) {
                due.push((to, envelope));
            }
        }
        due
    }
}

/// Whether `envelope` carries the whole document saved (Automerge's chunk of type 0, a
/// document, after the magic bytes and the checksum), not changes.
fn saved_document(envelope: &Envelope) -> bool {
    let Envelope::Message { payload, .. } = envelope else {
        return false;
    };
    let message = decode(&SyncPayload(payload.clone())).expect("decodes");
    message
        .0
        .changes
        .iter()
        .any(|chunk| chunk.get(8) == Some(&0))
}

/// What a document holds, by key: what must be equal on both sides at the end, and
/// across runs that differ only in how changes travelled.
fn contents(doc: &Document) -> BTreeMap<String, String> {
    doc.keys(ROOT)
        .map(|key| {
            let value = doc
                .get(ROOT, &key)
                .ok()
                .flatten()
                .map(|(value, _)| value.to_string())
                .unwrap_or_default();
            (key, value)
        })
        .collect()
}

/// Runs the model from `seed` (the writes, and apart from them the channel); pushes
/// allowed below `push_below` bytes (0: never). Returns both documents' contents, once
/// quiet.
fn run(seed: u64, push_below: usize) -> (BTreeMap<String, String>, BTreeMap<String, String>) {
    let mut writes = Rng::new(seed);
    let mut net = Rng::new(seed ^ 0x5eed);
    let base = Instant::now();
    let at = |tick: u32| base + TICK * tick;
    let bytes = SyncBytes::default();
    let mut sides = [0u8, 1].map(|which| Side {
        doc: Document::new().with_actor(automerge::ActorId::from([which + 1; 16])),
        session: SyncSession::new(Backoff::new(
            Duration::from_secs(1),
            Duration::from_secs(30),
        )),
        from_peer: HashMap::new(),
    });
    let mut channel = Channel {
        queue: BinaryHeap::new(),
        envelopes: HashMap::new(),
        seq: 0,
        loss: 0.2,
        dup: 0.1,
    };

    let mut tick = 0u32;
    let mut quiet_for = 0u32;
    loop {
        tick += 1;
        assert!(
            tick < WRITING + QUIET_WITHIN,
            "seed {seed}: not quiet {QUIET_WITHIN} ticks after the writes"
        );
        if tick == WRITING {
            channel.loss = 0.0;
            channel.dup = 0.0;
        }
        let now = at(tick);
        let mut busy = false;

        // Deliveries, as `swarm::sync::handle_sync` makes them.
        for (to, envelope) in channel.due(tick) {
            let side = &mut sides[to];
            let message = match envelope {
                Envelope::Status { heads } => {
                    // Behind the peer: pull (`relay::pull_if_behind`).
                    if heads
                        .iter()
                        .any(|head| side.doc.get_change_by_hash(head).is_none())
                    {
                        side.session.pull_from(heads);
                    } else if side.session.reply.is_some() && heads == side.doc.get_heads() {
                        side.session.reply_settled();
                    }
                    continue;
                }
                Envelope::Changes { heads, changes } => {
                    (Received::pushed(heads, changes).expect("decodes"), true)
                }
                Envelope::Message { payload, delivered } => {
                    (decode(&SyncPayload(payload)).expect("decodes"), delivered)
                }
            };
            busy = true;
            let (message, delivered) = message;
            let before = side.doc.get_heads();
            side.session
                .receive(&mut side.doc, message)
                .expect("applies");
            side.session.reply_settled();
            side.session.resend_if_needed(&side.doc, "peer", now);
            let after = side.doc.get_heads();
            if delivered {
                side.session.answered_as_delivered(&side.doc, &after);
            }
            if after != before {
                for change in side.doc.get_changes_meta(&before) {
                    side.from_peer.entry(change.hash).or_insert(true);
                }
                side.session.came_from_peer(&side.doc, &before);
            }
        }

        // Writes.
        if tick < WRITING {
            for (which, side) in sides.iter_mut().enumerate() {
                if writes.chance(0.05) {
                    let key = format!("{which}-{tick}");
                    let value = writes.next_u64() as i64;
                    let before = side.doc.get_heads();
                    try_change(&mut side.doc, |doc| doc.put(ROOT, key, value)).unwrap();
                    for change in side.doc.get_changes_meta(&before) {
                        side.from_peer.insert(change.hash, false);
                    }
                }
            }
        }

        // Sends, as `swarm::sync::deliver_sync_to_peer` makes them.
        for which in 0..2 {
            let to = 1 - which;
            let side = &mut sides[which];
            side.session.take_overdue_reply(now);
            let level = side.session.reply.is_none() && side.session.have_all_theirs(&side.doc);
            let pushed = (level && push_below > 0)
                .then(|| side.session.push(&side.doc, push_below))
                .flatten();
            let envelope =
                match pushed {
                    Some((heads, changes)) => Some(Envelope::Changes { heads, changes }),
                    None => side.session.generate(&side.doc, now, bytes).map(|payload| {
                        Envelope::Message {
                            payload: payload.0,
                            delivered: level,
                        }
                    }),
                };
            let Some(envelope) = envelope else {
                continue;
            };
            busy = true;
            if level {
                let heads = side.doc.get_heads();
                side.session.assume_delivered(&heads);
            }
            // What it carries for the first time (what Automerge chose to send): none of
            // it may have come from the peer. Unless it is the whole document saved, which
            // Automerge sends a peer that lacks more than a third of it: the peer's own
            // changes go along then, by design, and count as sent.
            let carried = side
                .session
                .sending
                .as_ref()
                .map_or_else(Vec::new, |sending| sending.sent.clone());
            let carried = if saved_document(&envelope) {
                Vec::new()
            } else {
                carried
            };
            side.session.send_succeeded(side.session.id());
            // No echo: nothing goes back to the side it came from.
            for hash in carried {
                assert_ne!(
                    side.from_peer.get(&hash),
                    Some(&true),
                    "seed {seed}: a change went back to the side it came from"
                );
            }
            channel.send(&mut net, tick, to, envelope);
        }

        // The keepalive STATUS.
        if tick % STATUS_EVERY == 0 {
            for which in 0..2 {
                let heads = sides[which].doc.get_heads();
                channel.send(&mut net, tick, 1 - which, Envelope::Status { heads });
            }
        }

        let awaiting = sides.iter().any(|side| side.session.reply.is_some());
        let level = sides[0].doc.get_heads() == sides[1].doc.get_heads();
        if tick > WRITING && !busy && !awaiting && level {
            quiet_for += 1;
            // Quiet through two keepalives: nobody had anything to say.
            if quiet_for > 2 * STATUS_EVERY {
                break;
            }
        } else {
            quiet_for = 0;
        }
    }

    // Nothing awaited, nothing in the transport's hands, no offer out.
    for side in &sides {
        assert!(
            side.session.reply.is_none(),
            "seed {seed}: an answer still awaited"
        );
        assert!(
            side.session.sending.is_none(),
            "seed {seed}: a send not finished"
        );
        assert!(
            !side.session.delta_offered,
            "seed {seed}: an offer still out"
        );
    }
    let [a, b] = sides;
    (contents(&a.doc), contents(&b.doc))
}

/// Both sides end with the same document, without a change going back to where it came
/// from, with nothing awaited; with pushes as without.
#[test]
fn sessions_converge_over_a_lossy_channel() {
    for_seeds(8, |rng| {
        let seed = rng.next_u64();
        let (a, b) = run(seed, 1024);
        assert_eq!(a, b, "seed {seed}: the sides differ");
        assert!(!a.is_empty(), "seed {seed}: nothing was written");
        let (unpushed, _) = run(seed, 0);
        assert_eq!(a, unpushed, "seed {seed}: pushes made a difference");
    });
}