temporalio-sdk 0.2.0

//! Functionality related to defining and interacting with workflows
//!
//! This module contains traits and types for implementing workflows using the
//! `#[workflow]` and `#[workflow_methods]` macros.
//!
//! Example usage:
//! ```
//! use temporalio_macros::{workflow, workflow_methods};
//! use temporalio_sdk::{
//!     SyncWorkflowContext, WorkflowContext, WorkflowContextView, WorkflowResult,
//! };
//!
//! #[workflow]
//! pub struct MyWorkflow {
//!     counter: u32,
//! }
//!
//! #[workflow_methods]
//! impl MyWorkflow {
//!     #[init]
//!     pub fn new(ctx: &WorkflowContextView, input: String) -> Self {
//!         Self { counter: 0 }
//!     }
//!
//!     // Async run method uses ctx.state() for reading
//!     #[run]
//!     pub async fn run(ctx: &mut WorkflowContext<Self>) -> WorkflowResult<String> {
//!         let counter = ctx.state(|s| s.counter);
//!         Ok(format!("Done with counter: {}", counter))
//!     }
//!
//!     // Sync signals use &mut self for direct mutations
//!     #[signal]
//!     pub fn increment(&mut self, ctx: &mut SyncWorkflowContext<Self>, amount: u32) {
//!         self.counter += amount;
//!     }
//!
//!     // Queries use &self with read-only context
//!     #[query]
//!     pub fn get_counter(&self, ctx: &WorkflowContextView) -> u32 {
//!         self.counter
//!     }
//! }
//! ```

/// Deterministic `select!` for use in Temporal workflows.
///
/// Polls branches in declaration order (top to bottom), ensuring deterministic
/// behavior across workflow replays. Delegates to [`futures_util::select_biased!`].
///
/// All workflow futures (timers, activities, child workflows, etc.) implement
/// `FusedFuture`, so they can be stored in variables and passed to `select!`
/// without needing `.fuse()`.
///
/// # Example
///
/// ```ignore
/// use temporalio_sdk::workflows::select;
/// use temporalio_sdk::WorkflowContext;
/// use std::time::Duration;
///
/// # async fn hidden(ctx: &mut WorkflowContext<()>) {
/// select! {
///     _ = ctx.timer(Duration::from_secs(60)) => { /* timer fired */ }
///     reason = ctx.cancelled() => { /* cancelled */ }
/// };
/// # }
/// ```
#[doc(inline)]
pub use crate::__temporal_select as select;

/// Deterministic `join!` for use in Temporal workflows.
///
/// Polls all futures concurrently to completion in declaration order,
/// ensuring deterministic behavior across workflow replays. Delegates
/// to [`futures_util::join!`].
///
/// # Example
///
/// ```ignore
/// use temporalio_sdk::workflows::join;
///
/// # async fn hidden() {
/// let future_a = async { 1 };
/// let future_b = async { 2 };
/// let (a, b) = join!(future_a, future_b);
/// # }
/// ```
#[doc(inline)]
pub use crate::__temporal_join as join;

/// Deterministic `join_all` for use in Temporal workflows.
///
/// Polls a collection of futures concurrently to completion in declaration order,
/// returning a `Vec` of their results. Delegates to [`futures_util::future::join_all`].
///
/// # Example
///
/// ```ignore
/// use temporalio_sdk::workflows::join_all;
/// use temporalio_sdk::WorkflowContext;
/// use std::time::Duration;
///
/// # async fn hidden(ctx: &mut WorkflowContext<()>) {
/// let timers = vec![
///     ctx.timer(Duration::from_secs(1)),
///     ctx.timer(Duration::from_secs(2)),
/// ];
/// let results = join_all(timers).await;
/// # }
/// ```
pub use futures_util::future::join_all;

use crate::{
    BaseWorkflowContext, SyncWorkflowContext, WorkflowContext, WorkflowContextView,
    WorkflowTermination,
};
use futures_util::future::{Fuse, FutureExt, LocalBoxFuture};
use std::{
    cell::RefCell,
    collections::HashMap,
    fmt::Debug,
    pin::Pin,
    rc::Rc,
    sync::Arc,
    task::{Context as TaskContext, Poll},
};
use temporalio_common::{
    QueryDefinition, SignalDefinition, UpdateDefinition, WorkflowDefinition,
    data_converters::{
        GenericPayloadConverter, PayloadConversionError, PayloadConverter, SerializationContext,
        SerializationContextData, TemporalDeserializable, TemporalSerializable,
    },
    protos::temporal::api::{
        common::v1::{Payload, Payloads},
        failure::v1::Failure,
    },
};

