crossflow 0.0.6

/*
 * Copyright (C) 2023 Open Source Robotics Foundation
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 *
*/

use bevy_ecs::prelude::{Commands, Entity};

use std::future::Future;

use smallvec::SmallVec;

use crate::{
    Accessible, Accessing, Accessor, AddOperation, AsMap, Buffer, BufferKeys, BufferLocation,
    BufferMap, BufferSettings, Bufferable, Buffering, Chain, Collect, ForkClone, ForkCloneOutput,
    ForkOptionOutput, ForkResultOutput, ForkTargetStorage, Gate, GateRequest, IncompatibleLayout,
    Injection, InputSlot, IntoAsyncMap, IntoBlockingMap, Joinable, Joined, Node, OperateBuffer,
    OperateCancel, OperateDynamicGate, OperateQuietCancel, OperateScope, OperateSplit,
    OperateStaticGate, Output, Provider, RequestOfMap, ResponseOfMap, Scope, ScopeEndpoints,
    ScopeSettings, ScopeSettingsStorage, Sendish, ServiceInstructions, SplitOutputs, Splittable,
    StreamPack, StreamTargetMap, StreamsOfMap, Trim, TrimBranch, UnusedTarget, Unzippable,
    make_option_branching, make_result_branching,
};

pub(crate) mod connect;
pub(crate) use connect::*;

/// Device used for building a workflow. Simply pass a mutable borrow of this
/// into any functions which ask for it.
///
/// Note that each scope has its own [`Builder`], and a panic will occur if a
/// [`Builder`] gets used in the wrong scope. As of right now there is no known
/// way to trick the compiler into using a [`Builder`] in the wrong scope, but
/// please open an issue with a minimal reproducible example if you find a way
/// to make it panic.
pub struct Builder<'w, 's, 'a> {
    pub(crate) context: BuilderScopeContext,
    pub(crate) commands: &'a mut Commands<'w, 's>,
}

#[derive(Clone, Copy, Debug)]
pub struct BuilderScopeContext {
    /// The scope that this builder is meant to help build
    pub(crate) scope: Entity,
    /// The target for cancellation workflows
    pub(crate) finish_scope_cancel: Entity,
}

impl<'w, 's, 'a> Builder<'w, 's, 'a> {
    /// Begin building a chain of operations off of an output.
    pub fn chain<'b, Response: 'static + Send + Sync>(
        &'b mut self,
        output: Output<Response>,
    ) -> Chain<'w, 's, 'a, 'b, Response> {
        output.chain(self)
    }

    /// Create a node for a provider. This will give access to an input slot, an
    /// output slots, and a pack of stream outputs which can all be connected to
    /// other nodes.
    pub fn create_node<P: Provider>(
        &mut self,
        provider: P,
    ) -> Node<P::Request, P::Response, P::Streams>
    where
        P::Request: 'static + Send + Sync,
        P::Response: 'static + Send + Sync,
        P::Streams: StreamPack,
    {
        let source = self.commands.spawn(()).id();
        let target = self.commands.spawn(UnusedTarget).id();
        provider.connect(Some(self.scope()), source, target, self.commands);

        let mut map = StreamTargetMap::default();
        let streams = <P::Streams as StreamPack>::spawn_node_streams(source, &mut map, self);
        self.commands.entity(source).insert(map);
        Node {
            input: InputSlot::new(self.scope(), source),
            output: Output::new(self.scope(), target),
            streams,
        }
    }

    /// Create a [node](Node) that provides a [blocking map](crate::BlockingMap).
    pub fn create_map_block<T, U>(
        &mut self,
        f: impl FnMut(T) -> U + 'static + Send + Sync,
    ) -> Node<T, U, ()>
    where
        T: 'static + Send + Sync,
        U: 'static + Send + Sync,
    {
        self.create_node(f.into_blocking_map())
    }

    /// Create a [node](Node) that provides an [async map](crate::AsyncMap).
    pub fn create_map_async<T, Task>(
        &mut self,
        f: impl FnMut(T) -> Task + 'static + Send + Sync,
    ) -> Node<T, Task::Output, ()>
    where
        T: 'static + Send + Sync,
        Task: Future + 'static + Sendish,
        Task::Output: 'static + Send + Sync,
    {
        self.create_node(f.into_async_map())
    }

    /// Create a map (either a [blocking map][1] or an
    /// [async map][2]) by providing a function that takes [`BlockingMap`][1] or
    /// [AsyncMap][2] as its only argument.
    ///
    /// [1]: crate::BlockingMap
    /// [2]: crate::AsyncMap
    pub fn create_map<M, F: AsMap<M>>(
        &mut self,
        f: F,
    ) -> Node<RequestOfMap<M, F>, ResponseOfMap<M, F>, StreamsOfMap<M, F>>
    where
        F::MapType: Provider,
        RequestOfMap<M, F>: 'static + Send + Sync,
        ResponseOfMap<M, F>: 'static + Send + Sync,
        StreamsOfMap<M, F>: StreamPack,
    {
        self.create_node(f.as_map())
    }

    /// Create a node that takes in a `(request, service)` at runtime and then
    /// passes the `request` into the `service`. All streams will be forwarded
    /// and the response of the service will be the node's output.
    ///
    /// This allows services to be injected into workflows as input, or for a
    /// service to be chosen during runtime.
    pub fn create_injection_node<Request, Response, Streams>(
        &mut self,
    ) -> Node<(Request, ServiceInstructions<Request, Response, Streams>), Response, Streams>
    where
        Request: 'static + Send + Sync,
        Response: 'static + Send + Sync + Unpin,
        Streams: StreamPack,
    {
        let source = self.commands.spawn(()).id();
        self.create_injection_impl::<Request, Response, Streams>(source)
    }

