aion-store 0.30.0

//! The visibility-store conformance scenarios: the list contract, pinned for
//! every backend by the same rows and the same requests.
//!
//! Each scenario seeds one namespace (plus a foreign one that must never
//! leak), then reads pages through [`VisibilityStore::list_workflows`] and
//! checks them against the contract in [`aion_core::listing`]: order per
//! field in both directions, keyset cursors that stay stable while rows are
//! inserted around them, filters applied before the limit, `count` equal to
//! the filtered total, and the bound-cursor refusal.

use std::collections::HashMap;
use std::sync::Arc;

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

use super::expect_eq;
use crate::StoreError;
use crate::visibility::{VisibilityRecord, VisibilityStore};

const NAMESPACE: &str = "conformance";
const FOREIGN_NAMESPACE: &str = "someone-else";

fn epoch() -> DateTime<Utc> {
    DateTime::<Utc>::default()
}

fn id(n: u128) -> WorkflowId {
    WorkflowId::new(uuid::Uuid::from_u128(n))
}

/// A row with every sortable field derived from `n`, so every scenario can
/// predict order from the seed alone.
fn row(n: u128, namespace: &str) -> VisibilityRecord {
    let n_i64 = i64::try_from(n).unwrap_or(i64::MAX);
    let started = epoch() + Duration::seconds(n_i64 * 60);
    let terminal = n.is_multiple_of(3);
    VisibilityRecord {
        namespace: namespace.to_owned(),
        workflow_id: id(n),
        run_id: RunId::new(uuid::Uuid::from_u128(1000 + n)),
        workflow_type: format!("type_{}", n % 4),
        status: if terminal {
            WorkflowStatus::Completed
        } else {
            WorkflowStatus::Running
        },
        started_at: started,
        // Reverse the start order so started_at and updated_at disagree.
        updated_at: epoch() + Duration::seconds((100 - n_i64) * 60),
        ended_at: terminal.then(|| started + Duration::seconds(30)),
        parent: n.is_multiple_of(5).then(|| id(n / 5)),
        display_name: Some(format!(
            "Run {}",
            (b'a' + u8::try_from(n % 26).unwrap_or(0)) as char
        )),
        kind: n
            .is_multiple_of(7)
            .then(|| String::from(aion_core::WORKLOOP_KIND)),
        failed_step: None,
        failure_reason: None,
        search_attributes: HashMap::new(),
        outstanding_leases: Vec::new(),
        package_version: None,
    }
}

async fn seed(store: &Arc<dyn VisibilityStore>, count: u128) -> Result<(), StoreError> {
    for n in 1..=count {
        store.record_visibility(row(n, NAMESPACE)).await?;
    }
    // Rows in another namespace: a namespace leak would surface them.
    // `(workflow_id, run_id)` is globally unique, so they get their own ids.
    for n in 501..=503 {
        store.record_visibility(row(n, FOREIGN_NAMESPACE)).await?;
    }
    Ok(())
}

fn list_request(sort: WorkflowSort, limit: u32) -> WorkflowListRequest {
    WorkflowListRequest {
        namespace: NAMESPACE.to_owned(),
        filter: WorkflowListFilter::default(),
        sort,
        cursor: None,
        limit,
    }
}

const fn sort(field: WorkflowSortField, direction: SortDirection) -> WorkflowSort {
    WorkflowSort { field, direction }
}

/// Walk every page of `request` and return the ids in delivery order.
async fn walk(
    store: &Arc<dyn VisibilityStore>,
    mut request: WorkflowListRequest,
) -> Result<(Vec<WorkflowId>, u64), StoreError> {
    let mut ids = Vec::new();
    let first = store.list_workflows(&request).await?;
    let count = first.count;
    let mut page = first;
    loop {
        let exhausted = page.next_cursor.is_none();
        let page_len = page.items.len();
        ids.extend(page.items.into_iter().map(|record| record.workflow_id));
        if exhausted {
            break;
        }
        if page_len != usize::try_from(request.limit).unwrap_or(usize::MAX) {
            return Err(StoreError::Backend(format!(
                "a page with a next_cursor must be full: got {page_len} of {}",
                request.limit
            )));
        }
        request.cursor = page.next_cursor;
        page = store.list_workflows(&request).await?;
    }
    Ok((ids, count))
}

