backbone_accounting/application/workflows/
gl_posting_workflow.rs

1//! GLPosting flow implementation
2//!
3//! Generated by metaphor-schema
4//!
5//! Record a balanced AccountingPost into the General Ledger:
6//! - idempotency check on (company_id, source_type, source_id, posting_type)
7//! - validate R1..R5 (balanced, >=2 lines, postable account, party-for-AR/AP, open period)
8//! - atomically write Journal + JournalLines + immutable Ledger rows (running balance)
9//! - update account balances, mark the AccountingPost posted, emit AccountingPostPosted
10
11use serde::{Deserialize, Serialize};
12use std::sync::Arc;
13use chrono;
14use backbone_core::flow::{WorkflowStep, WorkflowContext};
15
16/// Error type for flow execution
17#[derive(Debug, Clone, thiserror::Error)]
18pub enum FlowError {
19    #[error("No current step to execute")]
20    NoCurrentStep,
21
22    #[error("Step execution failed: {0}")]
23    StepFailed(String),
24
25    #[error("Condition evaluation failed: {0}")]
26    ConditionFailed(String),
27
28    #[error("Compensation failed: {0}")]
29    CompensationFailed(String),
30
31    #[error("Flow timed out")]
32    Timeout,
33
34    #[error("Flow cancelled")]
35    Cancelled,
36
37    #[error("Invalid state transition: {from} -> {to}")]
38    InvalidTransition { from: String, to: String },
39}
40
41/// Execution status for GLPosting flow
42#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
43#[serde(rename_all = "snake_case")]
44pub enum GLPostingFlowStatus {
45    /// Flow is pending execution
46    Pending,
47    /// Flow is currently running
48    Running,
49    /// Flow is waiting for an event or condition
50    Waiting,
51    /// Flow completed successfully
52    Completed,
53    /// Flow failed
54    Failed,
55    /// Flow was cancelled
56    Cancelled,
57    /// Flow is compensating (rolling back)
58    Compensating,
59}
60
61/// Steps in GLPosting flow
62#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
63#[serde(rename_all = "snake_case")]
64pub enum GLPostingFlowStep {
65    /// idempotency_check
66    IdempotencyCheck,
67    /// validate
68    Validate,
69    /// write_journal
70    WriteJournal,
71    /// write_journal_lines
72    WriteJournalLines,
73    /// write_ledger
74    WriteLedger,
75    /// update_account_balances
76    UpdateAccountBalances,
77    /// mark_posted
78    MarkPosted,
79    /// posted
80    Posted,
81    /// rejected
82    Rejected,
83}
84
85/// Instance of GLPosting flow execution
86#[derive(Debug, Clone, Serialize, Deserialize)]
87pub struct GLPostingFlowInstance {
88    /// Unique instance ID
89    pub id: String,
90    /// Current status
91    pub status: GLPostingFlowStatus,
92    /// Current step
93    pub current_step: Option<GLPostingFlowStep>,
94    /// Flow context (variables)
95    pub context: serde_json::Value,
96    /// Completed steps
97    pub completed_steps: Vec<GLPostingFlowStep>,
98    /// Error if failed
99    pub error: Option<String>,
100    /// Created timestamp
101    pub created_at: chrono::DateTime<chrono::Utc>,
102    /// Updated timestamp
103    pub updated_at: chrono::DateTime<chrono::Utc>,
104}
105
106impl GLPostingFlowInstance {
107    /// Create a new flow instance
108    pub fn new(id: impl Into<String>) -> Self {
109        let now = chrono::Utc::now();
110        Self {
111            id: id.into(),
112            status: GLPostingFlowStatus::Pending,
113            current_step: None,
114            context: serde_json::json!({}),
115            completed_steps: Vec::new(),
116            error: None,
117            created_at: now,
118            updated_at: now,
119        }
120    }
121
122    /// Check if flow is complete
123    pub fn is_complete(&self) -> bool {
124        matches!(
125            self.status,
126            GLPostingFlowStatus::Completed | GLPostingFlowStatus::Failed | GLPostingFlowStatus::Cancelled
127        )
128    }
129
130    /// Check if flow is running
131    pub fn is_running(&self) -> bool {
132        matches!(
133            self.status,
134            GLPostingFlowStatus::Running | GLPostingFlowStatus::Waiting
135        )
136    }
137
138    /// Set a context variable
139    pub fn set_context(&mut self, key: &str, value: serde_json::Value) {
140        if let serde_json::Value::Object(ref mut map) = self.context {
141            map.insert(key.to_string(), value);
142        }
143        self.updated_at = chrono::Utc::now();
144    }
145
146    /// Get a context variable
147    pub fn get_context(&self, key: &str) -> Option<&serde_json::Value> {
148        if let serde_json::Value::Object(ref map) = self.context {
149            map.get(key)
150        } else {
151            None
152        }
153    }
154}
155
156// Phase 1: FlowInstance satisfies backbone_core::flow::WorkflowContext.
157// The entity() accessor requires the entity to be carried in the instance;
158// add `pub entity: GLPosting` in the // <<< CUSTOM section and remove the todo!.
159#[allow(unused_variables)]
160impl WorkflowContext<GLPostingFlowInstance> for GLPostingFlowInstance {
161    fn entity(&self) -> &GLPostingFlowInstance { self }
162    fn set_var(&mut self, key: &str, value: serde_json::Value) {
163        self.set_context(key, value);
164    }
165    fn get_var(&self, key: &str) -> Option<&serde_json::Value> {
166        self.get_context(key)
167    }
168}
169
170/// Step handler trait for GLPosting flow
171#[async_trait::async_trait]
172pub trait GLPostingStepHandler: Send + Sync {
173    /// Handle idempotency_check step
174    async fn handle_idempotency_check(
175        &self,
176        instance: &mut GLPostingFlowInstance,
177    ) -> Result<Option<GLPostingFlowStep>, FlowError>;
178
179    /// Handle validate step
180    async fn handle_validate(
181        &self,
182        instance: &mut GLPostingFlowInstance,
183    ) -> Result<Option<GLPostingFlowStep>, FlowError>;
184
185    /// Handle write_journal step
186    async fn handle_write_journal(
187        &self,
188        instance: &mut GLPostingFlowInstance,
189    ) -> Result<Option<GLPostingFlowStep>, FlowError>;
190
191    /// Handle write_journal_lines step
192    async fn handle_write_journal_lines(
193        &self,
194        instance: &mut GLPostingFlowInstance,
195    ) -> Result<Option<GLPostingFlowStep>, FlowError>;
196
197    /// Handle write_ledger step
198    async fn handle_write_ledger(
199        &self,
200        instance: &mut GLPostingFlowInstance,
201    ) -> Result<Option<GLPostingFlowStep>, FlowError>;
202
203    /// Handle update_account_balances step
204    async fn handle_update_account_balances(
205        &self,
206        instance: &mut GLPostingFlowInstance,
207    ) -> Result<Option<GLPostingFlowStep>, FlowError>;
208
209    /// Handle mark_posted step
210    async fn handle_mark_posted(
211        &self,
212        instance: &mut GLPostingFlowInstance,
213    ) -> Result<Option<GLPostingFlowStep>, FlowError>;
214
215    /// Handle terminal step posted
216    async fn handle_posted(
217        &self,
218        instance: &mut GLPostingFlowInstance,
219    ) -> Result<(), FlowError>;
220
221    /// Handle terminal step rejected
222    async fn handle_rejected(
223        &self,
224        instance: &mut GLPostingFlowInstance,
225    ) -> Result<(), FlowError>;
226
227}
228
229/// Executor for GLPosting flow
230pub struct GLPostingFlowExecutor<H: GLPostingStepHandler> {
231    handler: Arc<H>,
232}
233
234impl<H: GLPostingStepHandler> GLPostingFlowExecutor<H> {
235    /// Create a new flow executor
236    pub fn new(handler: Arc<H>) -> Self {
237        Self { handler }
238    }
239
240    /// Start a new flow instance
241    pub async fn start(&self, instance_id: impl Into<String>) -> Result<GLPostingFlowInstance, FlowError> {
242        let mut instance = GLPostingFlowInstance::new(instance_id);
243        instance.status = GLPostingFlowStatus::Running;
244        instance.current_step = Some(GLPostingFlowStep::IdempotencyCheck);
245        Ok(instance)
246    }
247
248    /// Execute the current step
249    pub async fn execute_step(&self, instance: &mut GLPostingFlowInstance) -> Result<(), FlowError> {
250        let current_step = match instance.current_step {
251            Some(step) => step,
252            None => return Err(FlowError::NoCurrentStep),
253        };
254
255        let next_step = match current_step {
256            GLPostingFlowStep::IdempotencyCheck => {
257                self.handler.handle_idempotency_check(instance).await?
258            }
259            GLPostingFlowStep::Validate => {
260                self.handler.handle_validate(instance).await?
261            }
262            GLPostingFlowStep::WriteJournal => {
263                self.handler.handle_write_journal(instance).await?
264            }
265            GLPostingFlowStep::WriteJournalLines => {
266                self.handler.handle_write_journal_lines(instance).await?
267            }
268            GLPostingFlowStep::WriteLedger => {
269                self.handler.handle_write_ledger(instance).await?
270            }
271            GLPostingFlowStep::UpdateAccountBalances => {
272                self.handler.handle_update_account_balances(instance).await?
273            }
274            GLPostingFlowStep::MarkPosted => {
275                self.handler.handle_mark_posted(instance).await?
276            }
277            GLPostingFlowStep::Posted => {
278                self.handler.handle_posted(instance).await?;
279                None // Terminal step
280            }
281            GLPostingFlowStep::Rejected => {
282                self.handler.handle_rejected(instance).await?;
283                None // Terminal step
284            }
285        };
286
287        // Mark current step as completed
288        instance.completed_steps.push(current_step);
289
290        // Move to next step
291        match next_step {
292            Some(next) => {
293                instance.current_step = Some(next);
294            }
295            None => {
296                instance.current_step = None;
297                instance.status = GLPostingFlowStatus::Completed;
298            }
299        }
300
301        instance.updated_at = chrono::Utc::now();
302        Ok(())
303    }
304
305    /// Run the flow to completion
306    pub async fn run(&self, instance: &mut GLPostingFlowInstance) -> Result<(), FlowError> {
307        while !instance.is_complete() && instance.status != GLPostingFlowStatus::Waiting {
308            self.execute_step(instance).await?;
309        }
310        Ok(())
311    }
312
313    /// Cancel the flow
314    pub fn cancel(&self, instance: &mut GLPostingFlowInstance) {
315        instance.status = GLPostingFlowStatus::Cancelled;
316        instance.updated_at = chrono::Utc::now();
317    }
318
319    /// Mark flow as failed
320    pub fn fail(&self, instance: &mut GLPostingFlowInstance, error: impl Into<String>) {
321        instance.status = GLPostingFlowStatus::Failed;
322        instance.error = Some(error.into());
323        instance.updated_at = chrono::Utc::now();
324    }
325}
326
327#[cfg(test)]
328mod tests {
329    use super::*;
330
331    #[test]
332    fn test_flow_instance_creation() {
333        let instance = GLPostingFlowInstance::new("test-1");
334        assert_eq!(instance.id, "test-1");
335        assert_eq!(instance.status, GLPostingFlowStatus::Pending);
336        assert!(!instance.is_complete());
337    }
338
339    #[test]
340    fn test_flow_context() {
341        let mut instance = GLPostingFlowInstance::new("test-2");
342        instance.set_context("key", serde_json::json!("value"));
343        let value = instance.get_context("key");
344        assert_eq!(value, Some(&serde_json::json!("value")));
345    }
346
347    #[test]
348    fn test_flow_status_transitions() {
349        let mut instance = GLPostingFlowInstance::new("test-3");
350        assert!(!instance.is_running());
351
352        instance.status = GLPostingFlowStatus::Running;
353        assert!(instance.is_running());
354        assert!(!instance.is_complete());
355
356        instance.status = GLPostingFlowStatus::Completed;
357        assert!(instance.is_complete());
358        assert!(!instance.is_running());
359    }
360}
backbone_accounting/application/workflows/gl_posting_workflow.rs

backbone_accounting/application/workflows/
gl_posting_workflow.rs