    /// Connect the output of one into the input slot of another node.
    pub fn connect<T: 'static + Send + Sync>(&mut self, output: Output<T>, input: InputSlot<T>) {
        assert_eq!(output.scope(), input.scope());
        self.commands.queue(Connect {
            original_target: output.id(),
            new_target: input.id(),
        });
    }

    /// Create a [`Buffer`] which can be used to store and pull data within
    /// a scope. This is often used along with joining to synchronize multiple
    /// branches.
    pub fn create_buffer<T: 'static + Send + Sync>(
        &mut self,
        settings: BufferSettings,
    ) -> Buffer<T> {
        let source = self.commands.spawn(()).id();
        self.commands.queue(AddOperation::new(
            Some(self.scope()),
            source,
            OperateBuffer::<T>::new(settings),
        ));

        Buffer {
            location: BufferLocation {
                scope: self.scope(),
                source,
            },
            _ignore: Default::default(),
        }
    }

    /// Create an isolated scope within the workflow. This can be useful for
    /// racing multiple branches, creating an uninterruptible segment within
    /// your workflow, or being able to run the same multiple instances of the
    /// same sub-workflow in parallel without them interfering with each other.
    ///
    /// A value can be sent into the scope by connecting an [`Output`] of a node
    /// in the parent scope to the [`InputSlot`] of the node which gets returned
    /// by this function. Each time a value is sent into the scope, it will run
    /// through the workflow of the scope with a unique session ID. Even if
    /// multiple values are sent in from the same session, they will each be
    /// assigned their own unique session ID while inside of this scope.
    pub fn create_scope<Request, Response, Streams, Settings>(
        &mut self,
        build: impl FnOnce(Scope<Request, Response, Streams>, &mut Builder) -> Settings,
    ) -> Node<Request, Response, Streams>
    where
        Request: 'static + Send + Sync,
        Response: 'static + Send + Sync,
        Streams: StreamPack,
        Settings: Into<ScopeSettings>,
    {
        let scope_id = self.commands.spawn(()).id();
        let exit_scope = self.commands.spawn(UnusedTarget).id();
        self.create_scope_impl(scope_id, exit_scope, build)
    }

    /// Alternative to [`Self::create_scope`] for pure input/output scopes (i.e.
    /// there are no output streams). Using this signature should allow the
    /// compiler to infer all the generic arguments when there are no streams.
    pub fn create_io_scope<Request, Response, Settings>(
        &mut self,
        build: impl FnOnce(Scope<Request, Response, ()>, &mut Builder) -> Settings,
    ) -> Node<Request, Response, ()>
    where
        Request: 'static + Send + Sync,
        Response: 'static + Send + Sync,
        Settings: Into<ScopeSettings>,
    {
        self.create_scope::<Request, Response, (), Settings>(build)
    }

    /// Create an operation that clones its inputs and sends them off to any
    /// number of targets.
    pub fn create_fork_clone<T>(&mut self) -> (InputSlot<T>, ForkCloneOutput<T>)
    where
        T: Clone + 'static + Send + Sync,
    {
        let source = self.commands.spawn(()).id();
        self.commands.queue(AddOperation::new(
            Some(self.scope()),
            source,
            ForkClone::<T>::new(ForkTargetStorage::new()),
        ));
        (
            InputSlot::new(self.scope(), source),
            ForkCloneOutput::new(self.scope(), source),
        )
    }

    /// Create an operation that unzips its inputs and sends each element off to
    /// a different output.
    pub fn create_unzip<T>(&mut self) -> (InputSlot<T>, T::Unzipped)
    where
        T: Unzippable + 'static + Send + Sync,
    {
        let source = self.commands.spawn(()).id();
        (
            InputSlot::new(self.scope(), source),
            T::unzip_output(Output::<T>::new(self.scope(), source), self),
        )
    }

    /// Create an operation that creates a fork for a [`Result`] input. The value
    /// inside the [`Result`] will be unpacked and sent down a different branch
    /// depending on whether it was in the [`Ok`] or [`Err`] variant.
    pub fn create_fork_result<T, E>(&mut self) -> (InputSlot<Result<T, E>>, ForkResultOutput<T, E>)
    where
        T: 'static + Send + Sync,
        E: 'static + Send + Sync,
    {
        let source = self.commands.spawn(()).id();
        let target_ok = self.commands.spawn(UnusedTarget).id();
        let target_err = self.commands.spawn(UnusedTarget).id();

        self.commands.queue(AddOperation::new(
            Some(self.scope()),
            source,
            make_result_branching::<T, E>(ForkTargetStorage::from_iter([target_ok, target_err])),
        ));

        (
            InputSlot::new(self.scope(), source),
            ForkResultOutput {
                ok: Output::new(self.scope(), target_ok),
                err: Output::new(self.scope(), target_err),
            },
        )
    }

    /// Create an operation that creates a fork for an [`Option`] input. The value
    /// inside the [`Option`] will be unpacked and sent down a different branch
    /// depending on whether it was in the [`Some`] or [`None`] variant.
    ///
    /// For the [`None`] variant a unit `()` output will be sent, also called
    /// a trigger.
    pub fn create_fork_option<T>(&mut self) -> (InputSlot<Option<T>>, ForkOptionOutput<T>)
    where
        T: 'static + Send + Sync,
    {
        let source = self.commands.spawn(()).id();
        let target_some = self.commands.spawn(UnusedTarget).id();
        let target_none = self.commands.spawn(UnusedTarget).id();

        self.commands.queue(AddOperation::new(
            Some(self.scope()),
            source,
            make_option_branching::<T>(ForkTargetStorage::from_iter([target_some, target_none])),
        ));

        (
            InputSlot::new(self.scope(), source),
            ForkOptionOutput {
                some: Output::new(self.scope(), target_some),
                none: Output::new(self.scope(), target_none),
            },
        )
    }

    /// Alternative way of calling [`Joinable::join`]
    pub fn join<'b, B: Joinable>(&'b mut self, buffers: B) -> Chain<'w, 's, 'a, 'b, B::Item> {
        buffers.join(self)
    }

    /// Try joining a map of buffers into a single value.
    pub fn try_join<'b, J: Joined>(
        &'b mut self,
        buffers: &BufferMap,
    ) -> Result<Chain<'w, 's, 'a, 'b, J>, IncompatibleLayout> {
        J::try_join_from(buffers, self)
    }

    /// Alternative way of calling [`Accessible::listen`].
    pub fn listen<'b, B: Accessible>(&'b mut self, buffers: B) -> Chain<'w, 's, 'a, 'b, B::Keys> {
        buffers.listen(self)
    }

    /// Try listening to a map of buffers.
    pub fn try_listen<'b, Keys: Accessor>(
        &'b mut self,
        buffers: &BufferMap,
    ) -> Result<Chain<'w, 's, 'a, 'b, Keys>, IncompatibleLayout> {
        Keys::try_listen_from(buffers, self)
    }

    /// Create a node that combines its inputs with access to some buffers. You
    /// must specify one ore more buffers to access. FOr multiple buffers,
    /// combine then into a tuple or an [`Iterator`]. Tuples of buffers can be
    /// nested inside each other.
    ///
    /// Other [outputs](Output) can also be passed in as buffers. These outputs
    /// will be transformed into a buffer with default buffer settings.
    pub fn create_buffer_access<T, B: Bufferable>(
        &mut self,
        buffers: B,
    ) -> Node<T, (T, BufferKeys<B>)>
    where
        B::BufferType: Accessing,
        T: 'static + Send + Sync,
    {
        let buffers = buffers.into_buffer(self);
        buffers.access(self)
    }

    /// Try to create access to some buffers. Same as [`Self::create_buffer_access`]
    /// except it will return an error if the buffers in the [`BufferMap`] are not
    /// compatible with the keys that are being asked for.
    pub fn try_create_buffer_access<T, Keys: Accessor>(
        &mut self,
        buffers: &BufferMap,
    ) -> Result<Node<T, (T, Keys)>, IncompatibleLayout>
    where
        T: 'static + Send + Sync,
    {
        Keys::try_buffer_access(buffers, self)
    }

    /// Collect incoming workflow threads into a container.
    ///
    /// If `max` is specified, the collection will always be sent out once it
    /// reaches that maximum value.
    ///
    /// If `min` is greater than 0 then the collection will not be sent out unless
    /// it is equal to or greater than that value. Note that this means the
    /// collect operation could force the workflow into cancelling if it cannot
    /// reach the minimum number of elements. A `min` of 0 means that if an
    /// upstream thread is disposed and the collect node is no longer reachable
    /// then it will fire off with an empty collection.
    ///
    /// If the `min` limit is satisfied and there are no remaining workflow
    /// threads that can reach this collect operation, then the collection will
    /// be sent out with however many elements it happens to have collected.
    pub fn create_collect<T, const N: usize>(
        &mut self,
        min: usize,
        max: Option<usize>,
    ) -> Node<T, SmallVec<[T; N]>>
    where
        T: 'static + Send + Sync,
    {
        if let Some(max) = max {
            assert!(0 < max);
            assert!(min <= max);
        }

        let source = self.commands.spawn(()).id();
        let target = self.commands.spawn(UnusedTarget).id();
        self.commands.queue(AddOperation::new(
            Some(self.scope()),
            source,
            Collect::<T, N>::new(target, min, max),
        ));

        Node {
            input: InputSlot::new(self.scope(), source),
            output: Output::new(self.scope(), target),
            streams: (),
        }
    }

    /// Collect all workflow threads that are moving towards this node.
    pub fn create_collect_all<T, const N: usize>(&mut self) -> Node<T, SmallVec<[T; N]>>
    where
        T: 'static + Send + Sync,
    {
        self.create_collect(0, None)
    }

    /// Collect an exact number of threads that are moving towards this node.
    pub fn create_collect_n<T, const N: usize>(&mut self, n: usize) -> Node<T, SmallVec<[T; N]>>
    where
        T: 'static + Send + Sync,
    {
        self.create_collect(n, Some(n))
    }

    /// Create a new split operation in the workflow. The [`InputSlot`] can take
    /// in values that you want to split, and [`SplitOutputs::build`] will let
    /// you build connections to the split value.
    pub fn create_split<T>(&mut self) -> (InputSlot<T>, SplitOutputs<T>)
    where
        T: 'static + Send + Sync + Splittable,
    {
        let source = self.commands.spawn(()).id();
        self.commands.queue(AddOperation::new(
            Some(self.scope()),
            source,
            OperateSplit::<T>::default(),
        ));

        (
            InputSlot::new(self.scope(), source),
            SplitOutputs::new(self.scope(), source),
        )
    }

    /// Create an input slot that will cancel the current scope when it gets
    /// triggered. This can be used on types that implement [`ToString`].
    ///
    /// If you need to cancel for a type that does not implement [`ToString`]
    /// then convert it to a trigger `()` and then connect it to
    /// [`Self::create_quiet_cancel`].
    pub fn create_cancel<T>(&mut self) -> InputSlot<T>
    where
        T: 'static + Send + Sync + ToString,
    {
        let source = self.commands.spawn(()).id();
        self.commands.queue(AddOperation::new(
            Some(self.scope()),
            source,
            OperateCancel::<T>::new(),
        ));

        InputSlot::new(self.scope(), source)
    }

    /// Create an input slot that will cancel that current scope when it gets
    /// triggered.
    ///
    /// If you want the cancellation message to include information about the
    /// input value that triggered it, use [`Self::create_cancel`].
    pub fn create_quiet_cancel(&mut self) -> InputSlot<()> {
        let source = self.commands.spawn(()).id();
        self.commands.queue(AddOperation::new(
            Some(self.scope()),
            source,
            OperateQuietCancel,
        ));

        InputSlot::new(self.scope(), source)
    }

    /// This method allows you to define a cleanup workflow that branches off of
    /// this scope that will activate during the scope's cleanup phase. The
    /// input to the cleanup workflow will be a key to access to one or more
    /// buffers from the parent scope.
    ///
    /// Each different cleanup workflow that you define using this function will
    /// be given its own unique session ID when it gets run. You can define any
    /// number of cleanup workflows.
    ///
    /// The parent scope will only finish its cleanup phase after all cleanup
    /// workflows for the scope have finished, either by terminating or by being
    /// cancelled themselves.
    ///
    /// Cleanup workflows that you define with this function will always be run
    /// no matter how the scope finished. If you want a cleanup workflow that
    /// only runs when the scope is cancelled, use [`Self::on_cancel`]. If you
    /// want a cleanup workflow that only runs when the scope terminates
    /// successfully, then use [`Self::on_terminate`]. For easier runtime
    /// flexibility you can also use [`Self::on_cleanup_if`].
    pub fn on_cleanup<B, Settings>(
        &mut self,
        from_buffers: B,
        build: impl FnOnce(Scope<BufferKeys<B>, (), ()>, &mut Builder) -> Settings,
    ) where
        B: Bufferable,
        B::BufferType: Accessing,
        Settings: Into<ScopeSettings>,
    {
        from_buffers.into_buffer(self).on_cleanup(self, build);
    }

    /// Define a cleanup workflow that only gets run if the scope was cancelled.
    ///
    /// When the scope gets dropped before it terminates (i.e. the parent scope
    /// finished while this scope was active, and this scope is interruptible)
    /// that also counts as cancelled.
    ///
    /// A scope which is set to be uninterruptible will still experience a
    /// cancellation if its terminal node becomes unreachable.
    ///
    /// If you want a cleanup workflow that only runs when the scope terminates
    /// successfully then use [`Self::on_terminate`]. If you want a cleanup
    /// workflow that always runs when the scope is finished, then use
    /// [`Self::on_cleanup`].
    pub fn on_cancel<B, Settings>(
        &mut self,
        from_buffers: B,
        build: impl FnOnce(Scope<BufferKeys<B>, (), ()>, &mut Builder) -> Settings,
    ) where
        B: Bufferable,
        B::BufferType: Accessing,
        Settings: Into<ScopeSettings>,
    {
        from_buffers.into_buffer(self).on_cancel(self, build);
    }

    /// Define a cleanup workflow that only gets run if the scope was successfully
    /// terminated. That means an input reached its terminal node, and now the
    /// scope is cleaning up.
    ///
    /// If you want a cleanup workflow that only runs when the scope is cancelled
    /// then use [`Self::on_cancel`]. If you want a cleanup workflow that always
    /// runs when then scope is finished, then use [`Self::on_cleanup`].
    pub fn on_terminate<B, Settings>(
        &mut self,
        from_buffers: B,
        build: impl FnOnce(Scope<BufferKeys<B>, (), ()>, &mut Builder) -> Settings,
    ) where
        B: Bufferable,
        B::BufferType: Accessing,
        Settings: Into<ScopeSettings>,
    {
        from_buffers.into_buffer(self).on_terminate(self, build);
    }

    /// Define a sub-workflow that will be run when this workflow is being cleaned
    /// up if the conditions are met. A workflow enters the cleanup stage when
    /// its terminal node is reached or if it gets cancelled.
    pub fn on_cleanup_if<B, Settings>(
        &mut self,
        conditions: CleanupWorkflowConditions,
        from_buffers: B,
        build: impl FnOnce(Scope<BufferKeys<B>, (), ()>, &mut Builder) -> Settings,
    ) where
        B: Bufferable,
        B::BufferType: Accessing,
        Settings: Into<ScopeSettings>,
    {
        from_buffers
            .into_buffer(self)
            .on_cleanup_if(self, conditions, build);
    }

    /// Create a node that trims (cancels) other nodes in the workflow when it
    /// gets activated. The input into the node will be passed along as output
    /// after the trimming is confirmed to be completed.
    pub fn create_trim<T>(&mut self, branches: impl IntoIterator<Item = TrimBranch>) -> Node<T, T>
    where
        T: 'static + Send + Sync,
    {
        let branches: SmallVec<[_; 16]> = branches.into_iter().collect();
        for branch in &branches {
            branch.verify_scope(self.scope());
        }

        let source = self.commands.spawn(()).id();
        let target = self.commands.spawn(UnusedTarget).id();
        self.commands.queue(AddOperation::new(
            Some(self.scope()),
            source,
            Trim::<T>::new(branches, target),
        ));

        Node {
            input: InputSlot::new(self.scope(), source),
            output: Output::new(self.scope(), target),
            streams: (),
        }
    }

    /// Create a gate node that can open and close the gates on one or more
    /// buffers. Feed a [`GateRequest`] into the node and all the associated
    /// buffers will be opened or closed based on the action inside the request.
    ///
    /// The data inside the request will be passed along as output once the
    /// gate action is finished.
    ///
    /// See [`Gate`] to understand what happens when a gate is opened or closed.
    pub fn create_gate<T, B>(&mut self, buffers: B) -> Node<GateRequest<T>, T>
    where
        B: Bufferable,
        T: 'static + Send + Sync,
    {
        let buffers = buffers.into_buffer(self);
        buffers.verify_scope(self.scope());

        let source = self.commands.spawn(()).id();
        let target = self.commands.spawn(UnusedTarget).id();
        self.commands.queue(AddOperation::new(
            Some(self.scope()),
            source,
            OperateDynamicGate::<T, _>::new(buffers, target),
        ));

        Node {
            input: InputSlot::new(self.scope(), source),
            output: Output::new(self.scope(), target),
            streams: (),
        }
    }

    /// Create a gate node that always opens or always closes the gates on one
    /// or more buffers.
    ///
    /// The data sent into the node will be passed back out as input, unchanged.
    ///
    /// See [`Gate`] to understand what happens when a gate is opened or closed.
    pub fn create_gate_action<T, B>(&mut self, action: Gate, buffers: B) -> Node<T, T>
    where
        B: Bufferable,
        T: 'static + Send + Sync,
    {
        let buffers = buffers.into_buffer(self);
        buffers.verify_scope(self.scope());

        let source = self.commands.spawn(()).id();
        let target = self.commands.spawn(UnusedTarget).id();
        self.commands.queue(AddOperation::new(
            Some(self.scope()),
            source,
            OperateStaticGate::<T, _>::new(buffers, target, action),
        ));

        Node {
            input: InputSlot::new(self.scope(), source),
            output: Output::new(self.scope(), target),
            streams: (),
        }
    }

    /// Create a gate node that always opens the gates on one or more buffers.
    ///
    /// See [`Gate`] to understand what happens when a gate is opened or closed.
    pub fn create_gate_open<B, T>(&mut self, buffers: B) -> Node<T, T>
    where
        B: Bufferable,
        T: 'static + Send + Sync,
    {
        self.create_gate_action(Gate::Open, buffers)
    }

    /// Create a gate node that always closes the gates on one or more buffers.
    ///
    /// See [`Gate`] to understand what happens when a gate is opened or closed.
    pub fn create_gate_close<T, B>(&mut self, buffers: B) -> Node<T, T>
    where
        B: Bufferable,
        T: 'static + Send + Sync,
    {
        self.create_gate_action(Gate::Closed, buffers)
    }

    /// Get the scope that this builder is building for.
    pub fn scope(&self) -> Entity {
        self.context.scope
    }

    /// Borrow the commands for the builder
    pub fn commands(&mut self) -> &mut Commands<'w, 's> {
        self.commands
    }

    /// Used internally to create scopes in different ways.
    pub(crate) fn create_scope_impl<Request, Response, Streams, Settings>(
        &mut self,
        scope_id: Entity,
        exit_scope: Entity,
        build: impl FnOnce(Scope<Request, Response, Streams>, &mut Builder) -> Settings,
    ) -> Node<Request, Response, Streams>
    where
        Request: 'static + Send + Sync,
        Response: 'static + Send + Sync,
        Streams: StreamPack,
        Settings: Into<ScopeSettings>,
    {
        // NOTE(@mxgrey): When changing the implementation of this function,
        // remember to similarly update the implementation of IncrementalScopeBuilder
        let ScopeEndpoints {
            terminal,
            enter_scope,
            finish_scope_cancel,
        } = OperateScope::add::<Request, Response>(
            Some(self.scope()),
            scope_id,
            Some(exit_scope),
            self.commands,
        );

        let (stream_in, stream_out) =
            Streams::spawn_scope_streams(scope_id, self.scope(), self.commands);

        let mut builder = Builder {
            context: BuilderScopeContext {
                scope: scope_id,
                finish_scope_cancel,
            },
            commands: self.commands,
        };

        let scope = Scope {
            start: Output::new(scope_id, enter_scope),
            terminate: InputSlot::new(scope_id, terminal),
            streams: stream_in,
        };

        let settings = build(scope, &mut builder).into();
        self.commands
            .entity(scope_id)
            .insert(ScopeSettingsStorage(settings));

        Node {
            input: InputSlot::new(self.scope(), scope_id),
            output: Output::new(self.scope(), exit_scope),
            streams: stream_out,
        }
    }

    pub(crate) fn create_injection_impl<Request, Response, Streams>(
        &mut self,
        source: Entity,
    ) -> Node<(Request, ServiceInstructions<Request, Response, Streams>), Response, Streams>
    where
        Request: 'static + Send + Sync,
        Response: 'static + Send + Sync,
        Streams: StreamPack,
    {
        let target = self.commands.spawn(UnusedTarget).id();

        let mut map = StreamTargetMap::default();
        let streams = Streams::spawn_node_streams(source, &mut map, self);
        self.commands.entity(source).insert(map);
        self.commands.queue(AddOperation::new(
            Some(self.scope()),
            source,
            Injection::<Request, Response, Streams>::new(target),
        ));

        Node {
            input: InputSlot::new(self.scope(), source),
            output: Output::new(self.scope(), target),
            streams,
        }
    }

    pub fn context(&self) -> BuilderScopeContext {
        self.context
    }
}

