legato 0.0.8

use crate::{
    LegatoApp, LegatoFrontend, LegatoMsg,
    config::Config,
    dsl::{
        ir::{DSLParams, NodeId, Port, Value},
        parse::legato_parser,
        pipeline::Pipeline,
    },
    graph::{Connection, ConnectionEntry},
    midi::{MidiRuntimeFrontend, MidiStore},
    node::LegatoNode,
    nodes::audio::{
        delay::DelayLine,
        mixer::{MonoFanOut, TrackMixer},
    },
    params::{ParamKey, ParamMeta},
    pipes::{Pipe, PipeRegistry},
    ports::Ports,
    registry::{
        NodeRegistry, audio_registry_factory, control_registry_factory, midi_registry_factory,
    },
    resources::{DelayLineKey, ResourceBuilder, SampleKey},
    runtime::{NodeKey, Runtime, RuntimeFrontend, build_runtime},
    sample::{AudioSampleFrontend, AudioSampleHandle},
    spec::NodeSpec,
};
use arc_swap::ArcSwapOption;
use std::{
    collections::HashMap,
    marker::PhantomData,
    sync::{Arc, atomic::AtomicU64},
};

/// ValidationError covers logical issues
/// when lowering from the AST to the IR.
///
/// These might be bad parameters,
/// bad values, nodes that don't exist, etc.
#[derive(Clone, PartialEq, Debug)]
pub enum ValidationError {
    ParseError(String),
    NodeNotFound(String),
    NamespaceNotFound(String),
    InvalidParameter(String),
    MissingRequiredParameters(String),
    MissingRequiredParameter(String),
    ResourceNotFound(String),
    PipeNotFound(String),
}

// Typestates for the builder
pub struct Unconfigured;
pub struct Configured;
pub struct ContainsNodes;
pub struct Connected;
pub struct ReadyToBuild;

// Different traits for varying levels
pub trait CanRegister {}
pub trait CanAddNode {}
pub trait CanConnect {}
pub trait CanApplyPipe {}
pub trait CanSetSink {}
pub trait CanBuild {}

pub trait CanAddMidiRuntime {}

// Setting up "permissions" for different structs. May be too complicated but also easy to add more states with overlapping permissiosn

impl CanRegister for Unconfigured {}
impl CanRegister for Configured {}
impl CanRegister for ContainsNodes {}

impl CanAddMidiRuntime for Configured {}
impl CanAddMidiRuntime for ContainsNodes {}
impl CanAddMidiRuntime for Connected {}
impl CanAddMidiRuntime for ReadyToBuild {}

impl CanAddNode for Configured {}
impl CanAddNode for ContainsNodes {}

impl CanApplyPipe for ContainsNodes {}

impl CanConnect for ContainsNodes {}
impl CanConnect for Connected {}

impl CanSetSink for ContainsNodes {}
impl CanSetSink for Connected {}

impl CanBuild for ReadyToBuild {}

pub struct DslBuilding;

impl CanRegister for DslBuilding {}
impl CanAddNode for DslBuilding {}
impl CanConnect for DslBuilding {}
impl CanApplyPipe for DslBuilding {}
impl CanSetSink for DslBuilding {}
impl CanBuild for DslBuilding {}

// Convenience struct for moving from one state to another
impl<S> LegatoBuilder<S> {
    #[inline]
    fn into_state<T>(self) -> LegatoBuilder<T> {
        LegatoBuilder {
            runtime: self.runtime,
            namespaces: self.namespaces,
            working_name_lookup: self.working_name_lookup,
            delay_name_to_key: self.delay_name_to_key,
            resource_builder: self.resource_builder,
            sample_frontends: self.sample_frontends,
            sample_name_to_key: self.sample_name_to_key,
            pipe_lookup: self.pipe_lookup,
            last_selection: self.last_selection,
            midi_runtime_frontend: self.midi_runtime_frontend,
            _state: PhantomData,
        }
    }
}

pub struct LegatoBuilder<State> {
    runtime: Runtime,
    // String to registries of node spec (including factory fn) lookup
    namespaces: HashMap<String, NodeRegistry>,
    // Lookup from string to NodeKey
    working_name_lookup: HashMap<String, NodeKey>,
    // Lookup from string to Pipe Fn
    pipe_lookup: PipeRegistry,
    // Resources being built. These can be pased to node factories
    resource_builder: ResourceBuilder,
    // Name to key maps
    sample_name_to_key: HashMap<String, SampleKey>,
    delay_name_to_key: HashMap<String, DelayLineKey>,
    sample_frontends: HashMap<String, AudioSampleFrontend>,
    // When adding a node or piping, this tracks and sets the node key for pipes
    last_selection: Option<SelectionKind>,
    // The midi runtime that can be added to the runtime
    midi_runtime_frontend: Option<MidiRuntimeFrontend>,
    _state: PhantomData<State>,
}

