dex-sdk 0.2.2

// Copyright (c) 2026 Super Durable, Inc.
//
// Licensed under the Super Durable Source 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-Super-Durable-1.0

use std::time::Duration;

use std::sync::Arc;

use dex_protocol::dex::{AttributeSyncConfig, IndexConfig};

use crate::registry::physical_name;
use crate::step::{ErasedValue, TypedValue};
use crate::{Attribute, AttributeMap, FlowConfig, RetryPolicy, Value};

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
/// Controls whether [`crate::Client::start_flow_with_options`] may reuse a Flow ID.
pub enum IdReusePolicy {
    /// Uses the Dex server's configured reuse policy.
    Default,
    /// Allows reuse only when the previous run failed.
    AllowIfPreviousFailed,
    /// Allows reuse when no run with the ID is currently active.
    AllowIfNotRunning,
    /// Rejects reuse regardless of the previous run status.
    Disallow,
    /// Terminates an active run before starting the replacement.
    TerminateIfRunning,
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
/// Controls how Dex responds when a positive soft Flow timeout expires.
pub enum FlowTimeoutPolicy {
    /// Uses the Flow's registered timeout handler when present, and fails otherwise.
    Default,
    /// Fails with [`crate::FlowErrorType::FlowTimeout`] and permits Flow retries.
    Fail,
    /// Cancels without retrying the Flow.
    Cancel,
    /// Invokes the registered [`crate::FlowTimeoutHandler`] once.
    Handler,
}

#[derive(Clone)]
/// Configures a new Flow execution.
///
/// All optional fields are absent by default. Initial Attribute values are encoded when the start
/// request is sent. Use a stable request ID to make retries of the same logical request idempotent.
///
/// # Examples
///
/// ```
/// use dex_sdk::{Attribute, IdReusePolicy, StartFlowOptions};
/// use std::time::Duration;
///
/// let customer = Attribute::<String>::new("customer");
/// let options = StartFlowOptions::new()
///     .timeout(Duration::from_secs(3_600))
///     .id_reuse_policy(IdReusePolicy::AllowIfNotRunning)
///     .initial_attribute(&customer, "customer-42".to_owned())
///     .request_id("create-order-42");
/// ```
pub struct StartFlowOptions {
    pub(crate) timeout: Option<Duration>,
    pub(crate) timeout_policy: FlowTimeoutPolicy,
    pub(crate) start_delay: Option<Duration>,
    pub(crate) id_reuse_policy: IdReusePolicy,
    pub(crate) retry_policy: Option<RetryPolicy>,
    pub(crate) config_override: Option<FlowConfig>,
    pub(crate) ignore_already_started: bool,
    pub(crate) request_id: Option<String>,
    pub(crate) attributes: Vec<InitialAttribute>,
}

#[derive(Clone)]
pub(crate) struct InitialAttribute {
    pub(crate) key: String,
    pub(crate) value: Arc<dyn ErasedValue>,
    pub(crate) index_config: Option<IndexConfig>,
    pub(crate) sync_config: Option<AttributeSyncConfig>,
}

impl InitialAttribute {
    pub(crate) fn for_attribute<T: Value>(attribute: &Attribute<T>, value: T) -> Self {
        Self {
            key: attribute.name().to_string(),
            value: Arc::new(TypedValue(value)),
            index_config: attribute.index().map(|index| index.proto_config(false)),
            sync_config: attribute.sync_config(),
        }
    }

    pub(crate) fn for_attribute_map<T: Value>(
        attribute: &AttributeMap<T>,
        instance: &str,
        value: T,
    ) -> Self {
        Self {
            key: physical_name(attribute.name(), instance),
            value: Arc::new(TypedValue(value)),
            index_config: attribute.index().map(|index| index.proto_config(true)),
            sync_config: attribute.sync_config(),
        }
    }
}

impl StartFlowOptions {
    /// Creates start options that preserve all server defaults.
    pub fn new() -> Self {
        Self {
            timeout: None,
            timeout_policy: FlowTimeoutPolicy::Default,
            start_delay: None,
            id_reuse_policy: IdReusePolicy::Default,
            retry_policy: None,
            config_override: None,
            ignore_already_started: false,
            request_id: None,
            attributes: Vec::new(),
        }
    }

    /// Sets Dex's durable soft timeout. A zero duration disables it.
    pub fn timeout(mut self, value: Duration) -> Self {
        self.timeout = Some(value);
        self
    }

    /// Sets the action taken when the positive Flow timeout expires.
    ///
    /// Handler is rejected before start when the registered Flow has no timeout handler.
    pub fn timeout_policy(mut self, value: FlowTimeoutPolicy) -> Self {
        self.timeout_policy = value;
        self
    }

    /// Delays the first Step after the server accepts the Flow.
    pub fn start_delay(mut self, value: Duration) -> Self {
        self.start_delay = Some(value);
        self
    }

    /// Sets how the server handles an existing Flow with the same ID.
    pub fn id_reuse_policy(mut self, value: IdReusePolicy) -> Self {
        self.id_reuse_policy = value;
        self
    }

    /// Sets the whole-Flow retry policy after a terminal failure.
    pub fn retry_policy(mut self, value: RetryPolicy) -> Self {
        self.retry_policy = Some(value);
        self
    }

    /// Adds an initial value for a declared Attribute.
    pub fn initial_attribute<T: Value>(mut self, attribute: &Attribute<T>, value: T) -> Self {
        self.attributes
            .push(InitialAttribute::for_attribute(attribute, value));
        self
    }

    /// Adds an initial value for one Attribute-map instance.
    pub fn initial_attribute_map<T: Value>(
        mut self,
        attribute: &AttributeMap<T>,
        instance: &str,
        value: T,
    ) -> Self {
        self.attributes.push(InitialAttribute::for_attribute_map(
            attribute, instance, value,
        ));
        self
    }

    /// Overrides the registered Flow configuration for this execution.
    pub fn config_override(mut self, value: FlowConfig) -> Self {
        self.config_override = Some(value);
        self
    }

    /// Returns the existing run ID instead of an error when the Flow already exists.
    pub fn ignore_already_started(mut self, value: bool) -> Self {
        self.ignore_already_started = value;
        self
    }

    /// Sets the idempotency key for this start request.
    pub fn request_id(mut self, value: impl Into<String>) -> Self {
        self.request_id = Some(value.into());
        self
    }
}

impl Default for StartFlowOptions {
    fn default() -> Self {
        Self::new()
    }
}

#[cfg(test)]
mod tests {
    use super::StartFlowOptions;
    use crate::{Attribute, AttributeMap};

    #[test]
    fn initial_attributes_retain_sync_configuration() {
        let plain = Attribute::<String>::new("plain");
        let synced = Attribute::<String>::new("synced").sync_to_attribute_store();
        let synced_map = AttributeMap::<String>::new("map").sync_to_attribute_store();
        let options = StartFlowOptions::new()
            .initial_attribute(&plain, "plain".to_string())
            .initial_attribute(&synced, "synced".to_string())
            .initial_attribute_map(&synced_map, "tenant-1", "mapped".to_string());

        assert!(options.attributes[0].sync_config.is_none());
        assert_eq!(
            options.attributes[1]
                .sync_config
                .as_ref()
                .map(|config| config.enabled),
            Some(true)
        );
        assert_eq!(
            options.attributes[2]
                .sync_config
                .as_ref()
                .map(|config| config.enabled),
            Some(true)
        );
    }
}