/// This struct is used to describe when a cleanup workflow should run. Currently
/// this is only two simple booleans, but in the future it could potentially
/// contain systems that make decisions at runtime, so for now we encapsulate
/// the idea of cleanup conditions in this struct so we enhance the capabilities
/// later without breaking API.
#[derive(Clone)]
pub struct CleanupWorkflowConditions {
    pub(crate) run_on_terminate: bool,
    pub(crate) run_on_cancel: bool,
}

impl CleanupWorkflowConditions {
    pub fn always_if(run_on_terminate: bool, run_on_cancel: bool) -> Self {
        CleanupWorkflowConditions {
            run_on_terminate,
            run_on_cancel,
        }
    }
}

#[cfg(test)]
mod tests {
    use crate::{CancellationCause, prelude::*, testing::*};
    use smallvec::SmallVec;
    use std::time::Instant;

    #[test]
    fn test_disconnected_workflow() {
        let mut context = TestingContext::minimal_plugins();

        let workflow = context.spawn_io_workflow(|_, _| {
            // Do nothing. Totally empty workflow.
        });
        // Test this repeatedly because we cache the result after the first time.
        check_unreachable(workflow, 1, &mut context);
        check_unreachable(workflow, 1, &mut context);
        check_unreachable(workflow, 1, &mut context);

        let workflow = context.spawn_io_workflow(|scope, builder| {
            let node = builder.create_map_block(|v| v);
            builder.connect(scope.start, node.input);
            // Create a tight infinite loop that will never reach the terminal
            builder.connect(node.output, node.input);
        });
        check_unreachable(workflow, 1, &mut context);
        check_unreachable(workflow, 1, &mut context);
        check_unreachable(workflow, 1, &mut context);

        let workflow = context.spawn_io_workflow(|scope, builder| {
            builder.chain(scope.start).map_block(|v| v).fork_clone((
                |chain: Chain<()>| chain.map_block(|v| v).map_block(|v| v).unused(),
                |chain: Chain<()>| chain.map_block(|v| v).map_block(|v| v).unused(),
                |chain: Chain<()>| chain.map_block(|v| v).map_block(|v| v).unused(),
            ));

            // Create an exit node that never connects to the scope's input.
            let exit_node = builder.create_map_block(|v| v);
            builder.connect(exit_node.output, scope.terminate);
        });
        check_unreachable(workflow, 1, &mut context);
        check_unreachable(workflow, 1, &mut context);
        check_unreachable(workflow, 1, &mut context);

        let workflow = context.spawn_io_workflow(|scope, builder| {
            let entry_buffer = builder.create_buffer::<()>(BufferSettings::keep_all());
            builder.chain(scope.start).map_block(|v| v).fork_clone((
                |chain: Chain<()>| chain.map_block(|v| v).connect(entry_buffer.input_slot()),
                |chain: Chain<()>| chain.map_block(|v| v).connect(entry_buffer.input_slot()),
                |chain: Chain<()>| chain.map_block(|v| v).connect(entry_buffer.input_slot()),
            ));

            // Create an exit buffer with no relationship to the entry buffer
            // which is the only thing that connects to the terminal node.
            let exit_buffer = builder.create_buffer::<()>(BufferSettings::keep_all());
            builder
                .listen(exit_buffer)
                .map_block(|_| ())
                .connect(scope.terminate);
        });
        check_unreachable(workflow, 1, &mut context);
        check_unreachable(workflow, 1, &mut context);
        check_unreachable(workflow, 1, &mut context);
    }