/// Error type for workflow operations
#[derive(Debug, thiserror::Error)]
pub enum WorkflowError {
    /// Error during payload conversion
    #[error("Payload conversion error: {0}")]
    PayloadConversion(#[from] PayloadConversionError),

    /// Workflow execution error
    #[error("Workflow execution error: {0}")]
    Execution(#[from] anyhow::Error),
}

impl From<WorkflowError> for Failure {
    fn from(err: WorkflowError) -> Self {
        Failure {
            message: err.to_string(),
            ..Default::default()
        }
    }
}

/// Trait implemented by workflow structs to enable execution by the worker.
///
/// This trait is typically generated by the `#[workflow_methods]` macro and should not
/// be implemented manually in most cases.
#[doc(hidden)]
pub trait WorkflowImplementation: Sized + 'static {
    /// The marker struct for the run method that implements `WorkflowDefinition`
    type Run: WorkflowDefinition;

    /// Whether this workflow has a user-defined `#[init]` method.
    /// Set to `true` by the macro when `#[init]` is present, `false` otherwise.
    const HAS_INIT: bool;

    /// Whether the init method accepts the workflow input.
    /// If true, input goes to init. If false, input goes to run.
    const INIT_TAKES_INPUT: bool;

    /// Returns the workflow type name.
    fn name() -> &'static str;

    /// Initialize the workflow instance.
    ///
    /// This is called when a new workflow execution starts. If `INIT_TAKES_INPUT` is true,
    /// `input` will be `Some`. Otherwise it's `None`.
    fn init(
        ctx: WorkflowContextView,
        input: Option<<Self::Run as WorkflowDefinition>::Input>,
    ) -> Self;

    /// Execute the workflow's main run function.
    ///
    /// If `INIT_TAKES_INPUT` is false, `input` will be `Some`. Otherwise it's `None`.
    fn run(
        ctx: WorkflowContext<Self>,
        input: Option<<Self::Run as WorkflowDefinition>::Input>,
    ) -> LocalBoxFuture<'static, Result<Payload, WorkflowTermination>>;

    /// Dispatch an update request by name. Returns `None` if no handler for that name.
    fn dispatch_update(
        _ctx: WorkflowContext<Self>,
        _name: &str,
        _payloads: Payloads,
        _converter: &PayloadConverter,
    ) -> Option<LocalBoxFuture<'static, Result<Payload, WorkflowError>>> {
        None
    }

    /// Validate an update request by name.
    ///
    /// Returns `None` if no handler for that name, `Some(Ok(()))` if valid,
    /// `Some(Err(...))` if validation failed.
    fn validate_update(
        &self,
        _ctx: WorkflowContextView,
        _name: &str,
        _payloads: &Payloads,
        _converter: &PayloadConverter,
    ) -> Option<Result<(), WorkflowError>> {
        None
    }

    /// Dispatch a signal by name.
    ///
    /// Returns `None` if no handler for that name. For sync signals, the mutation happens
    /// immediately and returns a completed future. For async signals, returns a future
    /// that must be polled to completion.
    fn dispatch_signal(
        _ctx: WorkflowContext<Self>,
        _name: &str,
        _payloads: Payloads,
        _converter: &PayloadConverter,
    ) -> Option<LocalBoxFuture<'static, Result<(), WorkflowError>>> {
        None
    }

    /// Dispatch a query by name.
    ///
    /// Returns `None` if no handler for that name, `Some(Ok(payload))` on success,
    /// `Some(Err(...))` on failure. Queries are synchronous and read-only.
    fn dispatch_query(
        &self,
        _ctx: WorkflowContextView,
        _name: &str,
        _payloads: &Payloads,
        _converter: &PayloadConverter,
    ) -> Option<Result<Payload, WorkflowError>> {
        None
    }
}

