feat: add Workflows support

Cargo.toml

@@ -74,7 +74,7 @@ web-sys = { version = "0.3.90", features = [
     "WritableStream",
     "WritableStreamDefaultWriter",
 ] }
-worker = { version = "0.7.5", path = "worker", features = ["queue", "d1", "axum", "timezone"] }
+worker = { version = "0.7.5", path = "worker", features = ["queue", "d1", "axum", "timezone", "workflow"] }
 worker-codegen = { path = "worker-codegen", version = "0.2.0" }
 worker-macros = { version = "0.7.5", path = "worker-macros", features = ["queue"] }
 worker-sys = { version = "0.7.5", path = "worker-sys", features = ["d1", "queue"] }

examples/workflow/Cargo.toml

@@ -0,0 +1,15 @@
+[package]
+name = "workflow-example"
+version = "0.1.0"
+edition = "2021"
+
+[package.metadata.release]
+release = false
+
+[lib]
+crate-type = ["cdylib"]
+
+[dependencies]
+serde = { version = "1", features = ["derive"] }
+serde_json = "1"
+worker = { path = "../../worker", features = ["workflow"] }

examples/workflow/src/lib.rs

@@ -0,0 +1,182 @@
+use serde::{Deserialize, Serialize};
+use worker::*;
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct MyParams {
+    pub email: String,
+    pub name: String,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct MyOutput {
+    pub message: String,
+    pub steps_completed: u32,
+}
+
+#[workflow]
+pub struct MyWorkflow {
+    #[allow(dead_code)]
+    env: Env,
+}
+
+impl WorkflowEntrypoint for MyWorkflow {
+    fn new(_ctx: Context, env: Env) -> Self {
+        Self { env }
+    }
+
+    async fn run(
+        &self,
+        event: WorkflowEvent<serde_json::Value>,
+        step: WorkflowStep,
+    ) -> Result<serde_json::Value> {
+        console_log!("Workflow started with instance ID: {}", event.instance_id);
+
+        let params: MyParams =
+            serde_json::from_value(event.payload).map_err(|e| Error::RustError(e.to_string()))?;
+
+        let email_for_validation = params.email.clone();
+        step.do_with_config(
+            "validate-params",
+            StepConfig {
+                retries: Some(RetryConfig {
+                    limit: 3,
+                    delay: "1 second".to_string(),
+                    backoff: None,
+                }),
+                timeout: None,
+            },
+            move || {
+                let email = email_for_validation.clone();
+                async move {
+                    if !email.contains('@') {
+                        return Err(NonRetryableError::new("invalid email address").into());
+                    }
+                    Ok(serde_json::json!({ "valid": true }))
+                }
+            },
+        )
+        .await?;
+
+        let name_for_step1 = params.name.clone();
+        let step1_result = step
+            .do_("initial-processing", move || {
+                let name = name_for_step1.clone();
+                async move {
+                    console_log!("Processing for user: {}", name);
+                    Ok(serde_json::json!({
+                        "processed": true,
+                        "user": name
+                    }))
+                }
+            })
+            .await?;
+
+        console_log!("Step 1 completed: {:?}", step1_result);
+
+        console_log!("Step 2: Sleeping for 10 seconds...");
+        step.sleep("wait-for-processing", "10 seconds").await?;
+
+        let email_for_step3 = params.email.clone();
+        let notification_result = step
+            .do_with_config(
+                "send-notification",
+                StepConfig {
+                    retries: Some(RetryConfig {
+                        limit: 3,
+                        delay: "5 seconds".to_string(),
+                        backoff: Some(Backoff::Exponential),
+                    }),
+                    timeout: Some("1 minute".to_string()),
+                },
+                move || {
+                    let email = email_for_step3.clone();
+                    async move {
+                        console_log!("Sending notification to: {}", email);
+                        if js_sys::Math::random() < 0.5 {
+                            return Err("notification service temporarily unavailable".into());
+                        }
+                        Ok(serde_json::json!({
+                            "notification_sent": true,
+                            "email": email
+                        }))
+                    }
+                },
+            )
+            .await?;
+
+        console_log!("Step 3 completed: {:?}", notification_result);
+
+        let output = MyOutput {
+            message: format!("Workflow completed for {}", params.name),
+            steps_completed: 3,
+        };
+
+        Ok(serde_json::to_value(output).unwrap())
+    }
+}
+
+#[event(fetch)]
+async fn fetch(mut req: Request, env: Env, _ctx: Context) -> Result<Response> {
+    let url = req.url()?;
+    let path = url.path();
+    let workflow = env.workflow("MY_WORKFLOW")?;
+
+    match (req.method(), path) {
+        (Method::Post, "/workflow") => {
+            let params: MyParams = req.json().await?;
+
+            let instance = workflow
+                .create(Some(CreateOptions {
+                    params: Some(params),
+                    ..Default::default()
+                }))
+                .await?;
+
+            Response::from_json(&serde_json::json!({
+                "id": instance.id()?,
+                "message": "Workflow created"
+            }))
+        }
+
+        (Method::Get, path) if path.starts_with("/workflow/") => {
+            let id = path.trim_start_matches("/workflow/");
+            let instance = workflow.get(id).await?;
+            let status = instance.status().await?;
+
+            Response::from_json(&serde_json::json!({
+                "id": instance.id()?,
+                "status": format!("{:?}", status.status),
+                "error": status.error,
+                "output": status.output
+            }))
+        }
+
+        (Method::Post, path) if path.starts_with("/workflow/") && path.ends_with("/pause") => {
+            let id = path
+                .trim_start_matches("/workflow/")
+                .trim_end_matches("/pause");
+            let instance = workflow.get(id).await?;
+            instance.pause().await?;
+
+            Response::from_json(&serde_json::json!({
+                "id": instance.id()?,
+                "message": "Workflow paused"
+            }))
+        }
+
+        (Method::Post, path) if path.starts_with("/workflow/") && path.ends_with("/resume") => {
+            let id = path
+                .trim_start_matches("/workflow/")
+                .trim_end_matches("/resume");
+            let instance = workflow.get(id).await?;
+            instance.resume().await?;
+
+            Response::from_json(&serde_json::json!({
+                "id": instance.id()?,
+                "message": "Workflow resumed"
+            }))
+        }
+
+        _ => Response::error("Not Found", 404),
+    }
+}

examples/workflow/wrangler.toml

@@ -0,0 +1,13 @@
+name = "workflow-example"
+main = "build/worker/shim.mjs"
+compatibility_date = "2024-10-22"
+
+[build]
+# For development: use local worker-build binary
+# For production: command = "cargo install -q worker-build && worker-build --release"
+command = "RUSTFLAGS='--cfg=web_sys_unstable_apis' ../../target/release/worker-build --release"
+
+[[workflows]]
+name = "my-workflow"
+binding = "MY_WORKFLOW"
+class_name = "MyWorkflow"

test/src/lib.rs

@@ -37,6 +37,7 @@ mod socket;
 mod sql_counter;
 mod sql_iterator;
 mod user;
+mod workflow;
 mod ws;
 
 #[derive(Deserialize, Serialize)]

test/src/router.rs

@@ -1,7 +1,7 @@
 use crate::{
     alarm, analytics_engine, assets, auto_response, cache, container, counter, d1, durable, fetch,
     form, js_snippets, kv, put_raw, queue, r2, rate_limit, request, secret_store, service, socket,
-    sql_counter, sql_iterator, user, ws, SomeSharedData, GLOBAL_STATE,
+    sql_counter, sql_iterator, user, workflow, ws, SomeSharedData, GLOBAL_STATE,
 };
 #[cfg(feature = "http")]
 use std::convert::TryInto;
@@ -17,6 +17,8 @@ use axum::{
     routing::{delete, get, head, options, patch, post, put},
     Extension,
 };
+#[cfg(feature = "http")]
+use worker::send::SendFuture;
 
 // Transform the argument into the correct form for the router.
 // For axum::Router:
@@ -59,16 +61,16 @@ macro_rules! format_route (
 #[cfg(feature = "http")]
 macro_rules! handler (
     ($name:path) => {
-        |Extension(env): Extension<Env>, Extension(data): Extension<SomeSharedData>, req: axum::extract::Request| async {
+        |Extension(env): Extension<Env>, Extension(data): Extension<SomeSharedData>, req: axum::extract::Request| SendFuture::new(async {
             let resp = $name(req.try_into().expect("convert request"), env, data).await.expect("handler result");
             Into::<http::Response<axum::body::Body>>::into(resp)
-        }
+        })
     };
     ($name:path, sync) => {
-        |Extension(env): Extension<Env>, Extension(data): Extension<SomeSharedData>, req: axum::extract::Request| async {
+        |Extension(env): Extension<Env>, Extension(data): Extension<SomeSharedData>, req: axum::extract::Request| SendFuture::new(async {
             let resp = $name(req.try_into().expect("convert request"), env, data).expect("handler result");
             Into::<http::Response<axum::body::Body>>::into(resp)
-        }
+        })
     };
 );
 #[cfg(not(feature = "http"))]
@@ -239,6 +241,18 @@ macro_rules! add_routes (
     add_route!($obj, get, format_route!("/rate-limit/key/{}", "key"), rate_limit::handle_rate_limit_with_key);
     add_route!($obj, get, "/rate-limit/bulk-test", rate_limit::handle_rate_limit_bulk_test);
     add_route!($obj, get, "/rate-limit/reset", rate_limit::handle_rate_limit_reset);
+    add_route!($obj, post, "/workflow/create", workflow::handle_workflow_create);
+    add_route!($obj, post, "/workflow/create-invalid", workflow::handle_workflow_create_invalid);
+    add_route!($obj, get, format_route!("/workflow/status/{}", "id"), workflow::handle_workflow_status);
+    add_route!($obj, post, "/workflow/event/create", workflow::handle_event_workflow_create);
+    add_route!($obj, post, format_route!("/workflow/event/send/{}", "id"), workflow::handle_event_workflow_send);
+    add_route!($obj, get, format_route!("/workflow/event/status/{}", "id"), workflow::handle_event_workflow_status);
+    add_route!($obj, post, "/workflow/lifecycle/create", workflow::handle_lifecycle_workflow_create);
+    add_route!($obj, get, format_route!("/workflow/lifecycle/status/{}", "id"), workflow::handle_lifecycle_workflow_status);
+    add_route!($obj, post, format_route!("/workflow/lifecycle/pause/{}", "id"), workflow::handle_lifecycle_workflow_pause);
+    add_route!($obj, post, format_route!("/workflow/lifecycle/resume/{}", "id"), workflow::handle_lifecycle_workflow_resume);
+    add_route!($obj, post, format_route!("/workflow/lifecycle/terminate/{}", "id"), workflow::handle_lifecycle_workflow_terminate);
+    add_route!($obj, post, format_route!("/workflow/lifecycle/restart/{}", "id"), workflow::handle_lifecycle_workflow_restart);
 });
 
 #[cfg(feature = "http")]