    fn check_unreachable(
        workflow: Service<(), ()>,
        flush_cycles: usize,
        context: &mut TestingContext,
    ) {
        let r = context.try_resolve_request((), workflow, flush_cycles);
        assert!(matches!(
            *r.unwrap_err().cause,
            CancellationCause::Unreachable(_)
        ));
    }

    #[test]
    fn test_fork_clone() {
        let mut context = TestingContext::minimal_plugins();

        let workflow = context.spawn_io_workflow(|scope, builder| {
            let fork = scope.start.fork_clone(builder);
            let branch_a = fork.clone_output(builder);
            let branch_b = fork.clone_output(builder);
            builder.connect(branch_a, scope.terminate);
            builder.connect(branch_b, scope.terminate);
        });

        let r = context.resolve_request(5.0, workflow);
        assert_eq!(r, 5.0);

        let workflow = context.spawn_io_workflow(|scope, builder| {
            builder.chain(scope.start).fork_clone((
                |chain: Chain<f64>| chain.connect(scope.terminate),
                |chain: Chain<f64>| chain.connect(scope.terminate),
            ));
        });

        let r = context.resolve_request(3.0, workflow);
        assert_eq!(r, 3.0);

        let workflow = context.spawn_io_workflow(|scope, builder| {
            builder.chain(scope.start).fork_clone((
                |chain: Chain<f64>| {
                    chain
                        .map_block(|t| WaitRequest {
                            duration: Duration::from_secs_f64(10.0 * t),
                            value: 10.0 * t,
                        })
                        .map(|r: AsyncMap<WaitRequest<f64>>| wait(r.request))
                        .connect(scope.terminate)
                },
                |chain: Chain<f64>| {
                    chain
                        .map_block(|t| WaitRequest {
                            duration: Duration::from_secs_f64(t / 100.0),
                            value: t / 100.0,
                        })
                        .map(|r: AsyncMap<WaitRequest<f64>>| wait(r.request))
                        .connect(scope.terminate)
                },
            ));
        });

        let r = context.resolve_request(1.0, workflow);
        assert_eq!(r, 0.01);

        let workflow = context.spawn_io_workflow(|scope, builder| {
            let (fork_input, fork_output) = builder.create_fork_clone();
            builder.connect(scope.start, fork_input);
            let a = fork_output.clone_output(builder);
            let b = fork_output.clone_output(builder);
            builder.join((a, b)).connect(scope.terminate);
        });

        let r = context.resolve_request(5, workflow);
        assert_eq!(r, (5, 5));
    }

