assay-workflow 0.4.6

Durable workflow engine with REST+SSE API on PostgreSQL 18 and SQLite backends. Embeddable library or standalone server (via assay-engine).
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77

//! Backend implementations of `WorkflowStore`.
//!
//! The trait itself lives in `assay-domain`; re-exported here so existing
//! `crate::store::WorkflowStore` and `crate::store::<DTO>` paths resolve
//! unchanged.

pub mod postgres;
pub mod sqlite;

pub use assay_domain::store::WorkflowStore;
pub use assay_domain::{NamespaceRecord, NamespaceStats, QueueStats};

use assay_domain::{RetryFailedActivityResult, SettleOutcome};

/// Marks a history event as not recording any activity's current
/// settlement: a payload the backfill could not read, or a settlement a
/// retry has superseded. Distinct from NULL, which the backfill revisits.
pub(crate) const NOT_A_SETTLEMENT: i64 = -1;

/// The `activity_id` a terminal activity event carries in its payload, or
/// `-1` when the payload predates the dedicated column or doesn't parse.
/// No activity has a negative id, so the sentinel is a durable "looked, and
/// there is nothing here" marker rather than a value the backfill revisits.
pub(crate) fn payload_activity_id(payload: Option<&str>) -> i64 {
    payload
        .and_then(|raw| serde_json::from_str::<serde_json::Value>(raw).ok())
        .and_then(|value| value.get("activity_id").and_then(serde_json::Value::as_i64))
        .unwrap_or(NOT_A_SETTLEMENT)
}

pub(crate) fn settle_outcome(was_terminal: bool, event_present: bool) -> SettleOutcome {
    match (was_terminal, event_present) {
        (false, _) => SettleOutcome::Settled,
        (true, false) => SettleOutcome::Repaired,
        (true, true) => SettleOutcome::AlreadySettled,
    }
}

pub(crate) fn retry_denial(
    status: String,
    parent_id: Option<String>,
    archived_at: Option<f64>,
) -> Option<RetryFailedActivityResult> {
    if archived_at.is_some() {
        Some(RetryFailedActivityResult::Archived)
    } else if parent_id.is_some() {
        Some(RetryFailedActivityResult::ChildWorkflow)
    } else if status != "FAILED" {
        Some(RetryFailedActivityResult::NotFailed { status })
    } else {
        None
    }
}

pub(crate) struct RetryEvent<'a> {
    pub activity_id: i64,
    pub activity_seq: i32,
    pub activity_name: &'a str,
    pub failed_event_seq: i32,
    pub requested_by: &'a str,
    pub reason: &'a str,
    pub invalidated_activities: u64,
}

impl RetryEvent<'_> {
    pub fn payload(&self) -> serde_json::Value {
        serde_json::json!({
            "activity_id": self.activity_id,
            "activity_seq": self.activity_seq,
            "name": self.activity_name,
            "failed_event_seq": self.failed_event_seq,
            "requested_by": self.requested_by,
            "reason": self.reason,
            "invalidated_activities": self.invalidated_activities,
        })
    }
}