aion_core/
listing.rs

1//! The workflow list contract: filter, sort, request, and page.
2//!
3//! These types are the ONE shape every list surface speaks — `POST
4//! /workflows/list`, gRPC `ListWorkflows`, the SDKs, and the ops console,
5//! whose TypeScript is generated from them. They live in this leaf crate
6//! because it is the only one that can cross the `ts-rs` boundary.
7//!
8//! Sorting is mandatory and the server assumes no default: a request without
9//! a `sort` is malformed, not "sorted somehow". Paging is keyset-only through
10//! an opaque cursor minted by the store; there is no offset.
11
12use chrono::{DateTime, Utc};
13use serde::{Deserialize, Serialize};
14
15use crate::{SearchAttributeValue, WorkflowId, WorkflowStatus, WorkflowSummary};
16
17/// The search attribute the server stamps on every run at start with the
18/// namespace it belongs to. The visibility projection reads it to place the
19/// row under its namespace; a run recorded without it lists under
20/// [`DEFAULT_NAMESPACE`] — see [`namespace_from_attributes`].
21pub const NAMESPACE_ATTRIBUTE: &str = "aion.namespace";
22
23/// The namespace a run belongs to when its history carries no
24/// [`NAMESPACE_ATTRIBUTE`]: a run recorded by an embedded engine that stamps
25/// no placement, or one recorded before namespaces existed. The engine
26/// routes such a run's recovery under this name, so the projection must
27/// list it under the same one or an upgraded store's runs vanish.
28pub const DEFAULT_NAMESPACE: &str = "default";
29
30/// The namespace `attributes` place a run in: the string value of
31/// [`NAMESPACE_ATTRIBUTE`], or [`DEFAULT_NAMESPACE`] when the attribute is
32/// absent or not a string.
33#[must_use]
34pub fn namespace_from_attributes<S: std::hash::BuildHasher>(
35    attributes: &std::collections::HashMap<String, SearchAttributeValue, S>,
36) -> String {
37    match attributes.get(NAMESPACE_ATTRIBUTE) {
38        Some(SearchAttributeValue::String(namespace)) => namespace.clone(),
39        _ => String::from(DEFAULT_NAMESPACE),
40    }
41}
42
43/// Workflow types the engine runs for its own plumbing — today the schedule
44/// coordinator, which durably hosts every schedule's lifecycle. They live in
45/// the same stores as user workflows, so every list HIDES them unless a
46/// request names one in `workflow_types` explicitly (the operator's escape
47/// hatch). A new engine-internal type must be added here, and the engine's
48/// own constant for it must equal this spelling.
49pub const INTERNAL_WORKFLOW_TYPES: &[&str] = &["aion.schedule_coordinator"];
50
51/// Whether `workflow_type` is an engine-internal type hidden from lists.
52#[must_use]
53pub fn is_internal_workflow_type(workflow_type: &str) -> bool {
54    INTERNAL_WORKFLOW_TYPES.contains(&workflow_type)
55}
56
57/// A run's document kind as a list predicate.
58#[derive(Serialize, Deserialize, ts_rs::TS, Clone, Copy, Debug, PartialEq, Eq, Hash)]
59#[serde(rename_all = "snake_case")]
60pub enum WorkflowKind {
61    /// An ordinary workflow — a run that carries no `aion.kind` attribute.
62    Workflow,
63    /// A workloop run (`aion.kind = "workloop"`).
64    Workloop,
65}
66
67impl WorkflowKind {
68    /// Whether a row's recorded kind attribute (the summary's `kind` field)
69    /// satisfies this predicate.
70    #[must_use]
71    pub fn matches(self, recorded_kind: Option<&str>) -> bool {
72        match self {
73            Self::Workflow => recorded_kind != Some(crate::WORKLOOP_KIND),
74            Self::Workloop => recorded_kind == Some(crate::WORKLOOP_KIND),
75        }
76    }
77}
78
79/// Every predicate a list request may carry. All optional; an empty filter
80/// matches the whole namespace. Predicates AND together.
81#[derive(Serialize, Deserialize, ts_rs::TS, Clone, Debug, Default, PartialEq, Eq)]
82pub struct WorkflowListFilter {
83    /// Restrict to one document kind.
84    #[serde(default)]
85    pub kind: Option<WorkflowKind>,
86    /// Match any of these workflow types exactly. Empty = any type. Naming an
87    /// engine-internal type here is the only way such a run lists.
88    #[serde(default)]
89    pub workflow_types: Vec<String>,
90    /// Match any of these projected statuses. Empty = any status.
91    #[serde(default)]
92    pub statuses: Vec<WorkflowStatus>,
93    /// Match runs started at or after this instant.
94    #[serde(default)]
95    pub started_after: Option<DateTime<Utc>>,
96    /// Match runs started at or before this instant.
97    #[serde(default)]
98    pub started_before: Option<DateTime<Utc>>,
99    /// Match runs whose latest recorded event is at or after this instant.
100    #[serde(default)]
101    pub updated_after: Option<DateTime<Utc>>,
102    /// Match runs whose latest recorded event is at or before this instant.
103    #[serde(default)]
104    pub updated_before: Option<DateTime<Utc>>,
105    /// Match children of this parent workflow.
106    #[serde(default)]
107    pub parent: Option<WorkflowId>,
108    /// Case-insensitive substring of the display name, OR a prefix of the
109    /// workflow id's canonical UUID string. Whitespace-only is treated as
110    /// absent.
111    #[serde(default)]
112    pub text: Option<String>,
113}
114
115impl WorkflowListFilter {
116    /// Whether `summary` satisfies every predicate. This is the ONE definition
117    /// of the filter's meaning: both store backends call it after their range
118    /// read, so a predicate can never mean two things.
119    ///
120    /// An engine-internal type ([`INTERNAL_WORKFLOW_TYPES`]) matches only when
121    /// `workflow_types` names it; an empty `workflow_types` means "every USER
122    /// type", never "everything".
123    #[must_use]
124    pub fn matches(&self, summary: &WorkflowSummary) -> bool {
125        self.kind
126            .is_none_or(|kind| kind.matches(summary.kind.as_deref()))
127            && self.matches_workflow_type(&summary.workflow_type)
128            && (self.statuses.is_empty() || self.statuses.contains(&summary.status))
129            && self
130                .started_after
131                .is_none_or(|bound| summary.started_at >= bound)
132            && self
133                .started_before
134                .is_none_or(|bound| summary.started_at <= bound)
135            && self
136                .updated_after
137                .is_none_or(|bound| summary.updated_at >= bound)
138            && self
139                .updated_before
140                .is_none_or(|bound| summary.updated_at <= bound)
141            && self
142                .parent
143                .as_ref()
144                .is_none_or(|parent| summary.parent.as_ref() == Some(parent))
145            && self.matches_text(summary)
146    }
147
148    fn matches_workflow_type(&self, workflow_type: &str) -> bool {
149        if self.workflow_types.is_empty() {
150            !is_internal_workflow_type(workflow_type)
151        } else {
152            self.workflow_types
153                .iter()
154                .any(|named| named == workflow_type)
155        }
156    }
157
158    /// The trimmed text predicate, or `None` when it is absent or blank.
159    #[must_use]
160    pub fn text_needle(&self) -> Option<&str> {
161        self.text
162            .as_deref()
163            .map(str::trim)
164            .filter(|needle| !needle.is_empty())
165    }
166
167    fn matches_text(&self, summary: &WorkflowSummary) -> bool {
168        let Some(needle) = self.text_needle() else {
169            return true;
170        };
171        let lowered = needle.to_lowercase();
172        let by_name = summary
173            .display_name
174            .as_deref()
175            .is_some_and(|name| name.to_lowercase().contains(&lowered));
176        let by_id = summary
177            .workflow_id
178            .to_string()
179            .to_lowercase()
180            .starts_with(&lowered);
181        by_name || by_id
182    }
183}
184
185/// The column a page is ordered by.
186#[derive(Serialize, Deserialize, ts_rs::TS, Clone, Copy, Debug, PartialEq, Eq, Hash)]
187#[serde(rename_all = "snake_case")]
188pub enum WorkflowSortField {
189    /// The run's start instant.
190    StartedAt,
191    /// The instant of the run's latest recorded event.
192    UpdatedAt,
193    /// The run's terminal instant; running workflows sort as the smallest value.
194    EndedAt,
195    /// The workflow type, bytewise.
196    WorkflowType,
197    /// The projected status, by canonical name.
198    Status,
199    /// The display name, case-insensitively; unnamed sorts as empty.
200    DisplayName,
201}
202
203impl WorkflowSortField {
204    /// Every field, in declaration order — the set a client may offer.
205    pub const ALL: [Self; 6] = [
206        Self::StartedAt,
207        Self::UpdatedAt,
208        Self::EndedAt,
209        Self::WorkflowType,
210        Self::Status,
211        Self::DisplayName,
212    ];
213}
214
215/// Ascending or descending.
216#[derive(Serialize, Deserialize, ts_rs::TS, Clone, Copy, Debug, PartialEq, Eq, Hash)]
217#[serde(rename_all = "snake_case")]
218pub enum SortDirection {
219    /// Smallest first.
220    Asc,
221    /// Largest first.
222    Desc,
223}
224
225/// The page order. Required on every request.
226#[derive(Serialize, Deserialize, ts_rs::TS, Clone, Copy, Debug, PartialEq, Eq, Hash)]
227pub struct WorkflowSort {
228    /// The ordering column.
229    pub field: WorkflowSortField,
230    /// The ordering direction.
231    pub direction: SortDirection,
232}
233
234/// One list request, identical on HTTP and gRPC.
235#[derive(Serialize, Deserialize, ts_rs::TS, Clone, Debug, PartialEq, Eq)]
236pub struct WorkflowListRequest {
237    /// The namespace to list. The server narrows to what the caller holds.
238    pub namespace: String,
239    /// Predicates, applied before the limit.
240    #[serde(default)]
241    pub filter: WorkflowListFilter,
242    /// Page order — required.
243    pub sort: WorkflowSort,
244    /// Continue after the row a previous page's `next_cursor` names. The
245    /// cursor is bound to `(namespace, filter, sort)`; under a different
246    /// query it is refused.
247    #[serde(default)]
248    pub cursor: Option<String>,
249    /// Maximum rows in the page. Zero is malformed.
250    pub limit: u32,
251}
252
253/// One page of the list.
254#[derive(Serialize, Deserialize, ts_rs::TS, Clone, Debug, PartialEq, Eq)]
255pub struct WorkflowListPage {
256    /// The rows, in the requested order. Exactly `limit` unless exhausted.
257    pub items: Vec<WorkflowSummary>,
258    /// Pass back as `cursor` for the next page; `None` when exhausted.
259    pub next_cursor: Option<String>,
260    /// How many rows match the filter in total.
261    pub count: u64,
262}
263
264#[cfg(test)]
265mod tests {
266    use chrono::{DateTime, Utc};
267
268    use super::{WorkflowKind, WorkflowListFilter};
269    use crate::{RunId, WorkflowId, WorkflowStatus, WorkflowSummary};
270
271    fn summary(display_name: Option<&str>, kind: Option<&str>) -> WorkflowSummary {
272        WorkflowSummary {
273            workflow_id: WorkflowId::new(uuid::Uuid::from_u128(
274                0xabc0_0000_0000_0000_0000_0000_0000_0001,
275            )),
276            run_id: RunId::new_v4(),
277            workflow_type: String::from("checkout"),
278            status: WorkflowStatus::Running,
279            started_at: DateTime::<Utc>::default(),
280            updated_at: DateTime::<Utc>::default() + chrono::Duration::seconds(5),
281            ended_at: None,
282            parent: None,
283            failed_step: None,
284            failure_reason: None,
285            display_name: display_name.map(str::to_owned),
286            kind: kind.map(str::to_owned),
287        }
288    }
289
290    #[test]
291    fn empty_filter_matches_everything() {
292        assert!(WorkflowListFilter::default().matches(&summary(None, None)));
293    }
294
295    #[test]
296    fn kind_predicate_reads_the_recorded_attribute() {
297        assert!(WorkflowKind::Workflow.matches(None));
298        assert!(!WorkflowKind::Workflow.matches(Some("workloop")));
299        assert!(WorkflowKind::Workloop.matches(Some("workloop")));
300        assert!(!WorkflowKind::Workloop.matches(None));
301    }
302
303    #[test]
304    fn text_matches_display_name_substring_case_insensitively() {
305        let filter = WorkflowListFilter {
306            text: Some(String::from("  NIGHTLY ")),
307            ..WorkflowListFilter::default()
308        };
309        assert!(filter.matches(&summary(Some("the nightly build"), None)));
310        assert!(!filter.matches(&summary(Some("weekly build"), None)));
311        assert!(!filter.matches(&summary(None, None)));
312    }
313
314    #[test]
315    fn text_matches_workflow_id_prefix() {
316        let filter = WorkflowListFilter {
317            text: Some(String::from("ABC00000")),
318            ..WorkflowListFilter::default()
319        };
320        assert!(filter.matches(&summary(None, None)));
321        let miss = WorkflowListFilter {
322            text: Some(String::from("bc00000")),
323            ..WorkflowListFilter::default()
324        };
325        assert!(
326            !miss.matches(&summary(None, None)),
327            "a prefix, not a substring"
328        );
329    }
330
331    #[test]
332    fn internal_types_hide_unless_named() {
333        let mut internal = summary(None, None);
334        internal.workflow_type = String::from("aion.schedule_coordinator");
335        assert!(!WorkflowListFilter::default().matches(&internal));
336        let named = WorkflowListFilter {
337            workflow_types: vec![String::from("aion.schedule_coordinator")],
338            ..WorkflowListFilter::default()
339        };
340        assert!(named.matches(&internal));
341        assert!(!named.matches(&summary(None, None)));
342    }
343
344    #[test]
345    fn blank_text_is_absent() {
346        let filter = WorkflowListFilter {
347            text: Some(String::from("   ")),
348            ..WorkflowListFilter::default()
349        };
350        assert_eq!(filter.text_needle(), None);
351        assert!(filter.matches(&summary(None, None)));
352    }
353
354    #[test]
355    fn list_predicates_are_any_of_and_bounds_are_inclusive() {
356        let row = summary(None, None);
357        let filter = WorkflowListFilter {
358            workflow_types: vec![String::from("other"), String::from("checkout")],
359            statuses: vec![WorkflowStatus::Completed, WorkflowStatus::Running],
360            started_after: Some(row.started_at),
361            started_before: Some(row.started_at),
362            updated_after: Some(row.updated_at),
363            updated_before: Some(row.updated_at),
364            ..WorkflowListFilter::default()
365        };
366        assert!(filter.matches(&row));
367        let excluded = WorkflowListFilter {
368            statuses: vec![WorkflowStatus::Completed],
369            ..WorkflowListFilter::default()
370        };
371        assert!(!excluded.matches(&row));
372        let too_late = WorkflowListFilter {
373            updated_after: Some(row.updated_at + chrono::Duration::seconds(1)),
374            ..WorkflowListFilter::default()
375        };
376        assert!(!too_late.matches(&row));
377    }
378}
aion_core/listing.rs

aion_core/
listing.rs