    #[test]
    fn test_stream_reachability() {
        let mut context = TestingContext::minimal_plugins();

        // Test for streams from a blocking node
        let workflow = context.spawn_io_workflow(|scope, builder| {
            let stream_node = builder.create_map(|_: BlockingMap<(), StreamOf<u32>>| {
                // Do nothing. The purpose of this node is to just return without
                // sending off any streams.
            });

            builder.connect(scope.start, stream_node.input);
            builder
                .chain(stream_node.streams)
                .map_block(|value| 2 * value)
                .connect(scope.terminate);
        });

        let r = context.try_resolve_request((), workflow, ());
        assert!(r.is_err());

        // Test for streams from an async node
        let workflow = context.spawn_io_workflow(|scope, builder| {
            let stream_node = builder.create_map(|_: AsyncMap<(), StreamOf<u32>>| {
                async { /* Do nothing */ }
            });

            builder.connect(scope.start, stream_node.input);
            builder
                .chain(stream_node.streams)
                .map_block(|value| 2 * value)
                .connect(scope.terminate);
        });

        let r = context.try_resolve_request((), workflow, ());
        assert!(r.is_err());
    }

    use tokio::sync::mpsc::unbounded_channel;

    #[test]
    fn test_on_cleanup() {
        let mut context = TestingContext::minimal_plugins();

        let (sender, mut receiver) = unbounded_channel();
        let workflow = context.spawn_io_workflow(|scope, builder| {
            let input = scope.start.fork_clone(builder);

            let buffer = builder.create_buffer(BufferSettings::default());
            let input_to_buffer = input.clone_output(builder);
            builder.connect(input_to_buffer, buffer.input_slot());

            let none_node = builder.create_map_block(produce_none);
            let input_to_node = input.clone_output(builder);
            builder.connect(input_to_node, none_node.input);
            builder
                .chain(none_node.output)
                .cancel_on_none()
                .connect(scope.terminate);

            // The chain coming out of the none_node will result in the scope
            // being cancelled. After that, this scope should run, and the value
            // that went into the buffer should get sent over the channel.
            builder.on_cancel(buffer, |scope, builder| {
                builder
                    .chain(scope.start)
                    .consume_buffer::<8>()
                    .map_block(move |values| {
                        for value in values {
                            sender.send(value).unwrap();
                        }
                    })
                    .connect(scope.terminate);
            });
        });

        let r = context.try_resolve_request(5, workflow, ());
        assert!(r.is_err());

        let channel_output = receiver.try_recv().unwrap();
        assert_eq!(channel_output, 5);
        assert!(receiver.try_recv().is_err());
        assert!(context.confirm_buffers_empty().is_ok());

        let (cancel_sender, mut cancel_receiver) = unbounded_channel();
        let (terminate_sender, mut terminate_receiver) = unbounded_channel();
        let (cleanup_sender, mut cleanup_receiver) = unbounded_channel();
        let workflow = context.spawn_io_workflow(|scope, builder| {
            let input = scope.start.fork_clone(builder);

            let cancel_buffer = builder.create_buffer(BufferSettings::default());
            let input_to_cancel = input.clone_output(builder);
            builder.connect(input_to_cancel, cancel_buffer.input_slot());

            let terminate_buffer = builder.create_buffer(BufferSettings::default());
            let input_to_terminate = input.clone_output(builder);
            builder.connect(input_to_terminate, terminate_buffer.input_slot());

            let cleanup_buffer = builder.create_buffer(BufferSettings::default());
            let input_to_cleanup = input.clone_output(builder);
            builder.connect(input_to_cleanup, cleanup_buffer.input_slot());

            let filter_node =
                builder.create_map_block(|value: u64| if value >= 5 { Some(value) } else { None });
            let input_to_filter_node = input.clone_output(builder);
            builder.connect(input_to_filter_node, filter_node.input);
            builder
                .chain(filter_node.output)
                .cancel_on_none()
                .connect(scope.terminate);

            builder.on_cancel(cancel_buffer, |scope, builder| {
                builder
                    .chain(scope.start)
                    .consume_buffer::<8>()
                    .map_block(move |values| {
                        for value in values {
                            cancel_sender.send(value).unwrap();
                        }
                    })
                    .connect(scope.terminate);
            });

            builder.on_terminate(terminate_buffer, |scope, builder| {
                builder
                    .chain(scope.start)
                    .consume_buffer::<8>()
                    .map_block(move |values| {
                        for value in values {
                            terminate_sender.send(value).unwrap();
                        }
                    })
                    .connect(scope.terminate);
            });

            builder.on_cleanup(cleanup_buffer, |scope, builder| {
                builder
                    .chain(scope.start)
                    .consume_buffer::<8>()
                    .map_block(move |values| {
                        for value in values {
                            cleanup_sender.send(value).unwrap();
                        }
                    })
                    .connect(scope.terminate);
            });
        });

        let r = context.try_resolve_request(3, workflow, 10);
        assert!(r.is_err());

        assert_eq!(cancel_receiver.try_recv().unwrap(), 3);
        assert!(cancel_receiver.try_recv().is_err());
        assert_eq!(cleanup_receiver.try_recv().unwrap(), 3);
        assert!(cleanup_receiver.try_recv().is_err());
        assert!(terminate_receiver.try_recv().is_err());
        assert!(context.no_unhandled_errors());
        assert!(context.confirm_buffers_empty().is_ok());

        let r = context.try_resolve_request(6, workflow, 10).unwrap();
        assert_eq!(r, 6);

        assert_eq!(terminate_receiver.try_recv().unwrap(), 6);
        assert!(terminate_receiver.try_recv().is_err());
        assert_eq!(cleanup_receiver.try_recv().unwrap(), 6);
        assert!(cleanup_receiver.try_recv().is_err());
        assert!(cancel_receiver.try_recv().is_err());
        assert!(context.no_unhandled_errors());
        assert!(context.confirm_buffers_empty().is_ok());
    }