pub(super) async fn every_sort_field_orders_both_directions_with_id_tiebreak(
    store: Arc<dyn VisibilityStore>,
) -> Result<(), StoreError> {
    seed(&store, 12).await?;
    for field in WorkflowSortField::ALL {
        let (asc, asc_count) =
            walk(&store, list_request(sort(field, SortDirection::Asc), 5)).await?;
        let (desc, desc_count) =
            walk(&store, list_request(sort(field, SortDirection::Desc), 5)).await?;
        expect_eq(asc.len(), 12, "ascending walk delivers every row once")?;
        expect_eq(asc_count, 12, "count is the namespace total with no filter")?;
        expect_eq(desc_count, 12, "count is direction-independent")?;

        let mut expected_asc: Vec<VisibilityRecord> = (1..=12).map(|n| row(n, NAMESPACE)).collect();
        expected_asc.sort_by(|left, right| {
            crate::visibility::ordering::page_key(left, sort(field, SortDirection::Asc)).cmp(
                &crate::visibility::ordering::page_key(right, sort(field, SortDirection::Asc)),
            )
        });
        let expected_asc: Vec<WorkflowId> =
            expected_asc.into_iter().map(|r| r.workflow_id).collect();
        expect_eq(
            asc.clone(),
            expected_asc,
            &format!("{field:?} ascending follows the shared page-key order"),
        )?;

        // Descending is the ascending order reversed EXCEPT within a tie, where
        // the id tie-break stays ascending in both directions.
        let mut expected_desc: Vec<VisibilityRecord> =
            (1..=12).map(|n| row(n, NAMESPACE)).collect();
        expected_desc.sort_by(|left, right| {
            crate::visibility::ordering::page_key(left, sort(field, SortDirection::Desc)).cmp(
                &crate::visibility::ordering::page_key(right, sort(field, SortDirection::Desc)),
            )
        });
        let expected_desc: Vec<WorkflowId> =
            expected_desc.into_iter().map(|r| r.workflow_id).collect();
        expect_eq(
            desc,
            expected_desc,
            &format!("{field:?} descending follows the shared page-key order"),
        )?;
    }
    Ok(())
}

pub(super) async fn ties_break_on_workflow_id_ascending_in_both_directions(
    store: Arc<dyn VisibilityStore>,
) -> Result<(), StoreError> {
    // Four rows sharing one workflow type: the type sort is all ties.
    for n in [40_u128, 41, 42, 43] {
        let mut record = row(n, NAMESPACE);
        record.workflow_type = String::from("same");
        store.record_visibility(record).await?;
    }
    let field = WorkflowSortField::WorkflowType;
    let (asc, _) = walk(&store, list_request(sort(field, SortDirection::Asc), 10)).await?;
    let (desc, _) = walk(&store, list_request(sort(field, SortDirection::Desc), 10)).await?;
    let expected: Vec<WorkflowId> = [40, 41, 42, 43].into_iter().map(id).collect();
    expect_eq(asc, expected.clone(), "ties ascend by id when ascending")?;
    expect_eq(desc, expected, "ties STILL ascend by id when descending")
}

pub(super) async fn pages_never_overlap_or_skip_and_end_without_a_cursor(
    store: Arc<dyn VisibilityStore>,
) -> Result<(), StoreError> {
    seed(&store, 23).await?;
    let request = list_request(sort(WorkflowSortField::StartedAt, SortDirection::Desc), 7);
    let (ids, count) = walk(&store, request.clone()).await?;
    expect_eq(
        ids.len(),
        23,
        "every row delivered exactly once across pages",
    )?;
    let mut unique = ids.clone();
    unique.sort_by_key(ToString::to_string);
    unique.dedup();
    expect_eq(unique.len(), 23, "no row delivered twice")?;
    expect_eq(count, 23, "count is the total on every page")?;
    let last = store
        .list_workflows(&WorkflowListRequest {
            limit: 23,
            ..request
        })
        .await?;
    expect_eq(
        last.next_cursor,
        None,
        "a page that exhausts the range carries no cursor",
    )?;
    expect_eq(
        last.items.len(),
        23,
        "a limit covering the range returns it whole",
    )
}

