feat: add Workflows support

Cargo.toml

@@ -72,7 +72,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"