    #[test]
    fn test_double_collection() {
        let mut context = TestingContext::minimal_plugins();

        let delay = context.spawn_delay(Duration::from_secs_f32(0.01));

        let workflow = context.spawn_io_workflow(|scope, builder| {
            // We make the later collect node first so that its disposal update
            // gets triggered first. If we don't implement collect correctly
            // then the later collect may think it's unreachable and send out
            // its collection prematurely. This test is checking for that edge
            // case.
            let later_collect = builder.create_collect_all::<i32, 8>();
            let earlier_collect = builder.create_collect_all::<i32, 8>();

            builder
                .chain(scope.start)
                .spread()
                .then(delay)
                .map_block(|v| if v <= 4 { Some(v) } else { None })
                .dispose_on_none()
                .connect(earlier_collect.input);

            builder
                .chain(earlier_collect.output)
                .spread()
                .connect(later_collect.input);

            builder.connect(later_collect.output, scope.terminate);
        });

        let r = context.resolve_request([1, 2, 3, 4, 5], workflow);
        assert_eq!(r.as_slice(), &[1, 2, 3, 4]);

        let workflow = context.spawn_io_workflow(|scope, builder| {
            // We create a circular dependency between two collect operations.
            // This makes it impossible to determine which should fire when the
            // upstream is exhausted because both collect operations are upstream
            // of each other. This workflow should be cancelled before it even
            // starts.
            let earlier_collect = builder.create_collect_all::<i32, 8>();
            let later_collect = builder.create_collect_all::<i32, 8>();

            builder
                .chain(scope.start)
                .spread()
                .then(delay)
                .connect(earlier_collect.input);

            builder
                .chain(earlier_collect.output)
                .spread()
                .connect(later_collect.input);

            builder.chain(later_collect.output).spread().fork_clone((
                |chain: Chain<i32>| chain.connect(earlier_collect.input),
                |chain: Chain<i32>| chain.connect(scope.terminate),
            ));
        });

        let r = context.try_resolve_request([1, 2, 3, 4, 5], workflow, ());
        assert!(r.is_err());

        let workflow = context.spawn_io_workflow(|scope, builder| {
            // We create a circulaur dependency between two collect operations,
            // but this time one of those collect operations is scoped, and that
            // disambiguates the behavior of the collect operations. We should
            // get the intuitive workflow output from this.
            let earlier_collect = builder.create_collect_all::<i32, 8>();

            builder
                .chain(scope.start)
                .spread()
                .then(delay)
                .map_block(|v| if v <= 4 { Some(v) } else { None })
                .dispose_on_none()
                .connect(earlier_collect.input);

            let _ = builder
                .chain(earlier_collect.output)
                .then_io_scope(|scope, builder| {
                    builder
                        .chain(scope.start)
                        .spread()
                        .collect_all::<8>()
                        .connect(scope.terminate);
                })
                .fork_clone((
                    |chain: Chain<_>| chain.spread().connect(earlier_collect.input),
                    |chain: Chain<_>| chain.connect(scope.terminate),
                ));
        });

        check_collections(workflow, [1, 2, 3, 4], [1, 2, 3, 4], &mut context);
        check_collections(workflow, [1, 2, 3, 4, 5, 6], [1, 2, 3, 4], &mut context);
        check_collections(workflow, [1, 8, 2, 7, 3, 6], [1, 2, 3], &mut context);
        check_collections(
            workflow,
            [8, 7, 6, 5, 4, 3, 2, 1],
            [4, 3, 2, 1],
            &mut context,
        );
        check_collections(workflow, [6, 7, 8, 9, 10], [], &mut context);
    }