pub(super) async fn cursor_is_stable_while_rows_are_inserted_around_it(
    store: Arc<dyn VisibilityStore>,
) -> Result<(), StoreError> {
    // Seed even ids, page once, then insert odd ids on both sides of the
    // cursor: rows AFTER the cursor appear on later pages, rows BEFORE it do
    // not resurface, and nothing is skipped or duplicated.
    for n in (2..=20).step_by(2) {
        store.record_visibility(row(n, NAMESPACE)).await?;
    }
    let mut request = list_request(sort(WorkflowSortField::StartedAt, SortDirection::Asc), 4);
    let first = store.list_workflows(&request).await?;
    let first_ids: Vec<WorkflowId> = first.items.iter().map(|r| r.workflow_id.clone()).collect();
    expect_eq(
        first_ids,
        [2, 4, 6, 8].into_iter().map(id).collect(),
        "first page",
    )?;
    for n in [1_u128, 3, 5, 7, 9, 11, 13, 15, 17, 19, 21] {
        store.record_visibility(row(n, NAMESPACE)).await?;
    }
    request.cursor = first.next_cursor;
    let (rest, count) = walk(&store, request).await?;
    let expected: Vec<WorkflowId> = [9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21]
        .into_iter()
        .map(id)
        .collect();
    expect_eq(
        rest,
        expected,
        "rows inserted after the cursor appear; rows before it do not",
    )?;
    expect_eq(count, 21, "count reflects the namespace as it is now")
}

pub(super) async fn filters_apply_before_the_limit_and_count_is_the_filtered_total(
    store: Arc<dyn VisibilityStore>,
) -> Result<(), StoreError> {
    seed(&store, 30).await?;
    let all: Vec<VisibilityRecord> = (1..=30).map(|n| row(n, NAMESPACE)).collect();

    let filter = WorkflowListFilter {
        statuses: vec![WorkflowStatus::Completed],
        workflow_types: vec![String::from("type_0"), String::from("type_3")],
        ..WorkflowListFilter::default()
    };
    let expected: Vec<WorkflowId> = all
        .iter()
        .filter(|r| filter.matches(&r.summary()))
        .map(|r| r.workflow_id.clone())
        .collect();
    let request = WorkflowListRequest {
        filter: filter.clone(),
        ..list_request(sort(WorkflowSortField::StartedAt, SortDirection::Asc), 2)
    };
    let (ids, count) = walk(&store, request).await?;
    expect_eq(
        ids,
        expected.clone(),
        "filtered rows arrive in order, pages full until exhausted",
    )?;
    expect_eq(
        count,
        u64::try_from(expected.len()).unwrap_or(u64::MAX),
        "count is the filtered total",
    )?;

    let kind_only = WorkflowListRequest {
        filter: WorkflowListFilter {
            kind: Some(WorkflowKind::Workloop),
            ..WorkflowListFilter::default()
        },
        ..list_request(sort(WorkflowSortField::UpdatedAt, SortDirection::Desc), 50)
    };
    let page = store.list_workflows(&kind_only).await?;
    // updated_at runs opposite to n, so descending updated_at is ascending n.
    let expected_loops: Vec<WorkflowId> = [7, 14, 21, 28].into_iter().map(id).collect();
    expect_eq(
        page.items
            .iter()
            .map(|r| r.workflow_id.clone())
            .collect::<Vec<_>>(),
        expected_loops,
        "kind filter over updated_at desc",
    )?;
    expect_eq(page.count, 4, "kind count")?;

    let text = WorkflowListRequest {
        filter: WorkflowListFilter {
            text: Some(String::from("run C")),
            ..WorkflowListFilter::default()
        },
        ..list_request(sort(WorkflowSortField::DisplayName, SortDirection::Asc), 50)
    };
    let page = store.list_workflows(&text).await?;
    expect_eq(
        page.items
            .iter()
            .map(|r| r.workflow_id.clone())
            .collect::<Vec<_>>(),
        vec![id(2), id(28)],
        "text matches display names case-insensitively ('Run c' is n = 2 and 28)",
    )?;

    let by_parent = WorkflowListRequest {
        filter: WorkflowListFilter {
            parent: Some(id(1)),
            ..WorkflowListFilter::default()
        },
        ..list_request(sort(WorkflowSortField::StartedAt, SortDirection::Asc), 50)
    };
    let page = store.list_workflows(&by_parent).await?;
    expect_eq(
        page.items
            .iter()
            .map(|r| r.workflow_id.clone())
            .collect::<Vec<_>>(),
        vec![id(5)],
        "parent filter",
    )
}