impl LegatoBuilder<Unconfigured> {
    pub fn new(config: Config, ports: Ports) -> LegatoBuilder<Configured> {
        let mut namespaces = HashMap::new();
        let audio_registry = audio_registry_factory();
        let control_registry = control_registry_factory();
        let midi_registry = midi_registry_factory();

        namespaces.insert("audio".into(), audio_registry);
        namespaces.insert("control".into(), control_registry);
        namespaces.insert("midi".into(), midi_registry);

        namespaces.insert("user".into(), NodeRegistry::new());

        let runtime = build_runtime(config, ports);

        LegatoBuilder::<Configured> {
            runtime,
            resource_builder: ResourceBuilder::default(),
            sample_name_to_key: HashMap::new(),
            delay_name_to_key: HashMap::new(),
            sample_frontends: HashMap::new(),
            namespaces,
            working_name_lookup: HashMap::new(),
            pipe_lookup: PipeRegistry::default(),
            last_selection: None,
            midi_runtime_frontend: None,
            _state: std::marker::PhantomData,
        }
    }
}

impl LegatoBuilder<Configured> {
    pub fn build_dsl(self, graph: &str) -> (LegatoApp, LegatoFrontend) {
        let can_build = self.into_state::<DslBuilding>();
        can_build._build_dsl(graph)
    }
}

impl<S> LegatoBuilder<S>
where
    S: CanRegister,
{
    /// Add a new registry. Think of registries like "DLC" or packs of nodes that users or developers can extend
    pub fn add_node_registry(mut self, name: &'static str, registry: NodeRegistry) -> Self {
        self.namespaces.insert(name.into(), registry);
        self
    }
    /// Register a node to the "user" namespace
    pub fn register_node(mut self, namespace: &'static str, spec: NodeSpec) -> Self {
        match self.namespaces.get_mut(namespace) {
            Some(ns) => ns.declare_node(spec),
            None => panic!("Cannot find namespace {}", namespace),
        }
        self
    }
    /// Register a custom pipe for transforming nodes
    pub fn register_pipe(mut self, name: &'static str, pipe: Box<dyn Pipe>) -> Self {
        self.pipe_lookup.insert(name.into(), pipe);
        self
    }
}

impl<S> LegatoBuilder<S>
where
    S: CanAddMidiRuntime,
{
    pub fn set_midi_runtime(mut self, rt: MidiRuntimeFrontend) -> Self {
        self.midi_runtime_frontend = Some(rt);
        self
    }
}

impl<S> LegatoBuilder<S>
where
    S: CanAddNode,
{
    /// This pattern is used because we sometimes execute this in a non-owned context
    fn _add_node_ref_self(
        &mut self,
        namespace: &String,
        node_kind: &String,
        alias: &String,
        params: &DSLParams,
    ) {
        let ns = self
            .namespaces
            .get(namespace)
            .unwrap_or_else(|| panic!("Could not find namespace {}", namespace));

        let mut resource_builder_view = ResourceBuilderView {
            config: &self.runtime.get_config(),
            resource_builder: &mut self.resource_builder,
            sample_keys: &mut self.sample_name_to_key,
            delay_keys: &mut self.delay_name_to_key,
            sample_frontends: &mut self.sample_frontends,
        };

        let node = ns
            .get_node(&mut resource_builder_view, node_kind, params)
            .unwrap_or_else(|_| panic!("Could not find node {}", node_kind));

        let legato_node = LegatoNode::new(alias.into(), node_kind.into(), node);

        let key = self.runtime.add_node(legato_node);

        self.working_name_lookup.insert(alias.clone(), key);

        // Set the last node_ref_added
        self.last_selection = Some(SelectionKind::Single(key));
    }
    pub fn add_node(
        mut self,
        namespace: &String,
        node_kind: &String,
        alias: &String,
        params: &DSLParams,
    ) -> LegatoBuilder<ContainsNodes> {
        self._add_node_ref_self(namespace, node_kind, alias, params);
        self.into_state()
    }

    /// Skip the ceremony with namespaces, specs, etc. and just add a LegatoNode. This still requires an alias for connections and debugging
    pub fn add_node_raw(mut self, node: LegatoNode, alias: &str) -> LegatoBuilder<ContainsNodes> {
        let key = self.runtime.add_node(node);

        self.last_selection = Some(SelectionKind::Single(key));

        self.working_name_lookup.insert(alias.into(), key);

        self.into_state()
    }
}