    fn check_collections(
        workflow: Service<SmallVec<[i32; 8]>, SmallVec<[i32; 8]>>,
        input: impl IntoIterator<Item = i32>,
        expectation: impl IntoIterator<Item = i32>,
        context: &mut TestingContext,
    ) {
        let input: SmallVec<[i32; 8]> = SmallVec::from_iter(input);
        let expectation: SmallVec<[i32; 8]> = SmallVec::from_iter(expectation);
        let r = context.resolve_request(input, workflow);
        assert_eq!(r, expectation);
    }

    #[test]
    fn benchmarks() {
        let mut context = TestingContext::minimal_plugins();

        let workflow = context.spawn_io_workflow(|scope, builder| {
            let (start_test, end_test) = build_benchmark_fixture(scope, builder);
            builder.connect(start_test, end_test);
        });

        let result = context
            .try_resolve_request((), workflow, Duration::from_secs(10))
            .unwrap();
        println!("Performance for basic connection:\n{result:#?}");

        let workflow = context.spawn_io_workflow(|scope, builder| {
            let (start_test, end_test) = build_benchmark_fixture(scope, builder);
            builder
                .chain(start_test)
                .then_io_scope(|scope, builder| {
                    builder.connect(scope.start, scope.terminate);
                })
                .connect(end_test);
        });

        let result = context
            .try_resolve_request((), workflow, Duration::from_secs(10))
            .unwrap();
        println!("Performance for basic connection:\n{result:#?}");
    }