// NOTE: In the below traits, the dispatch functions take context by ownership while the handle
// methods take them by ref when sync and by ownership when async. They must be owned by async
// handlers since the returned futures must be 'static.

/// Trait for executing synchronous signal handlers on a workflow.
#[doc(hidden)]
pub trait ExecutableSyncSignal<S: SignalDefinition>: WorkflowImplementation {
    /// Handle an incoming signal with the given input.
    fn handle(&mut self, ctx: &mut SyncWorkflowContext<Self>, input: S::Input);

    /// Dispatch the signal with payload deserialization.
    fn dispatch(
        ctx: WorkflowContext<Self>,
        payloads: Payloads,
        converter: &PayloadConverter,
    ) -> LocalBoxFuture<'static, Result<(), WorkflowError>> {
        match deserialize_input::<S::Input>(payloads.payloads, converter) {
            Ok(input) => {
                let mut sync_ctx = ctx.sync_context();
                ctx.state_mut(|wf| Self::handle(wf, &mut sync_ctx, input));
                std::future::ready(Ok(())).boxed_local()
            }
            Err(e) => std::future::ready(Err(e)).boxed_local(),
        }
    }
}

/// Trait for executing asynchronous signal handlers on a workflow.
#[doc(hidden)]
pub trait ExecutableAsyncSignal<S: SignalDefinition>: WorkflowImplementation {
    /// Handle an incoming signal with the given input.
    fn handle(ctx: WorkflowContext<Self>, input: S::Input) -> LocalBoxFuture<'static, ()>;

    /// Dispatch the signal with payload deserialization.
    fn dispatch(
        ctx: WorkflowContext<Self>,
        payloads: Payloads,
        converter: &PayloadConverter,
    ) -> LocalBoxFuture<'static, Result<(), WorkflowError>> {
        match deserialize_input::<S::Input>(payloads.payloads, converter) {
            Ok(input) => Self::handle(ctx, input).map(|()| Ok(())).boxed_local(),
            Err(e) => std::future::ready(Err(e)).boxed_local(),
        }
    }
}

/// Trait for executing query handlers on a workflow.
///
/// Queries are read-only operations that do not mutate workflow state.
/// They must be synchronous.
#[doc(hidden)]
pub trait ExecutableQuery<Q: QueryDefinition>: WorkflowImplementation {
    /// Handle a query with the given input and return the result.
    ///
    /// Queries take `&self` (immutable) and cannot modify workflow state.
    /// Returning an error will cause the query to fail with that error message.
    fn handle(
        &self,
        ctx: &WorkflowContextView,
        input: Q::Input,
    ) -> Result<Q::Output, Box<dyn std::error::Error + Send + Sync>>;

    /// Dispatch the query with payload deserialization and output serialization.
    fn dispatch(
        &self,
        ctx: &WorkflowContextView,
        payloads: &Payloads,
        converter: &PayloadConverter,
    ) -> Result<Payload, WorkflowError> {
        let input = deserialize_input::<Q::Input>(payloads.payloads.clone(), converter)?;
        let output = self.handle(ctx, input).map_err(wrap_handler_error)?;
        serialize_output(&output, converter)
    }
}

/// Trait for executing synchronous update handlers on a workflow.
#[doc(hidden)]
pub trait ExecutableSyncUpdate<U: UpdateDefinition>: WorkflowImplementation {
    /// Handle an update with the given input and return the result.
    /// Returning an error will cause the update to fail with that error message.
    fn handle(
        &mut self,
        ctx: &mut SyncWorkflowContext<Self>,
        input: U::Input,
    ) -> Result<U::Output, Box<dyn std::error::Error + Send + Sync>>;