impl<S> LegatoBuilder<S>
where
    S: CanConnect,
{
    /// This pattern is used because we sometimes execute this in a non-owned context
    fn _connect_ref_self(&mut self, connection: AddConnectionProps) {
        let source_indicies: Vec<usize> = match connection.source_kind {
            Port::None => {
                let ports = self.runtime.get_node_ports(&connection.source);
                ports.audio_out.iter().enumerate().map(|(i, _)| i).collect()
            }
            Port::Index(port) => vec![port],
            Port::Named(ref port) => {
                let ports = self.runtime.get_node_ports(&connection.source);
                let index = ports
                    .audio_out
                    .iter()
                    .find(|x| x.name == port)
                    .unwrap_or_else(|| panic!("Could not find index for named port {}", port))
                    .index;

                vec![index]
            }
            Port::Slice(start, end) => {
                if end < start {
                    panic!("End slice cannot be less than start!");
                }

                (start..end).collect::<Vec<_>>()
            }
            Port::Stride { start, end, stride } => {
                if end < start {
                    panic!("End slice cannot be less than start!");
                }

                (start..end).step_by(stride).collect()
            }
        };

        let sink_indicies: Vec<usize> = match connection.sink_kind {
            Port::None => {
                let ports = self.runtime.get_node_ports(&connection.sink);
                ports.audio_in.iter().enumerate().map(|(i, _)| i).collect()
            }
            Port::Index(port) => vec![port],
            Port::Named(ref port) => {
                let ports = self.runtime.get_node_ports(&connection.sink);
                let index = ports
                    .audio_in
                    .iter()
                    .find(|x| x.name == port)
                    .unwrap_or_else(|| panic!("Could not find index for named port {}", port))
                    .index;

                vec![index]
            }
            Port::Slice(start, end) => {
                if end < start {
                    panic!("End slice cannot be less than start!");
                }

                (start..end).collect::<Vec<_>>()
            }
            Port::Stride { start, end, stride } => {
                if end < start {
                    panic!("End slice cannot be less than start!");
                }

                (start..end).step_by(stride).collect()
            }
        };

        let source_arity = source_indicies.len();
        let sink_arity = sink_indicies.len();

        match (source_arity, sink_arity) {
            (1, 1) => one_to_one(
                &mut self.runtime,
                connection,
                source_indicies[0],
                sink_indicies[0],
            ),
            (1, n) if n >= 1 => one_to_n(
                &mut self.runtime,
                connection,
                source_indicies[0],
                sink_indicies.as_slice(),
            ),
            (n, 1) if n >= 1 => n_to_one(
                &mut self.runtime,
                connection,
                source_indicies.as_slice(),
                sink_indicies[0],
            ),
            (n, m) if n == m => n_to_n(
                &mut self.runtime,
                connection,
                &source_indicies,
                &sink_indicies,
            ),
            (n, m) => unimplemented!("Cannot match request arity {}:{}", n, m),
        }
    }
    pub fn connect(mut self, connection: AddConnectionProps) -> LegatoBuilder<Connected> {
        self._connect_ref_self(connection);
        self.into_state()
    }
}

impl<S> LegatoBuilder<S>
where
    S: CanSetSink,
{
    pub fn set_sink(mut self, key: NodeKey) -> LegatoBuilder<ReadyToBuild> {
        self.runtime.set_sink_key(key).expect("Sink key not found");
        self.into_state()
    }
    pub fn set_source(mut self, key: NodeKey) -> LegatoBuilder<ReadyToBuild> {
        self.runtime.set_sink_key(key).expect("Sink key not found");
        self.into_state()
    }
}

