runledger_runtime/catalog/
workflow.rs

1use runledger_core::jobs::{WorkflowDagBuilder, WorkflowRunEnqueue};
2use serde_json::Value;
3use uuid::Uuid;
4
5use super::{CatalogError, JobCatalog};
6
7/// Workflow DAG builder that validates job types against a [`JobCatalog`].
8#[derive(Debug, Clone)]
9pub struct CatalogWorkflowDagBuilder<'a, 'catalog> {
10    pub(super) catalog: &'catalog JobCatalog,
11    pub(super) inner: WorkflowDagBuilder<'a>,
12}
13
14impl JobCatalog {
15    /// Starts a workflow DAG builder that validates step job types against the catalog.
16    #[must_use]
17    pub fn workflow_dag<'a>(
18        &self,
19        workflow_type: &'a str,
20        metadata: &'a Value,
21    ) -> CatalogWorkflowDagBuilder<'a, '_> {
22        CatalogWorkflowDagBuilder {
23            catalog: self,
24            inner: WorkflowDagBuilder::new(workflow_type, metadata),
25        }
26    }
27}
28
29impl<'a, 'catalog> CatalogWorkflowDagBuilder<'a, 'catalog> {
30    /// Sets the organization scope for the workflow run and its steps by default.
31    #[must_use]
32    pub fn organization_id(mut self, organization_id: Uuid) -> Self {
33        self.inner = self.inner.organization_id(organization_id);
34        self
35    }
36
37    /// Clears the workflow-level organization scope.
38    #[must_use]
39    pub fn clear_organization_id(mut self) -> Self {
40        self.inner = self.inner.clear_organization_id();
41        self
42    }
43
44    /// Sets the workflow idempotency key.
45    #[must_use]
46    pub fn idempotency_key(mut self, idempotency_key: &'a str) -> Self {
47        self.inner = self.inner.idempotency_key(idempotency_key);
48        self
49    }
50
51    /// Clears the workflow idempotency key.
52    #[must_use]
53    pub fn clear_idempotency_key(mut self) -> Self {
54        self.inner = self.inner.clear_idempotency_key();
55        self
56    }
57
58    /// Declares the step whose successful output becomes the workflow result.
59    ///
60    /// # Errors
61    /// Returns [`CatalogError::WorkflowBuild`] when the step key is blank.
62    pub fn result_step(mut self, step_key: &'a str) -> Result<Self, CatalogError> {
63        self.inner = self
64            .inner
65            .result_step(step_key)
66            .map_err(CatalogError::WorkflowBuild)?;
67        Ok(self)
68    }
69
70    /// Clears the workflow result step.
71    #[must_use]
72    pub fn clear_result_step(mut self) -> Self {
73        self.inner = self.inner.clear_result_step();
74        self
75    }
76
77    /// Adds a job step after validating `job_type_name` against enabled catalog entries.
78    ///
79    /// # Errors
80    /// Returns [`CatalogError`] when the job type is unknown or disabled, or when
81    /// the underlying workflow builder rejects the step.
82    pub fn job(
83        mut self,
84        step_key: &'a str,
85        job_type_name: &str,
86        payload: &'a Value,
87    ) -> Result<Self, CatalogError> {
88        let job_type = self
89            .catalog
90            .require_catalog_enabled_job_type(job_type_name)?;
91        self.inner = self
92            .inner
93            .job(step_key, job_type.as_str(), payload)
94            .map_err(CatalogError::WorkflowBuild)?;
95        Ok(self)
96    }
97
98    /// Adds success dependencies to an existing workflow step.
99    ///
100    /// # Errors
101    /// Returns [`CatalogError::WorkflowBuild`] when the underlying workflow
102    /// builder rejects the dependency edge.
103    pub fn after_success<I>(self, step_key: &'a str, prerequisites: I) -> Result<Self, CatalogError>
104    where
105        I: IntoIterator<Item = &'a str>,
106    {
107        self.inner
108            .after_success(step_key, prerequisites)
109            .map_err(CatalogError::WorkflowBuild)
110            .map(|inner| Self {
111                catalog: self.catalog,
112                inner,
113            })
114    }
115
116    /// Adds terminal-state dependencies to an existing workflow step.
117    ///
118    /// # Errors
119    /// Returns [`CatalogError::WorkflowBuild`] when the underlying workflow
120    /// builder rejects the dependency edge.
121    pub fn after_terminal<I>(
122        self,
123        step_key: &'a str,
124        prerequisites: I,
125    ) -> Result<Self, CatalogError>
126    where
127        I: IntoIterator<Item = &'a str>,
128    {
129        self.inner
130            .after_terminal(step_key, prerequisites)
131            .map_err(CatalogError::WorkflowBuild)
132            .map(|inner| Self {
133                catalog: self.catalog,
134                inner,
135            })
136    }
137
138    /// Builds the workflow enqueue payload.
139    ///
140    /// This is an alias for [`Self::try_build`].
141    ///
142    /// # Errors
143    /// Returns [`CatalogError::WorkflowBuild`] when final workflow validation fails.
144    pub fn build(self) -> Result<WorkflowRunEnqueue<'a>, CatalogError> {
145        self.try_build()
146    }
147
148    /// Builds the workflow enqueue payload.
149    ///
150    /// # Errors
151    /// Returns [`CatalogError::WorkflowBuild`] when final workflow validation fails.
152    pub fn try_build(self) -> Result<WorkflowRunEnqueue<'a>, CatalogError> {
153        self.inner.try_build().map_err(CatalogError::WorkflowBuild)
154    }
155}
runledger_runtime/catalog/workflow.rs

runledger_runtime/catalog/
workflow.rs
