a3s-flow 0.6.1

Durable workflow engine and Rust SDK for A3S
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
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

use a3s_orm::Migration;

const EVENTS_SQL: &str = r#"
CREATE TABLE IF NOT EXISTS flow_events (
    run_id TEXT NOT NULL,
    sequence BIGINT NOT NULL CHECK (sequence >= 1),
    event_id TEXT NOT NULL,
    timestamp TEXT NOT NULL,
    event_json TEXT NOT NULL,
    PRIMARY KEY (run_id, sequence)
);

CREATE INDEX IF NOT EXISTS idx_flow_events_run_id_sequence
ON flow_events (run_id, sequence);
"#;

#[cfg(feature = "postgres")]
const POSTGRES_TASKS_SQL: &str = r#"
CREATE TABLE IF NOT EXISTS flow_tasks (
    queue_name TEXT NOT NULL,
    task_id TEXT NOT NULL,
    task_json TEXT NOT NULL,
    status TEXT NOT NULL CHECK (status IN ('pending', 'inflight')),
    enqueued_at_nanos BIGINT NOT NULL,
    leased_at_nanos BIGINT,
    lease_id TEXT,
    updated_at_nanos BIGINT NOT NULL,
    PRIMARY KEY (queue_name, task_id)
);

CREATE UNIQUE INDEX IF NOT EXISTS idx_flow_tasks_queue_lease
ON flow_tasks (queue_name, lease_id)
WHERE lease_id IS NOT NULL;

CREATE INDEX IF NOT EXISTS idx_flow_tasks_pending_order
ON flow_tasks (queue_name, status, enqueued_at_nanos, task_id);

CREATE TABLE IF NOT EXISTS flow_task_dead_letters (
    queue_name TEXT NOT NULL,
    dead_letter_id TEXT NOT NULL,
    lease_id TEXT NOT NULL,
    task_json TEXT NOT NULL,
    reason TEXT NOT NULL,
    dead_lettered_at_nanos BIGINT NOT NULL,
    leased_at_nanos BIGINT,
    PRIMARY KEY (queue_name, dead_letter_id)
);

CREATE INDEX IF NOT EXISTS idx_flow_task_dead_letters_queue_time
ON flow_task_dead_letters (queue_name, dead_lettered_at_nanos, dead_letter_id);
"#;

#[cfg(any(feature = "postgres", feature = "sqlite"))]
const RETENTION_SQL: &str = r#"
CREATE TABLE IF NOT EXISTS flow_history_holds (
    run_id TEXT NOT NULL,
    hold_id TEXT NOT NULL,
    reason TEXT NOT NULL,
    created_at TEXT NOT NULL,
    PRIMARY KEY (run_id, hold_id)
);

CREATE TABLE IF NOT EXISTS flow_history_tombstones (
    run_id TEXT PRIMARY KEY,
    deleted_at TEXT NOT NULL,
    terminal_sequence BIGINT NOT NULL CHECK (terminal_sequence >= 1),
    terminal_event_id TEXT NOT NULL,
    terminal_event_key TEXT NOT NULL,
    history_sha256 TEXT NOT NULL
);
"#;

#[cfg(feature = "sqlite")]
pub(crate) fn sqlite_migrations() -> Vec<Migration> {
    vec![
        Migration::new(
            "a3s-flow-0001-events",
            "create Flow event history",
            EVENTS_SQL,
        ),
        Migration::new(
            "a3s-flow-0002-retention",
            "create Flow history retention guards and tombstones",
            RETENTION_SQL,
        ),
    ]
}

#[cfg(feature = "postgres")]
pub(crate) fn postgres_migrations() -> Vec<Migration> {
    vec![
        Migration::new(
            "a3s-flow-0001-events",
            "create Flow event history",
            EVENTS_SQL,
        ),
        Migration::new(
            "a3s-flow-0002-tasks",
            "create Flow task dispatch tables",
            POSTGRES_TASKS_SQL,
        ),
        Migration::new(
            "a3s-flow-0003-retention",
            "create Flow history retention guards and tombstones",
            RETENTION_SQL,
        ),
    ]
}