bcode_workflow_store 0.0.1-alpha.0

Durable workflow persistence for bcode
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
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132

//! Revision-pinned exact node reads, independent of whole-definition decoding.
use super::{Database, DatabaseValue, RepresentationError as Error};
use switchy_database::{
    query::{FilterableQuery, SortDirection},
    schema::{DataType, create_index, create_table},
};
const NODES: &str = "workflow_definition_nodes_v2";
#[cfg(test)]
mod tests;
pub(super) async fn install(db: &dyn Database) -> Result<(), Error> {
    create_table(NODES)
        .column(super::column("identity", DataType::Text, false))
        .primary_key("identity")
        .column(super::column("run_id", DataType::Text, false))
        .column(super::column("revision", DataType::Int, false))
        .column(super::column("node_id", DataType::Text, false))
        .column(super::column("node_json", DataType::Text, false))
        .execute(db)
        .await?;
    create_index("idx_workflow_definition_nodes_v2_page")
        .table(NODES)
        .columns(vec!["run_id", "revision", "node_id"])
        .execute(db)
        .await?;
    Ok(())
}
pub(super) async fn persist(
    db: &dyn Database,
    run: &str,
    revision: u64,
    definition: &bcode_workflow::WorkflowDefinition,
) -> Result<(), Error> {
    for node in definition.nodes.values() {
        let json = serde_json::to_string(node).map_err(|_| Error::InvalidAdmission)?;
        if json.len() > super::MAX_ADMISSION_BYTES {
            return Err(Error::InvalidAdmission);
        }
        db.insert(NODES)
            .value(
                "identity",
                serde_json::to_string(&(run, revision, &node.id))
                    .map_err(|_| Error::InvalidAdmission)?,
            )
            .value("run_id", run)
            .value(
                "revision",
                i64::try_from(revision).map_err(|_| Error::InvalidAdmission)?,
            )
            .value("node_id", node.id.as_str())
            .value("node_json", json)
            .execute(db)
            .await?;
    }
    Ok(())
}
/// Read the revision scalar only; never decode the graph.
/// # Errors
/// Rejects absent, malformed or unsupported revision state.
pub async fn revision(db: &dyn Database, run: &str) -> Result<u64, Error> {
    let rows = db
        .select("workflow_graphs_v2")
        .columns(&["revision"])
        .where_eq("run_id", run)
        .limit(1)
        .execute(db)
        .await?;
    match rows.first().and_then(|row| row.get("revision")) {
        Some(DatabaseValue::Int64(n)) if n > 0 => {
            u64::try_from(n).map_err(|_| Error::MaintenanceRequired)
        }
        _ => Err(Error::MaintenanceRequired),
    }
}
fn decode(row: &switchy_database::Row) -> Result<bcode_workflow::NodeDefinition, Error> {
    match row.get("node_json") {
        Some(DatabaseValue::String(json)) if json.len() <= super::MAX_ADMISSION_BYTES => {
            serde_json::from_str(&json).map_err(|_| Error::MaintenanceRequired)
        }
        _ => Err(Error::MaintenanceRequired),
    }
}
/// Read one canonical node at an exact retained revision without opening a whole snapshot.
/// # Errors
/// Rejects malformed node payloads or storage failure. Absence is explicit.
pub async fn node(
    db: &dyn Database,
    run: &str,
    revision: u64,
    id: &str,
) -> Result<Option<bcode_workflow::NodeDefinition>, Error> {
    let rows = db
        .select(NODES)
        .columns(&["node_json"])
        .where_eq("run_id", run)
        .where_eq(
            "revision",
            i64::try_from(revision).map_err(|_| Error::InvalidAdmission)?,
        )
        .where_eq("node_id", id)
        .limit(1)
        .execute(db)
        .await?;
    rows.first().map(decode).transpose()
}
/// Page exact retained node definitions. Page count is not a graph-size cap.
/// # Errors
/// Rejects invalid bounds, malformed node payloads or storage failure.
pub async fn nodes(
    db: &dyn Database,
    run: &str,
    revision: u64,
    after: Option<&str>,
    limit: u16,
) -> Result<Vec<bcode_workflow::NodeDefinition>, Error> {
    if limit == 0 || limit > 100 {
        return Err(Error::InvalidAdmission);
    }
    let mut query = db
        .select(NODES)
        .columns(&["node_json"])
        .where_eq("run_id", run)
        .where_eq(
            "revision",
            i64::try_from(revision).map_err(|_| Error::InvalidAdmission)?,
        )
        .sort("node_id", SortDirection::Asc)
        .limit(usize::from(limit));
    if let Some(after) = after {
        query = query.where_gt("node_id", after);
    }
    query.execute(db).await?.iter().map(decode).collect()
}