holochain 0.7.0-dev.31

use crate::conductor::integration_dump;
use crate::sweettest::DurationOrSeconds;
use crate::sweettest::SweetCell;
use crate::test_utils::consistency::request_published_ops;
use holo_hash::*;
use holochain_state::dht_store::DhtStoreRead;
use holochain_types::prelude::*;
use std::collections::HashMap;
use std::collections::HashSet;
use std::fmt::Write;
use std::time::Duration;

/// Consistency was failed to be reached. Here's a report.
#[derive(derive_more::From)]
pub struct ConsistencyError(String);

impl std::fmt::Debug for ConsistencyError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "{}", self.0)
    }
}

/// Alias
pub type ConsistencyResult = Result<(), ConsistencyError>;

async fn delay(elapsed: Duration) {
    let delay = if elapsed > Duration::from_secs(10) {
        CONSISTENCY_DELAY_HIGH
    } else if elapsed > Duration::from_secs(1) {
        CONSISTENCY_DELAY_MID
    } else {
        CONSISTENCY_DELAY_LOW
    };
    tokio::time::sleep(delay).await
}

/// Extra conditions that must be satisfied for consistency to be reached.
///
/// Without supplying extra conditions, it's expected that at the time of beginning
/// the consistency awaiting, all ops which will be published have already been published.
/// However, in cases where more publishing is expected, such as when warrants will be authored
/// due to recently publishing invalid ops, these conditions can be used to make sure that
/// the consistency check will not proceed until all publishing expectations have occurred.
#[derive(Debug, Default, Clone)]
pub struct ConsistencyConditions {
    /// This many warrants must have been published against the keyed agent.
    warrants_issued: HashMap<AgentPubKey, usize>,
}

impl From<()> for ConsistencyConditions {
    fn from(_: ()) -> Self {
        Self::default()
    }
}

impl From<Vec<(AgentPubKey, usize)>> for ConsistencyConditions {
    fn from(items: Vec<(AgentPubKey, usize)>) -> Self {
        Self {
            warrants_issued: items.iter().cloned().collect(),
        }
    }
}

impl ConsistencyConditions {
    fn check<'a>(
        &self,
        published_ops: impl Iterator<Item = &'a DhtOp>,
    ) -> Result<bool, ConsistencyError> {
        let mut checked = self.warrants_issued.clone();
        for v in checked.values_mut() {
            *v = 0;
        }

        for op in published_ops {
            if let DhtOp::WarrantOp(op) = op {
                let author = &op.action_author();
                if let Some(count) = checked.get_mut(author) {
                    *count += 1;
                    if *count > *self.warrants_issued.get(author).unwrap() {
                        return Err(format!(
                            "Expected exactly {} warrants to be published against agent {author}, but found more",
                            self.warrants_issued.get(author).unwrap(),
                        )
                    .into());
                    }
                }
            }
        }

        Ok(checked == self.warrants_issued)
    }

    /// Return the total number of warrants expected to be published
    pub fn num_warrants(&self) -> usize {
        self.warrants_issued.values().sum()
    }
}

/// Wait for all cells to reach consistency,
/// with the option to specify that some cells are offline.
///
/// Cells paired with a `false` value will have their authored ops counted towards the total,
/// but not their integrated ops (since they are not online to integrate things).
/// This is useful for tests where nodes go offline.
#[cfg_attr(feature = "instrument", tracing::instrument(skip_all))]
pub async fn await_conditional_consistency<'a, I: IntoIterator<Item = (&'a SweetCell, bool)>>(
    timeout: impl Into<DurationOrSeconds>,
    conditions: impl Into<ConsistencyConditions>,
    all_cells: I,
) -> ConsistencyResult {
    #[allow(clippy::type_complexity)]
    let all_cell_dbs: Vec<(AgentPubKey, DhtStoreRead, Option<DhtStoreRead>)> = all_cells
        .into_iter()
        .map(|(c, online)| {
            (
                c.agent_pubkey().clone(),
                c.dht_store().as_read(),
                online.then(|| c.dht_store().as_read()),
            )
        })
        .collect();
    let all_cell_dbs: Vec<_> = all_cell_dbs
        .iter()
        .map(|c| (&c.0, &c.1, c.2.as_ref()))
        .collect();
    wait_for_integration_diff_conditional(
        &all_cell_dbs[..],
        timeout.into().into_duration(),
        conditions.into(),
    )
    .await
}

