aion-store 0.30.0

//! The sort-key bytes every backend orders by.
//!
//! A page is defined by byte order over `sort_key(row) ++ workflow_id`, so the
//! in-memory store sorts by these bytes and the haematite backend stores them
//! as index keys — and the same request yields the same page on both. The
//! encodings are chosen so that plain bytewise comparison IS the intended
//! order:
//!
//! * instants → sign-flipped big-endian microseconds; an absent instant is
//!   all zeros, the smallest value;
//! * strings → their bytes plus a `0x00` terminator, so a string never sorts
//!   after its own extension (`a` < `ab`) in either direction;
//! * `display_name` is lowercased first (case-insensitive order); an unnamed
//!   run sorts as the empty string;
//! * `status` orders by its canonical name;
//! * descending order is the bitwise complement of the ascending key, because
//!   a forward-only range read over complemented keys IS the reverse order.
//!
//! The workflow id is appended raw (16 bytes) as the total-order tie-break
//! and is NEVER complemented: ties break on `workflow_id` ascending in both
//! directions, as the contract states.

use aion_core::{SortDirection, WorkflowSort, WorkflowSortField, WorkflowStatus};
use chrono::{DateTime, Utc};

use super::VisibilityRecord;

/// The complete page-order key of a row: the sort key under `sort`, then the
/// raw workflow id. Bytewise order over these keys is the page order.
#[must_use]
pub fn page_key(record: &VisibilityRecord, sort: WorkflowSort) -> Vec<u8> {
    let mut key = sort_key(record, sort);
    key.extend_from_slice(record.workflow_id.as_uuid().as_bytes());
    key
}

/// The sort-key bytes of a row under `sort`, without the tie-break.
#[must_use]
pub fn sort_key(record: &VisibilityRecord, sort: WorkflowSort) -> Vec<u8> {
    let ascending = match sort.field {
        WorkflowSortField::StartedAt => encode_instant(Some(record.started_at)),
        WorkflowSortField::UpdatedAt => encode_instant(Some(record.updated_at)),
        WorkflowSortField::EndedAt => encode_instant(record.ended_at),
        WorkflowSortField::WorkflowType => encode_text(&record.workflow_type),
        WorkflowSortField::Status => encode_text(status_name(record.status)),
        WorkflowSortField::DisplayName => {
            encode_text(&record.display_name.as_deref().unwrap_or("").to_lowercase())
        }
    };
    match sort.direction {
        SortDirection::Asc => ascending,
        SortDirection::Desc => ascending.into_iter().map(|byte| !byte).collect(),
    }
}

/// Eight bytes that order like the instant. Microsecond precision matches
/// what the store round-trips; the sign flip makes the two's-complement
/// integer order bytewise.
#[must_use]
pub fn encode_instant(instant: Option<DateTime<Utc>>) -> Vec<u8> {
    let micros = instant.map_or(i64::MIN, |instant| instant.timestamp_micros());
    // Flipping the sign bit maps i64 order onto u64 order; reinterpreting
    // the bytes (not casting) keeps the bit pattern exactly.
    let ordered = u64::from_ne_bytes((micros ^ i64::MIN).to_ne_bytes());
    ordered.to_be_bytes().to_vec()
}

/// A string's bytes plus a NUL terminator.
#[must_use]
pub fn encode_text(text: &str) -> Vec<u8> {
    let mut bytes = Vec::with_capacity(text.len() + 1);
    bytes.extend_from_slice(text.as_bytes());
    bytes.push(0);
    bytes
}

/// The canonical name a status sorts by.
#[must_use]
pub const fn status_name(status: WorkflowStatus) -> &'static str {
    match status {
        WorkflowStatus::Running => "running",
        WorkflowStatus::Completed => "completed",
        WorkflowStatus::Failed => "failed",
        WorkflowStatus::Cancelled => "cancelled",
        WorkflowStatus::TimedOut => "timed_out",
        WorkflowStatus::ContinuedAsNew => "continued_as_new",
        WorkflowStatus::Paused => "paused",
    }
}