impl<S> LegatoBuilder<S>
where
    S: CanApplyPipe,
{
    pub fn pipe(&mut self, pipe_name: &str, props: Option<Value>) {
        match self.last_selection {
            Some(_) => {
                if let Ok(pipe) = self.pipe_lookup.get(pipe_name) {
                    if let Some(last_selection) = &self.last_selection {
                        let mut view = SelectionView {
                            runtime: &mut self.runtime,
                            working_name_lookup: &mut self.working_name_lookup,
                            selection: last_selection.clone(),
                        };
                        pipe.pipe(&mut view, props);

                        self.last_selection = Some(view.get_selection_owned());
                    } else {
                        panic!(
                            "Cannot apply pipe when there is no last_selection! Please add a node first and apply a pipe directly after."
                        )
                    }
                } else {
                    panic!("Pipe not found {}", pipe_name);
                }
            }
            None => panic!("Cannot apply pipe to non-existent node!"),
        }
    }
}

impl<S> LegatoBuilder<S>
where
    S: CanBuild,
{
    pub fn build(self) -> (LegatoApp, LegatoFrontend) {
        let mut runtime = self.runtime;

        let (resources, param_store_frontend) = self.resource_builder.build();

        runtime.set_resources(resources);

        // Allocate all of the audio buffers needed at runtime
        runtime.prepare();

        if self.midi_runtime_frontend.is_some() {
            let ctx = runtime.get_context_mut();
            ctx.set_midi_store(MidiStore::new(256));
        }

        // TODO: Perhaps a different crate here instead of leaking
        let queue = Box::leak(Box::new(heapless::spsc::Queue::<LegatoMsg, 512>::new()));

        let (producer, consumer) = queue.split();

        let mut app = LegatoApp::new(runtime, consumer);

        if let Some(midi_rt) = self.midi_runtime_frontend {
            app.set_midi_runtime(midi_rt);
        }

        let rt_frontend = RuntimeFrontend::new(self.sample_frontends);

        let frontend = LegatoFrontend::new(rt_frontend, param_store_frontend, producer);

        (app, frontend)
    }
}

impl LegatoBuilder<DslBuilding> {
    fn _build_dsl(mut self, content: &str) -> (LegatoApp, LegatoFrontend) {
        let ast = legato_parser(content).unwrap();

        let ir = Pipeline::default().run_from_ast(ast);

        println!("{}", &ir);

        // Sanity check: every node must be a leaf before the builder runs.
        debug_assert!(
            !ir.has_unresolved_macros(),
            "_build_dsl: unresolved MacroRef nodes remain after pipeline"
        );

        // Map each IRNode (by NodeId) to a runtime NodeKey as we add nodes.
        let mut ir_to_runtime: HashMap<NodeId, NodeKey> = HashMap::new();

        for node_id in ir.topological_sort() {
            let node = ir.get_node(node_id).unwrap();
            let dsl_params = DSLParams::new(&node.params);

            // _add_node_ref_self populates working_name_lookup (needed by
            // pipes) and sets last_selection.
            self._add_node_ref_self(&node.namespace, &node.node_type, &node.alias, &dsl_params);

            let runtime_key = *self
                .working_name_lookup
                .get(&node.alias)
                .expect("alias must be in lookup immediately after _add_node_ref_self");

            ir_to_runtime.insert(node_id, runtime_key);

            // Pipes are applied right after the node they belong to, matching
            // the original builder contract.
            for pipe in &node.pipes {
                self.pipe(&pipe.name, pipe.params.clone());
            }
        }

        // Wire edges using the NodeId -> NodeKey map (no string lookups).
        for edge in ir.edges() {
            self._connect_ref_self(AddConnectionProps {
                source: ir_to_runtime[&edge.source],
                source_kind: edge.source_port.clone(),
                sink: ir_to_runtime[&edge.sink],
                sink_kind: edge.sink_port.clone(),
            });
        }

        // Resolve sink / source.
        let sink_id = ir
            .sink
            .expect("IRGraph has no sink — check the DSL for a `sink:` declaration");
        self.runtime
            .set_sink_key(ir_to_runtime[&sink_id])
            .expect("Could not set sink");

        if let Some(source_id) = ir.source {
            self.runtime
                .set_source_key(ir_to_runtime[&source_id])
                .expect("Could not set runtime source");
        }

        self.build()
    }
}

#[derive(Clone, Debug)]
pub enum NodeViewKind {
    Single(LegatoNode),
    Multiple(Vec<LegatoNode>),
}

#[derive(Clone, PartialEq, Debug)]
pub enum SelectionKind {
    Single(NodeKey),
    Multiple(Vec<NodeKey>),
}