/// Wait for all cell envs to reach consistency, meaning that every op
/// published by every cell has been integrated by every node.
///
/// Each cell is `(author, authored_store, online_dht_store)`. The published
/// set is read from each cell's own DHT store (self-authored ops), and the
/// integrated set from each online cell's DHT store. Cells with `None` for the
/// DHT store are treated as offline: their published ops still count towards
/// the total, but their integration is not checked.
pub async fn wait_for_integration_diff_conditional(
    cells: &[(&AgentPubKey, &DhtStoreRead, Option<&DhtStoreRead>)],
    timeout: Duration,
    conditions: ConsistencyConditions,
) -> ConsistencyResult {
    let start = tokio::time::Instant::now();
    let mut done = HashSet::new();
    let mut integrated = vec![HashMap::<DhtOpHash, DhtOp>::new(); cells.len()];
    let mut published = HashMap::<DhtOpHash, DhtOp>::new();
    let mut publish_complete = false;

    while start.elapsed() < timeout {
        if !publish_complete {
            published = HashMap::new();
            for (author, authored_store, _) in cells.iter() {
                let p = request_published_ops(authored_store, author)
                    .await
                    .unwrap()
                    .into_iter()
                    .map(|(_, _, op)| {
                        let hashed = DhtOpHashed::from_content_sync(op);
                        (hashed.hash, hashed.content)
                    });

                // Assert that there are no duplicates
                let expected = p.len() + published.len();
                published.extend(p);
                assert_eq!(published.len(), expected);
            }
        }

        let prev_publish_complete = publish_complete;
        publish_complete = conditions.check(published.values())?;

        if publish_complete {
            if !prev_publish_complete {
                tracing::info!("*** All expected ops were published ***");
            }
            // Compare the published ops to the integrated ops for each node
            for (i, (_, _, dht_db)) in cells.iter().enumerate() {
                if done.contains(&i) {
                    continue;
                }
                if let Some(db) = dht_db.as_ref() {
                    integrated[i] = get_integrated_ops(db)
                        .await
                        .into_iter()
                        .map(|op| {
                            let hashed = DhtOpHashed::from_content_sync(op);
                            (hashed.hash, hashed.content)
                        })
                        .collect();

                    if integrated[i] == published {
                        done.insert(i);
                        tracing::debug!(i, "Node reached consistency");
                    } else {
                        let total_time_waited = start.elapsed();
                        let queries = integration_dump(db).await.unwrap();
                        let num_integrated = integrated.len();
                        tracing::debug!(i, ?num_integrated, ?total_time_waited, counts = ?queries, "consistency-status");
                    }
                } else {
                    // If the DHT db is not provided, don't check integration
                    done.insert(i);
                }
            }
        }

        // If all nodes reached consistency, exit successfully
        if done.len() == cells.len() {
            return Ok(());
        }

        let total_time_waited = start.elapsed();
        delay(total_time_waited).await;
    }

    let header = format!(
        "{:53} {:>3} {:53} {:53} {}\n{}",
        "author",
        "seq",
        "op_hash",
        "action_hash",
        "op_type (action_type)",
        "-".repeat(53 + 3 + 53 + 53 + 4 + 21)
    );

    if !publish_complete {
        let published = published
            .values()
            .map(display_op)
            .collect::<Vec<_>>()
            .join("\n");
        return Err(format!("There are still ops which were expected to have been published which weren't:\n{header}\n{published}").into());
    }

    let mut report = String::new();
    let not_consistent = (0..cells.len())
        .filter(|i| !done.contains(i))
        .collect::<Vec<_>>();

    writeln!(
        report,
        "{} cells did not reach consistency: {:?}",
        not_consistent.len(),
        not_consistent
    )
    .unwrap();

    if not_consistent.is_empty() {
        unreachable!("At least one node must not have reached consistency");
    }

    for c in &not_consistent {
        let integrated = integrated[*c].clone();

        eprintln!("Agent {} is not consistent", cells[*c].0);

        let (unintegrated, unpublished) = diff_ops(published.values(), integrated.values());
        let diff = diff_report(unintegrated, unpublished);

        #[allow(clippy::comparison_chain)]
        if integrated.len() > published.len() {
            eprintln!(
                "{report}\nnum integrated ops ({}) > num published ops ({}), meaning you may not be accounting for all nodes in this test. Consistency may not be complete. Report:\n\n{header}\n{diff}",
                integrated.len(),
                published.len()
            );
        } else if integrated.len() < published.len() {
            let db = cells[*c].2.as_ref().expect("DhtStore must be provided");
            let integration_dump = integration_dump(db).await.unwrap();

            eprintln!(
                "{}\nConsistency not achieved after {:?}. Expected {} ops, but only {} integrated. Report:\n\n{}\n{}\n\n{:?}",
                report,
                timeout,
                published.len(),
                integrated.len(),
                header,
                diff,
                integration_dump
            );
        } else {
            unreachable!()
        }
    }

    Err(ConsistencyError(format!(
        "{} agents were inconsistent",
        not_consistent.len()
    )))
}