    /// Validate an update before it is applied.
    fn validate(
        &self,
        _ctx: &WorkflowContextView,
        _input: &U::Input,
    ) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
        Ok(())
    }

    /// Dispatch the update with payload deserialization and output serialization.
    fn dispatch(
        ctx: WorkflowContext<Self>,
        payloads: Payloads,
        converter: &PayloadConverter,
    ) -> LocalBoxFuture<'static, Result<Payload, WorkflowError>> {
        let input = match deserialize_input::<U::Input>(payloads.payloads, converter) {
            Ok(v) => v,
            Err(e) => return std::future::ready(Err(e)).boxed_local(),
        };
        let converter = converter.clone();
        let mut sync_ctx = ctx.sync_context();
        let result = ctx.state_mut(|wf| Self::handle(wf, &mut sync_ctx, input));
        match result {
            Ok(output) => match serialize_output(&output, &converter) {
                Ok(payload) => std::future::ready(Ok(payload)).boxed_local(),
                Err(e) => std::future::ready(Err(e)).boxed_local(),
            },
            Err(e) => std::future::ready(Err(wrap_handler_error(e))).boxed_local(),
        }
    }

    /// Dispatch validation with payload deserialization.
    fn dispatch_validate(
        &self,
        ctx: &WorkflowContextView,
        payloads: &Payloads,
        converter: &PayloadConverter,
    ) -> Result<(), WorkflowError> {
        let input = deserialize_input::<U::Input>(payloads.payloads.clone(), converter)?;
        self.validate(ctx, &input).map_err(wrap_handler_error)
    }
}

/// Trait for executing asynchronous update handlers on a workflow.
#[doc(hidden)]
pub trait ExecutableAsyncUpdate<U: UpdateDefinition>: WorkflowImplementation {
    /// Handle an update with the given input and return the result.
    /// Returning an error will cause the update to fail with that error message.
    fn handle(
        ctx: WorkflowContext<Self>,
        input: U::Input,
    ) -> LocalBoxFuture<'static, Result<U::Output, Box<dyn std::error::Error + Send + Sync>>>;

    /// Validate an update before it is applied.
    fn validate(
        &self,
        _ctx: &WorkflowContextView,
        _input: &U::Input,
    ) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
        Ok(())
    }

    /// Dispatch the update with payload deserialization and output serialization.
    fn dispatch(
        ctx: WorkflowContext<Self>,
        payloads: Payloads,
        converter: &PayloadConverter,
    ) -> LocalBoxFuture<'static, Result<Payload, WorkflowError>> {
        let input = match deserialize_input::<U::Input>(payloads.payloads, converter) {
            Ok(v) => v,
            Err(e) => return std::future::ready(Err(e)).boxed_local(),
        };
        let converter = converter.clone();
        async move {
            let output = Self::handle(ctx, input).await.map_err(wrap_handler_error)?;
            serialize_output(&output, &converter)
        }
        .boxed_local()
    }

    /// Dispatch validation with payload deserialization.
    fn dispatch_validate(
        &self,
        ctx: &WorkflowContextView,
        payloads: &Payloads,
        converter: &PayloadConverter,
    ) -> Result<(), WorkflowError> {
        let input = deserialize_input::<U::Input>(payloads.payloads.clone(), converter)?;
        self.validate(ctx, &input).map_err(wrap_handler_error)
    }
}

/// Data passed to handler dispatch methods (signals, updates, queries).
pub(crate) struct DispatchData<'a> {
    pub(crate) payloads: Payloads,
    pub(crate) headers: HashMap<String, Payload>,
    pub(crate) converter: &'a PayloadConverter,
}

/// Trait implemented by workflow types to enable registration with workers.
///
/// This trait is automatically generated by the `#[workflow_methods]` macro.
#[doc(hidden)]
pub trait WorkflowImplementer: WorkflowImplementation {
    /// Register this workflow and all its handlers with the given definitions container.
    fn register_all(defs: &mut WorkflowDefinitions);
}

