aion-core 0.29.0

//! The workflow list contract: filter, sort, request, and page.
//!
//! These types are the ONE shape every list surface speaks — `POST
//! /workflows/list`, gRPC `ListWorkflows`, the SDKs, and the ops console,
//! whose TypeScript is generated from them. They live in this leaf crate
//! because it is the only one that can cross the `ts-rs` boundary.
//!
//! Sorting is mandatory and the server assumes no default: a request without
//! a `sort` is malformed, not "sorted somehow". Paging is keyset-only through
//! an opaque cursor minted by the store; there is no offset.

use chrono::{DateTime, Utc};
use serde::{Deserialize, Serialize};

use crate::{SearchAttributeValue, WorkflowId, WorkflowStatus, WorkflowSummary};

/// The search attribute the server stamps on every run at start with the
/// namespace it belongs to. The visibility projection reads it to place the
/// row under its namespace; a run recorded without it lists under
/// [`DEFAULT_NAMESPACE`] — see [`namespace_from_attributes`].
pub const NAMESPACE_ATTRIBUTE: &str = "aion.namespace";

/// The namespace a run belongs to when its history carries no
/// [`NAMESPACE_ATTRIBUTE`]: a run recorded by an embedded engine that stamps
/// no placement, or one recorded before namespaces existed. The engine
/// routes such a run's recovery under this name, so the projection must
/// list it under the same one or an upgraded store's runs vanish.
pub const DEFAULT_NAMESPACE: &str = "default";

/// The namespace `attributes` place a run in: the string value of
/// [`NAMESPACE_ATTRIBUTE`], or [`DEFAULT_NAMESPACE`] when the attribute is
/// absent or not a string.
#[must_use]
pub fn namespace_from_attributes<S: std::hash::BuildHasher>(
    attributes: &std::collections::HashMap<String, SearchAttributeValue, S>,
) -> String {
    match attributes.get(NAMESPACE_ATTRIBUTE) {
        Some(SearchAttributeValue::String(namespace)) => namespace.clone(),
        _ => String::from(DEFAULT_NAMESPACE),
    }
}

/// Workflow types the engine runs for its own plumbing — today the schedule
/// coordinator, which durably hosts every schedule's lifecycle. They live in
/// the same stores as user workflows, so every list HIDES them unless a
/// request names one in `workflow_types` explicitly (the operator's escape
/// hatch). A new engine-internal type must be added here, and the engine's
/// own constant for it must equal this spelling.
pub const INTERNAL_WORKFLOW_TYPES: &[&str] = &["aion.schedule_coordinator"];

/// Whether `workflow_type` is an engine-internal type hidden from lists.
#[must_use]
pub fn is_internal_workflow_type(workflow_type: &str) -> bool {
    INTERNAL_WORKFLOW_TYPES.contains(&workflow_type)
}

/// A run's document kind as a list predicate.
#[derive(Serialize, Deserialize, ts_rs::TS, Clone, Copy, Debug, PartialEq, Eq, Hash)]
#[serde(rename_all = "snake_case")]
pub enum WorkflowKind {
    /// An ordinary workflow — a run that carries no `aion.kind` attribute.
    Workflow,
    /// A workloop run (`aion.kind = "workloop"`).
    Workloop,
}

impl WorkflowKind {
    /// Whether a row's recorded kind attribute (the summary's `kind` field)
    /// satisfies this predicate.
    #[must_use]
    pub fn matches(self, recorded_kind: Option<&str>) -> bool {
        match self {
            Self::Workflow => recorded_kind != Some(crate::WORKLOOP_KIND),
            Self::Workloop => recorded_kind == Some(crate::WORKLOOP_KIND),
        }
    }
}

/// Every predicate a list request may carry. All optional; an empty filter
/// matches the whole namespace. Predicates AND together.
#[derive(Serialize, Deserialize, ts_rs::TS, Clone, Debug, Default, PartialEq, Eq)]
pub struct WorkflowListFilter {
    /// Restrict to one document kind.
    #[serde(default)]
    pub kind: Option<WorkflowKind>,
    /// Match any of these workflow types exactly. Empty = any type. Naming an
    /// engine-internal type here is the only way such a run lists.
    #[serde(default)]
    pub workflow_types: Vec<String>,
    /// Match any of these projected statuses. Empty = any status.
    #[serde(default)]
    pub statuses: Vec<WorkflowStatus>,
    /// Match runs started at or after this instant.
    #[serde(default)]
    pub started_after: Option<DateTime<Utc>>,
    /// Match runs started at or before this instant.
    #[serde(default)]
    pub started_before: Option<DateTime<Utc>>,
    /// Match runs whose latest recorded event is at or after this instant.
    #[serde(default)]
    pub updated_after: Option<DateTime<Utc>>,
    /// Match runs whose latest recorded event is at or before this instant.
    #[serde(default)]
    pub updated_before: Option<DateTime<Utc>>,
    /// Match children of this parent workflow.
    #[serde(default)]
    pub parent: Option<WorkflowId>,
    /// Case-insensitive substring of the display name, OR a prefix of the
    /// workflow id's canonical UUID string. Whitespace-only is treated as
    /// absent.
    #[serde(default)]
    pub text: Option<String>,
}