#[cfg(test)]
mod tests {
    use std::collections::HashMap;

    use aion_core::{
        RunId, SortDirection, WorkflowId, WorkflowSort, WorkflowSortField, WorkflowStatus,
    };
    use chrono::{DateTime, Utc};

    use super::{encode_instant, encode_text, page_key, sort_key};
    use crate::visibility::VisibilityRecord;

    fn record(id: u128, display_name: Option<&str>, ended: Option<i64>) -> VisibilityRecord {
        let started = DateTime::<Utc>::default();
        VisibilityRecord {
            namespace: String::from("default"),
            workflow_id: WorkflowId::new(uuid::Uuid::from_u128(id)),
            run_id: RunId::new(uuid::Uuid::from_u128(id + 1)),
            workflow_type: String::from("checkout"),
            status: WorkflowStatus::Running,
            started_at: started,
            updated_at: started,
            ended_at: ended.map(|seconds| started + chrono::Duration::seconds(seconds)),
            parent: None,
            display_name: display_name.map(str::to_owned),
            kind: None,
            failed_step: None,
            failure_reason: None,
            search_attributes: HashMap::new(),
            outstanding_leases: Vec::new(),
            package_version: None,
        }
    }

    #[test]
    fn instants_order_bytewise_including_negative_and_absent() {
        let epoch = DateTime::<Utc>::default();
        let before = epoch - chrono::Duration::days(1);
        let after = epoch + chrono::Duration::microseconds(1);
        assert!(encode_instant(None) < encode_instant(Some(before)));
        assert!(encode_instant(Some(before)) < encode_instant(Some(epoch)));
        assert!(encode_instant(Some(epoch)) < encode_instant(Some(after)));
    }

    #[test]
    fn a_string_sorts_before_its_extension_in_both_directions() {
        assert!(encode_text("a") < encode_text("ab"));
        let desc = |text: &str| {
            encode_text(text)
                .into_iter()
                .map(|b| !b)
                .collect::<Vec<_>>()
        };
        assert!(desc("ab") < desc("a"));
    }

    #[test]
    fn descending_key_reverses_ascending_order_and_keeps_id_tiebreak() {
        let early = record(1, None, Some(1));
        let late = record(2, None, Some(5));
        let asc = WorkflowSort {
            field: WorkflowSortField::EndedAt,
            direction: SortDirection::Asc,
        };
        let desc = WorkflowSort {
            field: WorkflowSortField::EndedAt,
            direction: SortDirection::Desc,
        };
        assert!(page_key(&early, asc) < page_key(&late, asc));
        assert!(page_key(&late, desc) < page_key(&early, desc));

        let twin_a = record(1, None, Some(1));
        let twin_b = record(2, None, Some(1));
        assert!(
            page_key(&twin_a, desc) < page_key(&twin_b, desc),
            "ties break on id ascending even descending"
        );
    }

    #[test]
    fn running_sorts_first_ascending_by_ended_at() {
        let running = record(9, None, None);
        let ended = record(1, None, Some(1));
        let asc = WorkflowSort {
            field: WorkflowSortField::EndedAt,
            direction: SortDirection::Asc,
        };
        assert!(sort_key(&running, asc) < sort_key(&ended, asc));
    }

    #[test]
    fn display_name_orders_case_insensitively_and_unnamed_first() {
        let sort = WorkflowSort {
            field: WorkflowSortField::DisplayName,
            direction: SortDirection::Asc,
        };
        assert_eq!(
            sort_key(&record(1, Some("Beta"), None), sort),
            sort_key(&record(1, Some("beta"), None), sort)
        );
        assert!(
            sort_key(&record(1, None, None), sort)
                < sort_key(&record(1, Some("alpha"), None), sort)
        );
        assert!(
            sort_key(&record(1, Some("alpha"), None), sort)
                < sort_key(&record(1, Some("Beta"), None), sort)
        );
    }
}