/// Type-erased trait for workflow execution instances.
pub(crate) trait DynWorkflowExecution {
    /// Poll the run future.
    fn poll_run(&mut self, cx: &mut TaskContext<'_>) -> Poll<Result<Payload, WorkflowTermination>>;

    /// Validate an update request. Returns `None` if no handler.
    fn validate_update(&self, name: &str, data: &DispatchData)
    -> Option<Result<(), WorkflowError>>;

    /// Start an update handler. Returns `None` if no handler for that name.
    fn start_update(
        &mut self,
        name: &str,
        data: DispatchData,
    ) -> Option<LocalBoxFuture<'static, Result<Payload, WorkflowError>>>;

    /// Dispatch a signal by name. Returns `None` if no handler.
    fn dispatch_signal(
        &mut self,
        name: &str,
        data: DispatchData,
    ) -> Option<LocalBoxFuture<'static, Result<(), WorkflowError>>>;

    /// Dispatch a query by name. Returns `None` if no handler.
    fn dispatch_query(
        &self,
        name: &str,
        data: DispatchData,
    ) -> Option<Result<Payload, WorkflowError>>;
}

/// Manages a workflow execution, holding the context and run future.
pub(crate) struct WorkflowExecution<W: WorkflowImplementation> {
    ctx: WorkflowContext<W>,
    run_future: Fuse<LocalBoxFuture<'static, Result<Payload, WorkflowTermination>>>,
}

impl<W: WorkflowImplementation> WorkflowExecution<W>
where
    <W::Run as WorkflowDefinition>::Input: Send,
{
    /// Create a new workflow execution using the workflow's `init` method.
    pub(crate) fn new(
        base_ctx: BaseWorkflowContext,
        init_input: Option<<W::Run as WorkflowDefinition>::Input>,
        run_input: Option<<W::Run as WorkflowDefinition>::Input>,
    ) -> Self {
        let view = base_ctx.view();
        let workflow = W::init(view, init_input);
        Self::new_with_workflow(workflow, base_ctx, run_input)
    }

    /// Create a new workflow execution from an already-created workflow instance.
    pub(crate) fn new_with_workflow(
        workflow: W,
        base_ctx: BaseWorkflowContext,
        run_input: Option<<W::Run as WorkflowDefinition>::Input>,
    ) -> Self {
        let workflow = Rc::new(RefCell::new(workflow));
        let ctx = WorkflowContext::from_base(base_ctx, workflow);
        let run_future = W::run(ctx.clone(), run_input).fuse();

        Self { ctx, run_future }
    }
}

impl<W: WorkflowImplementation> DynWorkflowExecution for WorkflowExecution<W> {
    fn poll_run(&mut self, cx: &mut TaskContext<'_>) -> Poll<Result<Payload, WorkflowTermination>> {
        Pin::new(&mut self.run_future).poll(cx)
    }

    fn validate_update(
        &self,
        name: &str,
        data: &DispatchData,
    ) -> Option<Result<(), WorkflowError>> {
        let view = self.ctx.view();
        self.ctx
            .state(|wf| wf.validate_update(view, name, &data.payloads, data.converter))
    }

    fn start_update(
        &mut self,
        name: &str,
        data: DispatchData,
    ) -> Option<LocalBoxFuture<'static, Result<Payload, WorkflowError>>> {
        let ctx = self.ctx.with_headers(data.headers);
        W::dispatch_update(ctx, name, data.payloads, data.converter)
    }

    fn dispatch_signal(
        &mut self,
        name: &str,
        data: DispatchData,
    ) -> Option<LocalBoxFuture<'static, Result<(), WorkflowError>>> {
        let ctx = self.ctx.with_headers(data.headers);
        W::dispatch_signal(ctx, name, data.payloads, data.converter)
    }

    fn dispatch_query(
        &self,
        name: &str,
        data: DispatchData,
    ) -> Option<Result<Payload, WorkflowError>> {
        let view = self.ctx.view();
        self.ctx
            .state(|wf| wf.dispatch_query(view, name, &data.payloads, data.converter))
    }
}

/// Type alias for workflow execution factory functions.
///
/// Creates a new `WorkflowExecution` instance from the input payloads and context.
pub(crate) type WorkflowExecutionFactory = Arc<
    dyn Fn(
            Vec<Payload>,
            PayloadConverter,
            BaseWorkflowContext,
        ) -> Result<Box<dyn DynWorkflowExecution>, PayloadConversionError>
        + Send
        + Sync,
>;

/// Contains workflow registrations in a form ready for execution by workers.
#[derive(Default, Clone)]
pub struct WorkflowDefinitions {
    /// Maps workflow type name to execution factories
    workflows: HashMap<&'static str, WorkflowExecutionFactory>,
}

impl WorkflowDefinitions {
    /// Creates a new empty `WorkflowDefinitions`.
    pub fn new() -> Self {
        Self::default()
    }

    /// Register a workflow implementation.
    pub fn register_workflow<W: WorkflowImplementer>(&mut self) -> &mut Self {
        W::register_all(self);
        self
    }

