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    /// What the serving install says about itself (ADR-016): the count a
263    /// reader holds an UNATTRIBUTED `current_worker` against. Stamped by the
264    /// server at read time; a page built by the engine alone carries `None`,
265    /// which the server replaces before the page leaves it.
266    ///
267    /// `None` on a page a CLIENT decoded is "provenance not reported" — a
268    /// server that predates the field — and is kept distinct from `Some(0)`:
269    /// zero is a measurement the install made, absence is one nobody made.
270    #[serde(default, skip_serializing_if = "Option::is_none")]
271    pub provenance: Option<crate::ReadProvenance>,
272}
273
274/// The table of text-rule cases the ops console mirrors, and the gate that
275/// emits it. Kept in its own file so this one stays inside the line cap.
276#[cfg(test)]
277#[path = "listing_text_cases.rs"]
278mod listing_text_cases;
279
280#[cfg(test)]
281mod tests {
282    use chrono::{DateTime, Utc};
283
284    use super::{WorkflowKind, WorkflowListFilter};
285    use crate::{RunId, WorkflowId, WorkflowStatus, WorkflowSummary};
286
287    fn summary(display_name: Option<&str>, kind: Option<&str>) -> WorkflowSummary {
288        WorkflowSummary {
289            workflow_id: WorkflowId::new(uuid::Uuid::from_u128(
290                0xabc0_0000_0000_0000_0000_0000_0000_0001,
291            )),
292            run_id: RunId::new_v4(),
293            workflow_type: String::from("checkout"),
294            status: WorkflowStatus::Running,
295            started_at: DateTime::<Utc>::default(),
296            updated_at: DateTime::<Utc>::default() + chrono::Duration::seconds(5),
297            ended_at: None,
298            parent: None,
299            failed_step: None,
300            failure_reason: None,
301            display_name: display_name.map(str::to_owned),
302            kind: kind.map(str::to_owned),
303            current_worker: None,
304            package_version: None,
305        }
306    }
307
308    #[test]
309    fn empty_filter_matches_everything() {
310        assert!(WorkflowListFilter::default().matches(&summary(None, None)));
311    }
312
313    #[test]
314    fn kind_predicate_reads_the_recorded_attribute() {
315        assert!(WorkflowKind::Workflow.matches(None));
316        assert!(!WorkflowKind::Workflow.matches(Some("workloop")));
317        assert!(WorkflowKind::Workloop.matches(Some("workloop")));
318        assert!(!WorkflowKind::Workloop.matches(None));
319    }
320
321    #[test]
322    fn text_matches_display_name_substring_case_insensitively() {
323        let filter = WorkflowListFilter {
324            text: Some(String::from("  NIGHTLY ")),
325            ..WorkflowListFilter::default()
326        };
327        assert!(filter.matches(&summary(Some("the nightly build"), None)));
328        assert!(!filter.matches(&summary(Some("weekly build"), None)));
329        assert!(!filter.matches(&summary(None, None)));
330    }
331
332    #[test]
333    fn text_matches_workflow_id_prefix() {
334        let filter = WorkflowListFilter {
335            text: Some(String::from("ABC00000")),
336            ..WorkflowListFilter::default()
337        };
338        assert!(filter.matches(&summary(None, None)));
339        let miss = WorkflowListFilter {
340            text: Some(String::from("bc00000")),
341            ..WorkflowListFilter::default()
342        };
343        assert!(
344            !miss.matches(&summary(None, None)),
345            "a prefix, not a substring"
346        );
347    }
348
349    #[test]
350    fn internal_types_hide_unless_named() {
351        let mut internal = summary(None, None);
352        internal.workflow_type = String::from("aion.schedule_coordinator");
353        assert!(!WorkflowListFilter::default().matches(&internal));
354        let named = WorkflowListFilter {
355            workflow_types: vec![String::from("aion.schedule_coordinator")],
356            ..WorkflowListFilter::default()
357        };
358        assert!(named.matches(&internal));
359        assert!(!named.matches(&summary(None, None)));
360    }
361
362    #[test]
363    fn blank_text_is_absent() {
364        let filter = WorkflowListFilter {
365            text: Some(String::from("   ")),
366            ..WorkflowListFilter::default()
367        };
368        assert_eq!(filter.text_needle(), None);
369        assert!(filter.matches(&summary(None, None)));
370    }
371
372    #[test]
373    fn list_predicates_are_any_of_and_bounds_are_inclusive() {
374        let row = summary(None, None);
375        let filter = WorkflowListFilter {
376            workflow_types: vec![String::from("other"), String::from("checkout")],
377            statuses: vec![WorkflowStatus::Completed, WorkflowStatus::Running],
378            started_after: Some(row.started_at),
379            started_before: Some(row.started_at),
380            updated_after: Some(row.updated_at),
381            updated_before: Some(row.updated_at),
382            ..WorkflowListFilter::default()
383        };
384        assert!(filter.matches(&row));
385        let excluded = WorkflowListFilter {
386            statuses: vec![WorkflowStatus::Completed],
387            ..WorkflowListFilter::default()
388        };
389        assert!(!excluded.matches(&row));
390        let too_late = WorkflowListFilter {
391            updated_after: Some(row.updated_at + chrono::Duration::seconds(1)),
392            ..WorkflowListFilter::default()
393        };
394        assert!(!too_late.matches(&row));
395    }
396}
aion_core/listing.rs

aion_core/
listing.rs
