//! Benchmarks, as ignored tests (`cargo benchmarks`, `docs/development.md`): each times
//! what it measures in release builds and prints it; none asserts anything.
//!
//! `change_publishing`: what a change costs until the change publisher knows the paths
//! it touched (`swarm::changes`). Recorded as it is applied (`Replica::logged`, the
//! patch log), against diffing after it (`store::changed`, what the publisher did
//! before, and still does where the log doesn't cover it). Each workload is timed both
//! ways on the same input.

use std::time::{Duration, Instant};

use automerge::sync::{self, SyncDoc};
use automerge::transaction::Transactable;
use automerge::{ROOT, ReadDoc, ScalarValue};

use super::replica::{Replica, Writer, transact};
use super::store::{self, InLists};
use super::{ChangeHash, Document};
use crate::clock::UnixMillis;
use crate::document::replica::Epoch;

const WRITER: Writer = Writer {
    node: "continuum:bench",
    user: None,
};

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

/// A document with `keys` keys under `data`, each written by a change of its own.
fn document(keys: usize) -> Document {
    let mut doc = Document::new();
    let mut map = None;
    for key in 0..keys {
        transact(&mut doc, UnixMillis(1), &WRITER, |doc| {
            let data = match &map {
                Some(data) => data,
                None => map.insert(doc.put_object(ROOT, "data", automerge::ObjType::Map)?),
            };
            doc.put(data, format!("k{key}"), ScalarValue::Int(key as i64))
        })
        .expect("write");
    }
    doc
}

/// The median of `runs` timings of `f`.
fn median(runs: usize, mut f: impl FnMut() -> Duration) -> Duration {
    let mut times: Vec<Duration> = (0..runs).map(|_| f()).collect();
    times.sort_unstable();
    times[runs / 2]
}

fn report(workload: &str, logged: Duration, diffed: Duration) {
    println!(
        "{workload:<44} logged {:>10.3} ms   diffed {:>10.3} ms   ({:.2}×)",
        logged.as_secs_f64() * 1e3,
        diffed.as_secs_f64() * 1e3,
        diffed.as_secs_f64() / logged.as_secs_f64().max(f64::MIN_POSITIVE),
    );
}

/// `writes` local writes of one key each into a document of `keys` keys, the changed
/// paths taken after each (as the publisher does after every change it is told of).
fn local_writes(keys: usize, writes: usize) {
    let base = document(keys);
    let write = |doc: &mut automerge::transaction::Transaction<'_>, i: usize| {
        let data = doc.get(ROOT, "data")?.map(|(_, id)| id).expect("data");
        doc.put(
            &data,
            format!("k{}", i % keys),
            ScalarValue::Int(-(i as i64)),
        )
    };

    let logged = median(5, || {
        let mut replica = Replica::new(Epoch::GENESIS, base.clone());
        let started = Instant::now();
        for i in 0..writes {
            let before = sorted(replica.doc.get_heads());
            replica
                .transact(UnixMillis(2), &WRITER, |doc| write(doc, i))
                .expect("write");
            std::hint::black_box(replica.changed_since(&before, InLists::Index));
        }
        started.elapsed()
    });
    let diffed = median(5, || {
        let mut doc = base.clone();
        let started = Instant::now();
        for i in 0..writes {
            let before = sorted(doc.get_heads());
            transact(&mut doc, UnixMillis(2), &WRITER, |doc| write(doc, i)).expect("write");
            let after = sorted(doc.get_heads());
            std::hint::black_box(store::changed(&doc, &before, &after, InLists::Index));
        }
        started.elapsed()
    });
    report(
        &format!("{writes} local writes, {keys} keys"),
        logged,
        diffed,
    );
}

/// One sync message carrying `changes` changes to a node that has none of them, and
/// the changed paths after it.
fn sync_message(changes: usize) {
    let sender = document(changes);
    // The receiver opens; the sender answers with everything it lacks.
    let mut receiver_state = sync::State::new();
    let opening = Document::new()
        .generate_sync_message(&mut receiver_state)
        .expect("an opening");
    let mut sender_doc = sender.clone();
    let mut sender_state = sync::State::new();
    sender_doc
        .receive_sync_message(&mut sender_state, opening)
        .expect("opening");
    let message = sender_doc
        .generate_sync_message(&mut sender_state)
        .expect("the changes")
        .encode();

    let logged = median(5, || {
        let mut replica = Replica::new(Epoch::GENESIS, Document::new());
        let mut state = receiver_state.clone();
        let message = sync::Message::decode(&message).expect("decode");
        let started = Instant::now();
        let before = sorted(replica.doc.get_heads());
        replica.logged(|doc, log| {
            (
                doc.receive_sync_message_log_patches(&mut state, message, log)
                    .expect("receive"),
                true,
            )
        });
        std::hint::black_box(replica.changed_since(&before, InLists::Index));
        started.elapsed()
    });
    let diffed = median(5, || {
        let mut doc = Document::new();
        let mut state = receiver_state.clone();
        let message = sync::Message::decode(&message).expect("decode");
        let started = Instant::now();
        let before = sorted(doc.get_heads());
        doc.receive_sync_message(&mut state, message)
            .expect("receive");
        let after = sorted(doc.get_heads());
        std::hint::black_box(store::changed(&doc, &before, &after, InLists::Index));
        started.elapsed()
    });
    report(
        &format!("one sync message of {changes} changes"),
        logged,
        diffed,
    );
}

#[test]
#[ignore = "a benchmark: cargo benchmarks"]
fn bench_change_publishing() {
    println!(""); // put reports on new line
    local_writes(1_000, 1_000);
    local_writes(100_000, 1_000);
    sync_message(1_000);
    sync_message(10_000);
}