    /// Register a specific workflow's run method.
    #[doc(hidden)]
    pub fn register_workflow_run<W: WorkflowImplementation>(&mut self) -> &mut Self
    where
        <W::Run as WorkflowDefinition>::Input: Send,
    {
        let workflow_name = W::name();
        let factory: WorkflowExecutionFactory =
            Arc::new(move |payloads, converter: PayloadConverter, base_ctx| {
                let ser_ctx = SerializationContext {
                    data: &SerializationContextData::Workflow,
                    converter: &converter,
                };
                let input = converter.from_payloads(&ser_ctx, payloads)?;
                let (init_input, run_input) = if W::INIT_TAKES_INPUT {
                    (Some(input), None)
                } else {
                    (None, Some(input))
                };
                Ok(
                    Box::new(WorkflowExecution::<W>::new(base_ctx, init_input, run_input))
                        as Box<dyn DynWorkflowExecution>,
                )
            });
        self.workflows.insert(workflow_name, factory);
        self
    }

    /// Register a workflow with a custom factory for instance creation.
    pub fn register_workflow_run_with_factory<W, F>(&mut self, user_factory: F) -> &mut Self
    where
        W: WorkflowImplementation,
        <W::Run as WorkflowDefinition>::Input: Send,
        F: Fn() -> W + Send + Sync + 'static,
    {
        assert!(
            !W::HAS_INIT,
            "Workflows registered with a factory must not define an #[init] method. \
             The factory replaces init for instance creation."
        );

        let workflow_name = W::name();
        let user_factory = Arc::new(user_factory);
        let factory: WorkflowExecutionFactory =
            Arc::new(move |payloads, converter: PayloadConverter, base_ctx| {
                let ser_ctx = SerializationContext {
                    data: &SerializationContextData::Workflow,
                    converter: &converter,
                };
                let input: <W::Run as WorkflowDefinition>::Input =
                    converter.from_payloads(&ser_ctx, payloads)?;

                // User factory creates the instance - input always goes to run()
                let workflow = user_factory();
                Ok(Box::new(WorkflowExecution::<W>::new_with_workflow(
                    workflow,
                    base_ctx,
                    Some(input),
                )) as Box<dyn DynWorkflowExecution>)
            });

        self.workflows.insert(workflow_name, factory);
        self
    }

    /// Check if any workflows are registered.
    pub fn is_empty(&self) -> bool {
        self.workflows.is_empty()
    }

    /// Get the workflow execution factory for a given workflow type.
    pub(crate) fn get_workflow(&self, workflow_type: &str) -> Option<WorkflowExecutionFactory> {
        self.workflows.get(workflow_type).cloned()
    }

    /// Returns an iterator over registered workflow type names.
    pub fn workflow_types(&self) -> impl Iterator<Item = &'static str> + '_ {
        self.workflows.keys().copied()
    }
}

impl Debug for WorkflowDefinitions {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("WorkflowDefinitions")
            .field("workflows", &self.workflows.keys().collect::<Vec<_>>())
            .finish()
    }
}

/// Deserialize handler input from payloads.
pub fn deserialize_input<I: TemporalDeserializable + 'static>(
    payloads: Vec<Payload>,
    converter: &PayloadConverter,
) -> Result<I, WorkflowError> {
    let ctx = SerializationContext {
        data: &SerializationContextData::Workflow,
        converter,
    };
    converter.from_payloads(&ctx, payloads).map_err(Into::into)
}

/// Serialize handler output to a payload.
pub fn serialize_output<O: TemporalSerializable + 'static>(
    output: &O,
    converter: &PayloadConverter,
) -> Result<Payload, WorkflowError> {
    let ctx = SerializationContext {
        data: &SerializationContextData::Workflow,
        converter,
    };
    converter.to_payload(&ctx, output).map_err(Into::into)
}

/// Wrap a handler error into WorkflowError.
pub fn wrap_handler_error(e: Box<dyn std::error::Error + Send + Sync>) -> WorkflowError {
    WorkflowError::Execution(anyhow::anyhow!(e))
}

/// Serialize a workflow result value to a payload.
pub fn serialize_result<T: TemporalSerializable + 'static>(
    result: T,
    converter: &PayloadConverter,
) -> Result<Payload, WorkflowError> {
    serialize_output(&result, converter)
}