pub(super) async fn foreign_namespace_rows_never_leak(
    store: Arc<dyn VisibilityStore>,
) -> Result<(), StoreError> {
    seed(&store, 3).await?;
    let page = store
        .list_workflows(&list_request(
            sort(WorkflowSortField::StartedAt, SortDirection::Asc),
            10,
        ))
        .await?;
    expect_eq(page.count, 3, "only the requested namespace is counted")?;
    for record in &page.items {
        expect_eq(
            record.namespace.as_str(),
            NAMESPACE,
            "every row is in the requested namespace",
        )?;
    }
    let empty = store
        .list_workflows(&WorkflowListRequest {
            namespace: String::from("nobody"),
            ..list_request(sort(WorkflowSortField::StartedAt, SortDirection::Asc), 10)
        })
        .await?;
    expect_eq(empty.items.len(), 0, "an unknown namespace lists nothing")?;
    expect_eq(empty.count, 0, "and counts nothing")?;
    expect_eq(empty.next_cursor, None, "and has no cursor")
}

pub(super) async fn internal_types_hide_unless_named(
    store: Arc<dyn VisibilityStore>,
) -> Result<(), StoreError> {
    seed(&store, 2).await?;
    let mut internal = row(99, NAMESPACE);
    internal.workflow_type = String::from("aion.schedule_coordinator");
    store.record_visibility(internal).await?;
    let page = store
        .list_workflows(&list_request(
            sort(WorkflowSortField::StartedAt, SortDirection::Asc),
            10,
        ))
        .await?;
    expect_eq(
        page.count,
        2,
        "the coordinator is hidden from an unnamed list",
    )?;
    let named = store
        .list_workflows(&WorkflowListRequest {
            filter: WorkflowListFilter {
                workflow_types: vec![String::from("aion.schedule_coordinator")],
                ..WorkflowListFilter::default()
            },
            ..list_request(sort(WorkflowSortField::StartedAt, SortDirection::Asc), 10)
        })
        .await?;
    expect_eq(named.count, 1, "naming the type lists it")
}

pub(super) async fn a_row_moves_when_its_sort_key_changes(
    store: Arc<dyn VisibilityStore>,
) -> Result<(), StoreError> {
    seed(&store, 3).await?;
    let mut renamed = row(2, NAMESPACE);
    renamed.display_name = Some(String::from("zzz last"));
    renamed.status = WorkflowStatus::Failed;
    store.record_visibility(renamed.clone()).await?;
    let by_name = store
        .list_workflows(&list_request(
            sort(WorkflowSortField::DisplayName, SortDirection::Asc),
            10,
        ))
        .await?;
    expect_eq(
        by_name
            .items
            .iter()
            .map(|r| r.workflow_id.clone())
            .collect::<Vec<_>>(),
        vec![id(1), id(3), id(2)],
        "the renamed row sorts under its NEW name and is not also under the old one",
    )?;
    expect_eq(by_name.count, 3, "an upsert never duplicates a row")?;
    let failed = store
        .list_workflows(&WorkflowListRequest {
            filter: WorkflowListFilter {
                statuses: vec![WorkflowStatus::Failed],
                ..WorkflowListFilter::default()
            },
            ..list_request(sort(WorkflowSortField::Status, SortDirection::Asc), 10)
        })
        .await?;
    expect_eq(
        failed.items,
        vec![renamed],
        "the moved row reads back whole",
    )?;
    let fetched = store.get_visibility(&id(2)).await?;
    expect_eq(
        fetched.map(|r| r.status),
        Some(WorkflowStatus::Failed),
        "get reads the latest row",
    )
}