impl WorkflowListFilter {
    /// Whether `summary` satisfies every predicate. This is the ONE definition
    /// of the filter's meaning: both store backends call it after their range
    /// read, so a predicate can never mean two things.
    ///
    /// An engine-internal type ([`INTERNAL_WORKFLOW_TYPES`]) matches only when
    /// `workflow_types` names it; an empty `workflow_types` means "every USER
    /// type", never "everything".
    #[must_use]
    pub fn matches(&self, summary: &WorkflowSummary) -> bool {
        self.kind
            .is_none_or(|kind| kind.matches(summary.kind.as_deref()))
            && self.matches_workflow_type(&summary.workflow_type)
            && (self.statuses.is_empty() || self.statuses.contains(&summary.status))
            && self
                .started_after
                .is_none_or(|bound| summary.started_at >= bound)
            && self
                .started_before
                .is_none_or(|bound| summary.started_at <= bound)
            && self
                .updated_after
                .is_none_or(|bound| summary.updated_at >= bound)
            && self
                .updated_before
                .is_none_or(|bound| summary.updated_at <= bound)
            && self
                .parent
                .as_ref()
                .is_none_or(|parent| summary.parent.as_ref() == Some(parent))
            && self.matches_text(summary)
    }

    fn matches_workflow_type(&self, workflow_type: &str) -> bool {
        if self.workflow_types.is_empty() {
            !is_internal_workflow_type(workflow_type)
        } else {
            self.workflow_types
                .iter()
                .any(|named| named == workflow_type)
        }
    }

    /// The trimmed text predicate, or `None` when it is absent or blank.
    #[must_use]
    pub fn text_needle(&self) -> Option<&str> {
        self.text
            .as_deref()
            .map(str::trim)
            .filter(|needle| !needle.is_empty())
    }

    fn matches_text(&self, summary: &WorkflowSummary) -> bool {
        let Some(needle) = self.text_needle() else {
            return true;
        };
        let lowered = needle.to_lowercase();
        let by_name = summary
            .display_name
            .as_deref()
            .is_some_and(|name| name.to_lowercase().contains(&lowered));
        let by_id = summary
            .workflow_id
            .to_string()
            .to_lowercase()
            .starts_with(&lowered);
        by_name || by_id
    }
}

/// The column a page is ordered by.
#[derive(Serialize, Deserialize, ts_rs::TS, Clone, Copy, Debug, PartialEq, Eq, Hash)]
#[serde(rename_all = "snake_case")]
pub enum WorkflowSortField {
    /// The run's start instant.
    StartedAt,
    /// The instant of the run's latest recorded event.
    UpdatedAt,
    /// The run's terminal instant; running workflows sort as the smallest value.
    EndedAt,
    /// The workflow type, bytewise.
    WorkflowType,
    /// The projected status, by canonical name.
    Status,
    /// The display name, case-insensitively; unnamed sorts as empty.
    DisplayName,
}

impl WorkflowSortField {
    /// Every field, in declaration order — the set a client may offer.
    pub const ALL: [Self; 6] = [
        Self::StartedAt,
        Self::UpdatedAt,
        Self::EndedAt,
        Self::WorkflowType,
        Self::Status,
        Self::DisplayName,
    ];
}

/// Ascending or descending.
#[derive(Serialize, Deserialize, ts_rs::TS, Clone, Copy, Debug, PartialEq, Eq, Hash)]
#[serde(rename_all = "snake_case")]
pub enum SortDirection {
    /// Smallest first.
    Asc,
    /// Largest first.
    Desc,
}

/// The page order. Required on every request.
#[derive(Serialize, Deserialize, ts_rs::TS, Clone, Copy, Debug, PartialEq, Eq, Hash)]
pub struct WorkflowSort {
    /// The ordering column.
    pub field: WorkflowSortField,
    /// The ordering direction.
    pub direction: SortDirection,
}

/// One list request, identical on HTTP and gRPC.
#[derive(Serialize, Deserialize, ts_rs::TS, Clone, Debug, PartialEq, Eq)]
pub struct WorkflowListRequest {
    /// The namespace to list. The server narrows to what the caller holds.
    pub namespace: String,
    /// Predicates, applied before the limit.
    #[serde(default)]
    pub filter: WorkflowListFilter,
    /// Page order — required.
    pub sort: WorkflowSort,
    /// Continue after the row a previous page's `next_cursor` names. The
    /// cursor is bound to `(namespace, filter, sort)`; under a different
    /// query it is refused.
    #[serde(default)]
    pub cursor: Option<String>,
    /// Maximum rows in the page. Zero is malformed.
    pub limit: u32,
}