/// Selections are passed between pipes, and set after inserting nodes.
///
/// They expose a small view of operations on the runtime, so that pipes can
/// transform nodes BEFORE connections are formed. This is enforced via the
/// type state pattern.
#[derive(Debug)]
pub struct SelectionView<'a> {
    runtime: &'a mut Runtime,
    working_name_lookup: &'a mut HashMap<String, NodeKey>,
    selection: SelectionKind,
}

impl<'a> SelectionView<'a> {
    pub fn new(
        runtime: &'a mut Runtime,
        working_name_lookup: &'a mut HashMap<String, NodeKey>,
        selection: SelectionKind,
    ) -> Self {
        Self {
            runtime,
            working_name_lookup,
            selection,
        }
    }

    pub fn insert(&mut self, node: LegatoNode) {
        let working_name = node.name.clone();
        let key = self.runtime.add_node(node);

        // Update the lookup map
        self.working_name_lookup.insert(working_name, key);

        // After inserting a node, we add push to the selection. If we only had a single node, we now have two
        match &mut self.selection {
            SelectionKind::Single(old_node) => {
                self.selection = SelectionKind::Multiple(vec![*old_node, key])
            }
            SelectionKind::Multiple(nodes) => nodes.push(key),
        }
    }

    pub fn selection(&self) -> &SelectionKind {
        &self.selection
    }

    pub fn config(&self) -> Config {
        self.runtime.get_config()
    }

    pub fn replace(&mut self, key: NodeKey, node: LegatoNode) {
        let working_name = node.name.clone();

        self.runtime.replace_node(key, node);

        // Find the working name and replace the name with the new node name, but point to the same key.
        if let Some((old_key, _)) = self.working_name_lookup.iter().find(|(_, nk)| **nk == key) {
            self.working_name_lookup.remove(&old_key.clone());
            self.working_name_lookup.insert(working_name, key);
        }
    }

    pub fn get_node_mut(&mut self, key: &NodeKey) -> Option<&mut LegatoNode> {
        self.runtime.get_node_mut(key)
    }

    pub fn get_node(&self, key: &NodeKey) -> Option<&LegatoNode> {
        self.runtime.get_node(key)
    }

    pub fn get_key(&mut self, name: &'static str) -> Option<NodeKey> {
        self.working_name_lookup.get(name).copied()
    }

    pub fn delete(&mut self, key: NodeKey) {
        self.runtime.remove_node(key);

        // Remove the key from the working name lookup
        if let Some((old_key, _)) = self.working_name_lookup.iter().find(|(_, nk)| **nk == key) {
            self.working_name_lookup.remove(&old_key.clone());
        }
    }

    pub fn clone_node(&mut self, key: NodeKey) -> Option<LegatoNode> {
        self.runtime.get_node(&key).cloned()
    }

    pub fn get_selection_owned(self) -> SelectionKind {
        self.selection
    }
}

/// A small slice of the runtime exposed for nodes in their node factories.
///
/// This is useful to say reserve delay lines, or other shared logic.
///
/// TODO: Unify the interface between sample, delay, and param_store?
pub struct ResourceBuilderView<'a> {
    pub config: &'a Config,
    pub resource_builder: &'a mut ResourceBuilder,
    pub sample_keys: &'a mut HashMap<String, SampleKey>,
    pub delay_keys: &'a mut HashMap<String, DelayLineKey>,
    pub sample_frontends: &'a mut HashMap<String, AudioSampleFrontend>,
}

impl<'a> ResourceBuilderView<'a> {
    pub fn add_delay_line(&mut self, name: &str, delay_line: DelayLine) -> DelayLineKey {
        let key = self.resource_builder.add_delay_line(delay_line);
        self.delay_keys.insert(name.to_string(), key);

        key
    }

    pub fn replace_delay_line(&mut self, key: DelayLineKey, delay_line: DelayLine) {
        self.resource_builder.replace_delay_line(key, delay_line);
    }

    pub fn get_delay_line_key(&self, name: &String) -> Option<DelayLineKey> {
        self.delay_keys.get(name).cloned()
    }

    pub fn add_sampler(&mut self, name: &String) -> SampleKey {
        if let Some(&key) = self.sample_keys.get(name) {
            key
        } else {
            let data = ArcSwapOption::new(None);

            let handle = Arc::new(AudioSampleHandle {
                sample: data,
                sample_version: AtomicU64::new(0),
            });

            let frontend = AudioSampleFrontend::new(handle.clone());

            self.sample_frontends.insert(name.clone(), frontend);

            self.resource_builder.add_sample_resource(handle)
        }
    }