/// The one-row collapse: a successor generation's upsert REPLACES its
/// predecessor's row, so a workflow that continued as new is one list row —
/// current generation — never one per generation.
pub(super) async fn a_successor_generation_replaces_its_predecessors_row(
    store: Arc<dyn VisibilityStore>,
) -> Result<(), StoreError> {
    let predecessor = row(1, NAMESPACE);
    store.record_visibility(predecessor.clone()).await?;
    let mut successor = row(1, NAMESPACE);
    successor.run_id = RunId::new(uuid::Uuid::from_u128(2001));
    successor.status = WorkflowStatus::Running;
    store.record_visibility(successor.clone()).await?;
    let page = store
        .list_workflows(&list_request(
            sort(WorkflowSortField::StartedAt, SortDirection::Asc),
            10,
        ))
        .await?;
    expect_eq(
        page.items
            .iter()
            .map(|r| r.run_id.clone())
            .collect::<Vec<_>>(),
        vec![successor.run_id.clone()],
        "one row per workflow, carrying the current generation",
    )?;
    expect_eq(
        page.count,
        1,
        "the predecessor's row is gone from the count",
    )?;
    let fetched = store.get_visibility(&id(1)).await?;
    expect_eq(
        fetched.map(|r| r.run_id),
        Some(successor.run_id),
        "get answers the current generation",
    )
}

/// The prune's run guard: removing a superseded generation deletes nothing
/// once the successor's row stands; removing the current generation deletes
/// the workflow's row.
pub(super) async fn remove_visibility_prunes_only_the_named_generation(
    store: Arc<dyn VisibilityStore>,
) -> Result<(), StoreError> {
    let predecessor_run = RunId::new(uuid::Uuid::from_u128(1001));
    store.record_visibility(row(1, NAMESPACE)).await?;
    let mut successor = row(1, NAMESPACE);
    successor.run_id = RunId::new(uuid::Uuid::from_u128(2001));
    store.record_visibility(successor.clone()).await?;
    let removed = store.remove_visibility(&id(1), &predecessor_run).await?;
    expect_eq(
        removed,
        false,
        "a superseded generation's prune must not touch the current row",
    )?;
    expect_eq(
        store.get_visibility(&id(1)).await?.map(|r| r.run_id),
        Some(successor.run_id.clone()),
        "the current row survives the stale prune",
    )?;
    let removed = store.remove_visibility(&id(1), &successor.run_id).await?;
    expect_eq(
        removed,
        true,
        "the current generation's prune removes the row",
    )?;
    expect_eq(
        store.get_visibility(&id(1)).await?,
        None,
        "the row is gone after its own prune",
    )
}

pub(super) async fn invalid_queries_are_refused_typed(
    store: Arc<dyn VisibilityStore>,
) -> Result<(), StoreError> {
    seed(&store, 5).await?;
    let base = list_request(sort(WorkflowSortField::StartedAt, SortDirection::Asc), 2);
    let zero = store
        .list_workflows(&WorkflowListRequest {
            limit: 0,
            ..base.clone()
        })
        .await;
    if !matches!(zero, Err(StoreError::InvalidQuery(_))) {
        return Err(StoreError::Backend(format!(
            "limit 0 must be InvalidQuery, got {zero:?}"
        )));
    }
    let first = store.list_workflows(&base).await?;
    let replayed = store
        .list_workflows(&WorkflowListRequest {
            sort: sort(WorkflowSortField::StartedAt, SortDirection::Desc),
            cursor: first.next_cursor.clone(),
            ..base.clone()
        })
        .await;
    if !matches!(replayed, Err(StoreError::InvalidQuery(_))) {
        return Err(StoreError::Backend(format!(
            "a cursor replayed under another sort must be InvalidQuery, got {replayed:?}"
        )));
    }
    let garbage = store
        .list_workflows(&WorkflowListRequest {
            cursor: Some(String::from("not a cursor")),
            ..base
        })
        .await;
    if !matches!(garbage, Err(StoreError::InvalidQuery(_))) {
        return Err(StoreError::Backend(format!(
            "garbage cursor must be InvalidQuery, got {garbage:?}"
        )));
    }
    Ok(())
}