/// One page of the list.
#[derive(Serialize, Deserialize, ts_rs::TS, Clone, Debug, PartialEq, Eq)]
pub struct WorkflowListPage {
    /// The rows, in the requested order. Exactly `limit` unless exhausted.
    pub items: Vec<WorkflowSummary>,
    /// Pass back as `cursor` for the next page; `None` when exhausted.
    pub next_cursor: Option<String>,
    /// How many rows match the filter in total.
    pub count: u64,
    /// What the serving install says about itself (ADR-016): the count a
    /// reader holds an UNATTRIBUTED `current_worker` against. Stamped by the
    /// server at read time; a page built by the engine alone carries `None`,
    /// which the server replaces before the page leaves it.
    ///
    /// `None` on a page a CLIENT decoded is "provenance not reported" — a
    /// server that predates the field — and is kept distinct from `Some(0)`:
    /// zero is a measurement the install made, absence is one nobody made.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub provenance: Option<crate::ReadProvenance>,
}

/// The table of text-rule cases the ops console mirrors, and the gate that
/// emits it. Kept in its own file so this one stays inside the line cap.
#[cfg(test)]
#[path = "listing_text_cases.rs"]
mod listing_text_cases;

#[cfg(test)]
mod tests {
    use chrono::{DateTime, Utc};

    use super::{WorkflowKind, WorkflowListFilter};
    use crate::{RunId, WorkflowId, WorkflowStatus, WorkflowSummary};

    fn summary(display_name: Option<&str>, kind: Option<&str>) -> WorkflowSummary {
        WorkflowSummary {
            workflow_id: WorkflowId::new(uuid::Uuid::from_u128(
                0xabc0_0000_0000_0000_0000_0000_0000_0001,
            )),
            run_id: RunId::new_v4(),
            workflow_type: String::from("checkout"),
            status: WorkflowStatus::Running,
            started_at: DateTime::<Utc>::default(),
            updated_at: DateTime::<Utc>::default() + chrono::Duration::seconds(5),
            ended_at: None,
            parent: None,
            failed_step: None,
            failure_reason: None,
            display_name: display_name.map(str::to_owned),
            kind: kind.map(str::to_owned),
            current_worker: None,
            package_version: None,
        }
    }

    #[test]
    fn empty_filter_matches_everything() {
        assert!(WorkflowListFilter::default().matches(&summary(None, None)));
    }

    #[test]
    fn kind_predicate_reads_the_recorded_attribute() {
        assert!(WorkflowKind::Workflow.matches(None));
        assert!(!WorkflowKind::Workflow.matches(Some("workloop")));
        assert!(WorkflowKind::Workloop.matches(Some("workloop")));
        assert!(!WorkflowKind::Workloop.matches(None));
    }

    #[test]
    fn text_matches_display_name_substring_case_insensitively() {
        let filter = WorkflowListFilter {
            text: Some(String::from("  NIGHTLY ")),
            ..WorkflowListFilter::default()
        };
        assert!(filter.matches(&summary(Some("the nightly build"), None)));
        assert!(!filter.matches(&summary(Some("weekly build"), None)));
        assert!(!filter.matches(&summary(None, None)));
    }

    #[test]
    fn text_matches_workflow_id_prefix() {
        let filter = WorkflowListFilter {
            text: Some(String::from("ABC00000")),
            ..WorkflowListFilter::default()
        };
        assert!(filter.matches(&summary(None, None)));
        let miss = WorkflowListFilter {
            text: Some(String::from("bc00000")),
            ..WorkflowListFilter::default()
        };
        assert!(
            !miss.matches(&summary(None, None)),
            "a prefix, not a substring"
        );
    }

    #[test]
    fn internal_types_hide_unless_named() {
        let mut internal = summary(None, None);
        internal.workflow_type = String::from("aion.schedule_coordinator");
        assert!(!WorkflowListFilter::default().matches(&internal));
        let named = WorkflowListFilter {
            workflow_types: vec![String::from("aion.schedule_coordinator")],
            ..WorkflowListFilter::default()
        };
        assert!(named.matches(&internal));
        assert!(!named.matches(&summary(None, None)));
    }

    #[test]
    fn blank_text_is_absent() {
        let filter = WorkflowListFilter {
            text: Some(String::from("   ")),
            ..WorkflowListFilter::default()
        };
        assert_eq!(filter.text_needle(), None);
        assert!(filter.matches(&summary(None, None)));
    }

    #[test]
    fn list_predicates_are_any_of_and_bounds_are_inclusive() {
        let row = summary(None, None);
        let filter = WorkflowListFilter {
            workflow_types: vec![String::from("other"), String::from("checkout")],
            statuses: vec![WorkflowStatus::Completed, WorkflowStatus::Running],
            started_after: Some(row.started_at),
            started_before: Some(row.started_at),
            updated_after: Some(row.updated_at),
            updated_before: Some(row.updated_at),
            ..WorkflowListFilter::default()
        };
        assert!(filter.matches(&row));
        let excluded = WorkflowListFilter {
            statuses: vec![WorkflowStatus::Completed],
            ..WorkflowListFilter::default()
        };
        assert!(!excluded.matches(&row));
        let too_late = WorkflowListFilter {
            updated_after: Some(row.updated_at + chrono::Duration::seconds(1)),
            ..WorkflowListFilter::default()
        };
        assert!(!too_late.matches(&row));
    }
}