hyperhive/swarm-notify-router/src/dry_run.rs

//! Running a flows document over past events without delivering anything.
//!
//! Each event goes through the walk the router delivers with ([`flow::reached`]
//! and [`route::route`]), counting the nodes it reaches. The per-issue
//! snapshots start empty and carry from one event to the next, as the router's
//! do, so an assign event's `assignees_added` is the difference to the
//! previous replayed one. Every recipient counts as an agent.

use std::collections::BTreeMap;

use swarm_wire::notify_flows::Flow;
use swarm_wire::router_api::{DryRun, FlowHits, NodeHits};

use crate::event::{Delivery, Snapshot, fields};
use crate::{flow, route};

/// `flows` over `events`, oldest first, each a forge event type and body.
/// An event that names no issue or PR is not counted.
pub fn run(flows: &[Flow], events: &[(String, Vec<u8>)]) -> DryRun {
    let mut hits: Vec<BTreeMap<&str, u64>> = flows
        .iter()
        .map(|f| f.nodes.iter().map(|n| (n.id.as_str(), 0)).collect())
        .collect();
    let mut snapshots: BTreeMap<String, Snapshot> = BTreeMap::new();
    let mut run = 0;
    for (event_type, body) in events {
        let Some(delivery) = Delivery::parse(body) else {
            continue;
        };
        run += 1;
        let key = delivery.state_key();
        let before = snapshots.remove(&key).unwrap_or_default();
        let event_fields = fields(event_type, &delivery, &before);
        for (flow, counts) in flows.iter().zip(&mut hits) {
            for id in flow::reached(flow, event_type, &event_fields) {
                if let Some(count) = counts.get_mut(id) {
                    *count += 1;
                }
            }
        }
        let routed = route::route(event_type, &delivery, &before, flows, |_| true);
        snapshots.insert(key, routed.next);
    }
    DryRun {
        events: run,
        flows: flows
            .iter()
            .zip(hits)
            .map(|(flow, counts)| FlowHits {
                flow: flow.id.clone(),
                nodes: flow
                    .nodes
                    .iter()
                    .map(|n| NodeHits {
                        node: n.id.clone(),
                        hits: counts.get(n.id.as_str()).copied().unwrap_or(0),
                    })
                    .collect(),
            })
            .collect(),
    }
}

#[cfg(test)]
mod tests {
    use swarm_wire::notify_flows::{Document, builtins};

    use super::*;

    fn body(action: &str, number: i64, assignees: &[&str]) -> Vec<u8> {
        let people: Vec<_> = assignees
            .iter()
            .map(|a| serde_json::json!({ "login": a }))
            .collect();
        serde_json::to_vec(&serde_json::json!({
            "action": action,
            "repository": { "full_name": "o/r", "id": 1 },
            "issue": { "number": number, "title": "t", "html_url": "u", "assignees": people },
            "sender": { "login": "mara" },
        }))
        .expect("encodes")
    }

    fn count(r: &DryRun, flow: &str, node: &str) -> u64 {
        r.flows
            .iter()
            .find(|f| f.flow == flow)
            .and_then(|f| f.nodes.iter().find(|n| n.node == node))
            .unwrap_or_else(|| panic!("no node {flow}/{node}"))
            .hits
    }

    /// The built-ins over an assign, a second assign that adds nobody, an
    /// unassign and a close: each node counts the events that reached it.
    #[test]
    fn each_node_counts_the_events_that_reached_it() {
        let (flows, _) = Document::default().active();
        let events = vec![
            ("issue_assign".to_owned(), body("assigned", 5, &["atlas"])),
            ("issue_assign".to_owned(), body("assigned", 5, &["atlas"])),
            ("issue_assign".to_owned(), body("unassigned", 5, &[])),
            ("issues".to_owned(), body("closed", 5, &[])),
            ("issues".to_owned(), b"not json".to_vec()),
        ];
        let r = run(&flows, &events);
        assert_eq!(r.events, 4);
        // The source sees all three assign events, the match and the sink both
        // assigned ones, though the second adds nobody.
        assert_eq!(count(&r, "assigned", "source"), 3);
        assert_eq!(count(&r, "assigned", "match"), 2);
        assert_eq!(count(&r, "assigned", "sink"), 2);
        // Only the unassign reaches the unassigned flow's sink.
        assert_eq!(count(&r, "unassigned", "source"), 3);
        assert_eq!(count(&r, "unassigned", "sink"), 1);
        // The close is an `issues` event, which these flows start from only
        // in `closed`.
        assert_eq!(count(&r, "closed", "source"), 1);
        assert_eq!(count(&r, "closed", "sink"), 1);
        // No pull request push: the classifier flow counts zero.
        assert_eq!(count(&r, "head-classifier", "source"), 0);
        assert_eq!(count(&r, "head-classifier", "request"), 0);
    }

    /// A condition on a field that depends on the previous event of the
    /// issue (`assignees_added`) holds for the first assign event and not for
    /// an identical second one.
    #[test]
    fn the_issues_snapshot_carries_from_one_replayed_event_to_the_next() {
        use swarm_wire::notify_flows::{Condition, Edge, Node, NodeKind};

        let node = |id: &str, kind| Node {
            id: id.to_owned(),
            kind,
        };
        let edge = |from: &str, to: &str| Edge {
            from: from.to_owned(),
            to: to.to_owned(),
        };
        let flow = Flow {
            id: "added".to_owned(),
            name: "added".to_owned(),
            enabled: true,
            nodes: vec![
                node(
                    "source",
                    NodeKind::Source {
                        events: vec!["issue_assign".to_owned()],
                    },
                ),
                node(
                    "atlas",
                    NodeKind::Match {
                        when: vec![Condition {
                            field: "assignees_added".to_owned(),
                            equals: "atlas".to_owned(),
                        }],
                    },
                ),
            ],
            edges: vec![edge("source", "atlas")],
        };
        let assign = ("issue_assign".to_owned(), body("assigned", 5, &["atlas"]));
        let r = run(&[flow], &[assign.clone(), assign]);
        assert_eq!(count(&r, "added", "source"), 2);
        assert_eq!(count(&r, "added", "atlas"), 1);
    }

    /// Only the flows the document turns on are run.
    #[test]
    fn a_switched_off_built_in_is_not_listed() {
        let doc = Document {
            disabled_builtins: vec!["closed".to_owned()],
            ..Document::default()
        };
        let (flows, _) = doc.active();
        let r = run(&flows, &[]);
        assert_eq!(r.events, 0);
        let listed: Vec<&str> = r.flows.iter().map(|f| f.flow.as_str()).collect();
        assert!(!listed.contains(&"closed"));
        assert!(listed.contains(&"assigned"));
        assert_eq!(listed.len(), builtins().len() - 1);
    }
}