    pub fn add_param(&mut self, unique_name: String, meta: ParamMeta) -> ParamKey {
        self.resource_builder.add_param(unique_name, meta)
    }

    pub fn get_sampler_key(&self, name: &String) -> Result<SampleKey, ValidationError> {
        self.sample_keys.get(name).cloned().ok_or_else(|| {
            ValidationError::ResourceNotFound(format!("Could not find sample key {}", name))
        })
    }

    pub fn get_config(&self) -> &Config {
        self.config
    }
}

#[derive(Debug, PartialEq, Clone)]
pub struct AddConnectionProps {
    pub source: NodeKey,
    pub source_kind: Port,
    pub sink: NodeKey,
    pub sink_kind: Port,
}

pub enum AddConnectionKind {
    Index(usize),
    Named(&'static str),
    Auto,
}

// Utility functions for handling only audio connections for now

fn one_to_one(
    runtime: &mut Runtime,
    props: AddConnectionProps,
    source_index: usize,
    sink_index: usize,
) {
    runtime
        .add_edge(Connection {
            source: ConnectionEntry {
                node_key: props.source,
                port_index: source_index,
            },
            sink: ConnectionEntry {
                node_key: props.sink,
                port_index: sink_index,
            },
        })
        .expect("Could not add edge");
}

fn one_to_n(
    runtime: &mut Runtime,
    props: AddConnectionProps,
    source_index: usize,
    sink_indicies: &[usize],
) {
    let n = sink_indicies.len();

    // Fanout mixer going from 1 -> n
    let mixer = runtime.add_node(LegatoNode::new(
        format!("MonoFanOut{:?}{:?}", props.source, props.sink),
        "MonoFanOut".into(),
        Box::new(MonoFanOut::new(n)),
    ));

    // Wire mono to mixer
    runtime
        .add_edge(Connection {
            source: ConnectionEntry {
                node_key: props.source,
                port_index: source_index,
            },
            sink: ConnectionEntry {
                node_key: mixer,
                port_index: 0,
            },
        })
        .expect("Could not add edge");

    // Wire fanout connection to each sink. We add this node in order to change the gain when fanning out

    for sink_index in sink_indicies.iter() {
        runtime
            .add_edge(Connection {
                source: ConnectionEntry {
                    node_key: mixer,
                    port_index: 0,
                },
                sink: ConnectionEntry {
                    node_key: props.sink,
                    port_index: *sink_index,
                },
            })
            .expect("Could not add edge");
    }
}

fn n_to_one(
    runtime: &mut Runtime,
    props: AddConnectionProps,
    source_indicies: &[usize],
    sink_index: usize,
) {
    let n = source_indicies.len();

    // Make mixer with n mono tracks
    let mixer = runtime.add_node(LegatoNode::new(
        format!("TrackMixer{:?}{:?}", props.source, props.sink),
        "TrackMixer".into(),
        Box::new(TrackMixer::new(1, n, vec![1.0 / f32::sqrt(n as f32); n])),
    ));

    // Build connections into track mixer
    for (i, source_index) in source_indicies.iter().enumerate() {
        runtime
            .add_edge(Connection {
                source: ConnectionEntry {
                    node_key: props.source,
                    port_index: *source_index,
                },
                sink: ConnectionEntry {
                    node_key: mixer,
                    port_index: i,
                },
            })
            .expect("Could not add edge");
    }

    // Wire track mixer to sink index
    runtime
        .add_edge(Connection {
            source: ConnectionEntry {
                node_key: mixer,
                port_index: 0,
            },
            sink: ConnectionEntry {
                node_key: props.sink,
                port_index: sink_index,
            },
        })
        .expect("Could not add edge");
}

fn n_to_n(
    runtime: &mut Runtime,
    props: AddConnectionProps,
    source_indicies: &[usize],
    sink_indicies: &[usize],
) {
    assert!(source_indicies.len() == sink_indicies.len());
    source_indicies
        .iter()
        .zip(sink_indicies)
        .for_each(|(source, sink)| {
            runtime
                .add_edge(Connection {
                    source: ConnectionEntry {
                        node_key: props.source,
                        port_index: *source,
                    },
                    sink: ConnectionEntry {
                        node_key: props.sink,
                        port_index: *sink,
                    },
                })
                .expect("Could not add edge");
        });
}