nestrs_workflows/
lib.rs

1//! NestJS [`@nestjs/workflows`](https://docs.nestjs.com/fundamentals/workflows) analogue —
2//! durable multi-step workflows with resume-after-restart and human-approval waits.
3//!
4//! Persist steps via [`WorkflowsService::complete_step`]; park with
5//! [`WorkflowsService::wait_for_approval`] and continue with [`WorkflowsService::approve`].
6//! Combine with `nestrs-locks` for single-runner steppers and `nestrs-outbox` for
7//! side effects.
8//!
9//! Umbrella feature: `nestrs` / `workflows` → `nestrs::workflows`.
10#![doc(html_root_url = "https://docs.rs/nestrs-workflows/1.7.1")]
11
12mod module;
13mod options;
14mod store;
15
16pub use module::WorkflowsModule;
17pub use options::WorkflowsOptions;
18pub use store::{
19    InMemoryWorkflowStore, SharedWorkflowStore, WorkflowInstance, WorkflowStatus, WorkflowStore,
20};
21
22pub use nestrs_core::{DynamicModule, Injectable, ProviderRegistry};
23use serde_json::Value;
24use std::sync::Arc;
25
26pub struct WorkflowsService {
27    store: SharedWorkflowStore,
28}
29
30impl WorkflowsService {
31    pub fn from_options(opts: &WorkflowsOptions) -> Self {
32        Self {
33            store: opts.store.clone(),
34        }
35    }
36
37    pub async fn start(&self, name: &str, input: Value) -> WorkflowInstance {
38        self.store.create(name, input).await
39    }
40
41    pub async fn get(&self, id: &str) -> Option<WorkflowInstance> {
42        self.store.get(id).await
43    }
44
45    /// Advance to the next step, persisting result. Survives process restart via the store.
46    pub async fn complete_step(
47        &self,
48        id: &str,
49        step: &str,
50        result: Value,
51    ) -> Result<WorkflowInstance, String> {
52        self.store.complete_step(id, step, result).await
53    }
54
55    /// Park the workflow until [`Self::approve`].
56    pub async fn wait_for_approval(&self, id: &str) -> Result<WorkflowInstance, String> {
57        self.store
58            .set_status(id, WorkflowStatus::WaitingApproval)
59            .await
60    }
61
62    pub async fn approve(&self, id: &str) -> Result<WorkflowInstance, String> {
63        self.store.set_status(id, WorkflowStatus::Running).await
64    }
65}
66
67impl Injectable for WorkflowsService {
68    fn construct(_registry: &nestrs_core::ProviderRegistry) -> Arc<Self> {
69        panic!("WorkflowsService must be provided by WorkflowsModule::register(...)");
70    }
71}
72
73#[cfg(test)]
74mod tests {
75    use super::*;
76    use serde_json::json;
77
78    #[tokio::test]
79    async fn survives_logical_restart() {
80        let store = InMemoryWorkflowStore::shared();
81        let svc = WorkflowsService {
82            store: store.clone(),
83        };
84        let inst = svc.start("order", json!({"id": 1})).await;
85        svc.complete_step(&inst.id, "charge", json!({"ok": true}))
86            .await
87            .unwrap();
88        // "restart": new service, same store
89        let svc2 = WorkflowsService { store };
90        let loaded = svc2.get(&inst.id).await.expect("loaded");
91        assert!(loaded.steps.contains_key("charge"));
92        assert_eq!(loaded.status, WorkflowStatus::Running);
93    }
94}
nestrs_workflows/lib.rs

nestrs_workflows/
lib.rs