const CONSISTENCY_DELAY_LOW: Duration = Duration::from_millis(100);
const CONSISTENCY_DELAY_MID: Duration = Duration::from_millis(500);
const CONSISTENCY_DELAY_HIGH: Duration = Duration::from_millis(1000);

fn diff_ops<'a>(
    published: impl Iterator<Item = &'a DhtOp>,
    integrated: impl Iterator<Item = &'a DhtOp>,
) -> (Vec<String>, Vec<String>) {
    let mut published: Vec<_> = published.map(display_op).collect();
    let mut integrated: Vec<_> = integrated.map(display_op).collect();
    published.sort();
    integrated.sort();

    let mut unintegrated = vec![];
    let mut unpublished = vec![];

    for d in diff::slice(&published, &integrated) {
        match d {
            diff::Result::Left(l) => unintegrated.push(l.to_owned()),
            diff::Result::Right(r) => unpublished.push(r.to_owned()),
            _ => (),
        }
    }

    (unintegrated, unpublished)
}

fn diff_report(unintegrated: Vec<String>, unpublished: Vec<String>) -> String {
    let unintegrated = if unintegrated.is_empty() {
        "".to_string()
    } else {
        format!("Unintegrated:\n\n{}\n", unintegrated.join("\n"))
    };
    let unpublished = if unpublished.is_empty() {
        "".to_string()
    } else {
        format!("Unpublished:\n\n{}\n", unpublished.join("\n"))
    };

    format!("{unintegrated}{unpublished}")
}

fn display_op(op: &DhtOp) -> String {
    match op {
        DhtOp::ChainOp(op) => format!(
            "{} {:>3} {} {} {} ({})",
            op.action().author(),
            op.action().action_seq(),
            op.to_hash(),
            op.action().to_hash(),
            op.get_type(),
            op.action().action_type(),
        ),
        DhtOp::WarrantOp(op) => {
            format!("{} WARRANT ({})", op.author, op.get_type(),)
        }
    }
}

/// Get all [`DhtOps`](holochain_types::prelude::DhtOp) integrated by this node,
/// read from the new DHT store.
///
/// "Integrated" follows the new store semantics: locally-validated chain ops
/// (GET-cached copies excluded, rejected ops included). **Warrants are
/// deliberately excluded** — the legacy check inner-joined `Action` (warrants
/// have none) and so compared chain ops only; warrants need not reach every
/// node. Ops are reconstructed into legacy `DhtOp`s so their hashes match the
/// published set.
async fn get_integrated_ops(dht_store: &DhtStoreRead) -> Vec<DhtOp> {
    let chain = dht_store
        .integrated_chain_ops_for_dump(None)
        .await
        .unwrap()
        .into_iter()
        .map(|row| row.wire)
        .collect();
    crate::conductor::wire_rows_to_legacy_ops(chain, Vec::new())
}