Struct WorkflowContext 

pub struct WorkflowContext {
    pub run_id: Uuid,
    pub workflow_name: String,
    pub version: u32,
    pub started_at: DateTime<Utc>,
    pub auth: AuthContext,
    /* private fields */
}

Context available to workflow handlers.

Fields

run_id: Uuid

Workflow run ID.

workflow_name: String

Workflow name.

version: u32

Workflow version.

started_at: DateTime<Utc>

When the workflow started.

auth: AuthContext

Authentication context.

Implementations

impl WorkflowContext

pub fn new( run_id: Uuid, workflow_name: String, version: u32, db_pool: PgPool, http_client: Client, ) -> Self

Create a new workflow context.

pub fn resumed( run_id: Uuid, workflow_name: String, version: u32, started_at: DateTime<Utc>, db_pool: PgPool, http_client: Client, ) -> Self

Create a resumed workflow context.

pub fn with_env_provider(self, provider: Arc<dyn EnvProvider>) -> Self

Set environment provider.

pub fn with_resumed_from_sleep(self) -> Self

Mark that this context resumed from a sleep (timer expired).

pub fn with_suspend_channel(self, tx: Sender<SuspendReason>) -> Self

Set the suspend channel.

pub fn with_tenant(self, tenant_id: Uuid) -> Self

Set the tenant ID.

pub fn tenant_id(&self) -> Option<Uuid>

pub fn is_resumed(&self) -> bool

pub fn workflow_time(&self) -> DateTime<Utc>

pub fn db(&self) -> &PgPool

pub fn http(&self) -> &Client

pub fn with_auth(self, auth: AuthContext) -> Self

Set authentication context.

pub fn with_step_states(self, states: HashMap<String, StepState>) -> Self

Restore step states from persisted data.

pub fn get_step_state(&self, name: &str) -> Option<StepState>

pub fn is_step_completed(&self, name: &str) -> bool

pub fn is_step_started(&self, name: &str) -> bool

Check if a step has been started (running, completed, or failed).

Use this to guard steps that should only execute once, even across workflow suspension and resumption.

pub fn get_step_result<T: DeserializeOwned>(&self, name: &str) -> Option<T>

pub fn record_step_start(&self, name: &str)

Record step start.

If the step is already running or beyond (completed/failed), this is a no-op. This prevents race conditions when resuming workflows.

pub fn record_step_complete(&self, name: &str, result: Value)

Record step completion (fire-and-forget database update). Use record_step_complete_async if you need to ensure persistence before continuing.

pub async fn record_step_complete_async(&self, name: &str, result: Value)

Record step completion and wait for database persistence.

pub fn record_step_failure(&self, name: &str, error: impl Into<String>)

Record step failure.

pub fn record_step_compensated(&self, name: &str)

Record step compensation.

pub fn completed_steps_reversed(&self) -> Vec<String>

pub fn all_step_states(&self) -> HashMap<String, StepState>

pub fn elapsed(&self) -> Duration

pub fn register_compensation( &self, step_name: &str, handler: CompensationHandler, )

Register a compensation handler for a step.

pub fn get_compensation_handler( &self, step_name: &str, ) -> Option<CompensationHandler>

pub fn has_compensation(&self, step_name: &str) -> bool

pub async fn run_compensation(&self) -> Vec<(String, bool)>

Run compensation for all completed steps in reverse order. Returns a list of (step_name, success) tuples.

pub fn compensation_handlers(&self) -> HashMap<String, CompensationHandler>

pub async fn sleep(&self, duration: Duration) -> Result<()>

Sleep for a duration.

This suspends the workflow and persists the wake time to the database. The workflow scheduler will resume the workflow when the time arrives.

Example

// Sleep for 30 days
ctx.sleep(Duration::from_secs(30 * 24 * 60 * 60)).await?;

pub async fn sleep_until(&self, wake_at: DateTime<Utc>) -> Result<()>

Sleep until a specific time.

If the wake time has already passed, returns immediately.

Example

use chrono::{Utc, Duration};
let renewal_date = Utc::now() + Duration::days(30);
ctx.sleep_until(renewal_date).await?;

pub async fn wait_for_event<T: DeserializeOwned>( &self, event_name: &str, timeout: Option<Duration>, ) -> Result<T>

Wait for an external event with optional timeout.

The workflow suspends until the event arrives or the timeout expires. Events are correlated by the workflow run ID.

Example

let payment: PaymentConfirmation = ctx.wait_for_event(
    "payment_confirmed",
    Some(Duration::from_secs(7 * 24 * 60 * 60)), // 7 days
).await?;

pub fn parallel(&self) -> ParallelBuilder<'_>

Create a parallel builder for executing steps concurrently.

Example

let results = ctx.parallel()
    .step("fetch_user", || async { get_user(id).await })
    .step("fetch_orders", || async { get_orders(id).await })
    .step_with_compensate("charge_card",
        || async { charge_card(amount).await },
        |charge| async move { refund(charge.id).await })
    .run().await?;

let user: User = results.get("fetch_user")?;
let orders: Vec<Order> = results.get("fetch_orders")?;

pub fn step<T, F, Fut>( &self, name: impl Into<String>, f: F, ) -> StepRunner<'_, T>

where T: Serialize + DeserializeOwned + Clone + Send + Sync + 'static, F: Fn() -> Fut + Send + Sync + 'static, Fut: Future<Output = Result<T>> + Send + 'static,

Create a step runner for executing a workflow step.

This provides a fluent API for defining steps with retry, compensation, timeout, and optional behavior.

Examples

use std::time::Duration;

// Simple step
let data = ctx.step("fetch_data", || async {
    Ok(fetch_from_api().await?)
}).run().await?;

// Step with retry (3 attempts, 2 second delay)
ctx.step("send_email", || async {
    send_verification_email(&user.email).await
})
.retry(3, Duration::from_secs(2))
.run()
.await?;

// Step with compensation (rollback on later failure)
let charge = ctx.step("charge_card", || async {
    charge_credit_card(&card).await
})
.compensate(|charge_result| async move {
    refund_charge(&charge_result.charge_id).await
})
.run()
.await?;

// Optional step (failure won't trigger compensation)
ctx.step("notify_slack", || async {
    post_to_slack("User signed up!").await
})
.optional()
.run()
.await?;

// Step with timeout
ctx.step("slow_operation", || async {
    process_large_file().await
})
.timeout(Duration::from_secs(60))
.run()
.await?;

Trait Implementations

impl EnvAccess for WorkflowContext

fn env_provider(&self) -> &dyn EnvProvider

Get the environment provider.

fn env(&self, key: &str) -> Option<String>

Get an environment variable.

fn env_or(&self, key: &str, default: &str) -> String

Get an environment variable with a default value.

fn env_require(&self, key: &str) -> Result<String>

Get a required environment variable.

fn env_parse<T: FromStr>(&self, key: &str) -> Result<T>

where T::Err: Display,

Get an environment variable and parse it to the specified type.

fn env_parse_or<T: FromStr>(&self, key: &str, default: T) -> Result<T>

where T::Err: Display,

Get an environment variable and parse it, with a default.

fn env_contains(&self, key: &str) -> bool

Check if an environment variable is set.