    fn build_benchmark_fixture(
        scope: Scope<(), TimeStats>,
        builder: &mut Builder,
    ) -> (Output<Instant>, InputSlot<Instant>) {
        let initial_time = builder
            .commands()
            .spawn_service(get_initial_time.into_blocking_service());
        let finish_time = builder
            .commands()
            .spawn_service(finish_time_range.into_blocking_service());
        let collect_samples = builder
            .commands()
            .spawn_service(collect_samples.into_blocking_service());

        let samples = builder.create_buffer(BufferSettings::keep_all());

        let initial_node = builder.create_node(initial_time);
        let finish_node = builder.create_node(finish_time);

        builder.connect(scope.start, initial_node.input);
        builder.connect(finish_node.output, samples.input_slot());

        builder
            .listen(samples)
            .then(collect_samples)
            .dispose_on_none()
            .connect(scope.terminate);

        builder
            .listen(samples)
            .map_block(|_| ())
            .connect(initial_node.input);

        (initial_node.output, finish_node.input)
    }

    #[derive(Debug, Clone, Copy)]
    struct TimeRange {
        initial_time: Instant,
        finish_time: Instant,
    }

    #[derive(Debug, Clone, Copy)]
    struct TimeStats {
        #[allow(unused)]
        sample_count: usize,
        #[allow(unused)]
        average: Duration,
        #[allow(unused)]
        std_dev: Duration,
        #[allow(unused)]
        highest: Duration,
        #[allow(unused)]
        lowest: Duration,
    }

    impl TimeStats {
        fn new(samples: impl IntoIterator<Item = TimeRange>) -> Self {
            let samples: Vec<_> = samples
                .into_iter()
                .map(|s| s.finish_time - s.initial_time)
                .collect();
            let sample_count = samples.len();

            let mut highest = None;
            let mut lowest = None;
            let mut average = Duration::new(0, 0);
            for sample in &samples {
                average += *sample;
                if highest.is_none_or(|h| *sample > h) {
                    highest = Some(*sample);
                }

                if lowest.is_none_or(|l| *sample < l) {
                    lowest = Some(*sample);
                }
            }

            let average = average / sample_count as u32;
            let mut radicand = 0.0;
            for sample in samples {
                let delta = (sample.as_nanos() as i64 - average.as_nanos() as i64) as f64;
                radicand += f64::powf(delta, 2.0);
            }

            let std_dev = Duration::from_nanos(f64::sqrt(radicand) as u64);
            let highest = highest.unwrap();
            let lowest = lowest.unwrap();
            TimeStats {
                sample_count,
                average,
                std_dev,
                highest,
                lowest,
            }
        }
    }

    fn get_initial_time(_: In<()>) -> Instant {
        Instant::now()
    }

    fn finish_time_range(In(initial_time): In<Instant>) -> TimeRange {
        TimeRange {
            initial_time,
            finish_time: Instant::now(),
        }
    }

    fn collect_samples(
        In(key): In<BufferKey<TimeRange>>,
        access: BufferAccess<TimeRange>,
    ) -> Option<TimeStats> {
        let samples = access.get(&key).unwrap();
        if samples.len() >= 1000 {
            Some(TimeStats::new(samples.iter().copied()))
        } else {
            None
        }
    }
}