dex_sdk/

wait.rs

// Copyright (c) 2026 Super Durable, Inc.
//
// Licensed under the Sustainable Use License 1.0.
// You may not use this file except in compliance with the License.
// See the LICENSE file in the repository root.
//
// SPDX-License-Identifier: LicenseRef-Sustainable-Use-1.0

use std::any::TypeId;
use std::sync::Arc;
use std::time::Duration;

use dex_protocol::dex::Value as ProtoValue;

use crate::SubFlowOptions;

/// Describes when Dex may invoke a Step's `execute` handler.
///
/// Build waits from durable timer and Channel [`Condition`] values. Combinators consume their
/// inputs and preserve order, which determines timer indexes used by [`crate::TimerId`].
///
/// # Examples
///
/// ```
/// use dex_sdk::{Channel, Timer, Wait};
/// use std::time::Duration;
///
/// let approvals = Channel::<String>::new("approvals");
/// let wait = Wait::any_of([
///     approvals.for_one(),
///     Timer::by_duration(Duration::from_secs(300)),
/// ]);
/// ```
pub struct Wait {
    pub(crate) kind: WaitKind,
}

pub(crate) enum WaitKind {
    SkipImmediately,
    AllOf(Vec<Condition>),
    AnyOf(Vec<Condition>),
    AnyCombinationOf(Vec<ConditionCombination>),
}

impl Wait {
    /// Skips `wait_for` and asks Dex to invoke `execute` immediately.
    pub fn skip_immediately() -> Self {
        Self {
            kind: WaitKind::SkipImmediately,
        }
    }

    /// Waits until one condition is satisfied. The Condition does not need an ID.
    pub fn until(condition: Condition) -> Self {
        Self::all_of([condition])
    }

    /// Waits until every condition is satisfied. Conditions do not need IDs.
    pub fn all_of(conditions: impl IntoIterator<Item = Condition>) -> Self {
        Self {
            kind: WaitKind::AllOf(conditions.into_iter().collect()),
        }
    }

    /// Waits until at least one condition is satisfied. Conditions do not need IDs.
    ///
    /// Among ready conditions, Dex selects the first in canonical order: Timers, Channels, then
    /// SubFlows, preserving iterator order within each kind. An unready condition does not block a
    /// later ready condition. To wait for strict priority, return only the higher-priority
    /// condition.
    ///
    /// Channel consumption is not greedy across alternatives. Only the selected Channel condition
    /// consumes messages; other ready Channel conditions consume nothing.
    pub fn any_of(conditions: impl IntoIterator<Item = Condition>) -> Self {
        Self {
            kind: WaitKind::AnyOf(conditions.into_iter().collect()),
        }
    }

    /// Waits for every Condition in any combination.
    ///
    /// Channel consumption is limited to Channel conditions in the selected combination.
    /// Conditions belonging only to other ready combinations consume nothing.
    ///
    /// Every Condition needs a non-empty user ID. Clones of the same Condition
    /// may be reused across combinations; distinct Conditions need unique IDs.
    pub fn any_combination_of(
        combinations: impl IntoIterator<Item = ConditionCombination>,
    ) -> Self {
        Self {
            kind: WaitKind::AnyCombinationOf(combinations.into_iter().collect()),
        }
    }
}

/// Represents one durable timer, Channel, or SubFlow predicate.
///
/// Conditions are created by [`crate::Timer`], [`crate::Channel`], [`crate::ChannelMap`], or
/// [`crate::Join`]. Assign an ID when identifying a timer independently or using
/// [`Wait::any_combination_of`].
#[derive(Clone)]
pub struct Condition {
    pub(crate) identity: Arc<ConditionIdentity>,
    pub(crate) id: Option<String>,
    pub(crate) kind: ConditionKind,
}

pub(crate) struct ConditionIdentity;

#[derive(Clone)]
pub(crate) enum ConditionKind {
    Timer(Duration),
    Channel {
        name: String,
        instance: Option<String>,
        at_least: Option<usize>,
        at_most: Option<usize>,
    },
    SubFlow(Box<SubFlowDefinition>),
    /// Fails Wait mapping with this reason; builders defer validation errors this way.
    Invalid(String),
}

#[derive(Clone)]
pub(crate) struct SubFlowDefinition {
    pub(crate) flow_type: &'static str,
    pub(crate) type_id: TypeId,
    pub(crate) input: ProtoValue,
    pub(crate) options: SubFlowOptions,
}

impl Condition {
    pub(crate) fn timer(duration: Duration) -> Self {
        Self {
            identity: Arc::new(ConditionIdentity),
            id: None,
            kind: ConditionKind::Timer(duration),
        }
    }

    pub(crate) fn channel(
        name: String,
        instance: Option<String>,
        at_least: Option<usize>,
        at_most: Option<usize>,
    ) -> Self {
        Self {
            identity: Arc::new(ConditionIdentity),
            id: None,
            kind: ConditionKind::Channel {
                name,
                instance,
                at_least,
                at_most,
            },
        }
    }

    pub(crate) fn sub_flow(definition: SubFlowDefinition) -> Self {
        let id = definition.options.condition_id.clone();
        Self {
            identity: Arc::new(ConditionIdentity),
            id,
            kind: ConditionKind::SubFlow(Box::new(definition)),
        }
    }

    pub(crate) fn invalid(reason: String) -> Self {
        Self {
            identity: Arc::new(ConditionIdentity),
            id: None,
            kind: ConditionKind::Invalid(reason),
        }
    }

    /// Assigns a stable ID used by [`crate::TimerId::by_condition_id`].
    pub fn with_id(mut self, id: impl Into<String>) -> Self {
        self.id = Some(id.into());
        self
    }
}

/// Groups conditions that must all succeed as one branch of [`Wait::any_combination_of`].
pub struct ConditionCombination {
    pub(crate) conditions: Vec<Condition>,
}

impl ConditionCombination {
    /// Creates a branch satisfied only when all `conditions` are satisfied.
    pub fn all_of(conditions: impl IntoIterator<Item = Condition>) -> Self {
        Self {
            conditions: conditions.into_iter().collect(),
        }
    }
}