aion-store 0.30.0

//! The shared page fold: filter → order → cursor → limit → count.
//!
//! A backend hands this every row of the namespace it can reach cheaply
//! (the in-memory store: all of them; haematite: the rows its index range
//! read yields, already in key order) and gets back exactly the page the
//! contract promises. Keeping the fold in one place is what makes the two
//! backends answer identically — the conformance suite pins that.

use aion_core::WorkflowListRequest;

use super::cursor::WorkflowCursor;
use super::ordering::page_key;
use super::record::{VisibilityPage, VisibilityRecord};
use crate::StoreError;

/// A row with its page key already computed.
#[derive(Clone, Debug)]
pub struct PageCandidate {
    /// Bytewise order over this IS the page order.
    pub key: Vec<u8>,
    /// The row.
    pub record: VisibilityRecord,
}

impl PageCandidate {
    /// Compute the candidate for `record` under `request.sort`.
    #[must_use]
    pub fn new(record: VisibilityRecord, request: &WorkflowListRequest) -> Self {
        Self {
            key: page_key(&record, request.sort),
            record,
        }
    }
}

/// Validate `request`, decode its cursor, and fold `candidates` (every row in
/// the namespace, in ANY order) into the page.
///
/// The filter is applied to every candidate before anything else; `count` is
/// the number that pass it. Rows at or before the cursor are skipped, the
/// survivors are sorted by key, and the first `limit` become the page. A
/// `next_cursor` is minted only when at least one matching row follows the
/// page.
///
/// # Errors
///
/// [`StoreError::InvalidQuery`] for a zero `limit` or a cursor that was not
/// minted under this request; serialization errors from minting the cursor.
pub fn paginate(
    request: &WorkflowListRequest,
    candidates: impl IntoIterator<Item = PageCandidate>,
) -> Result<VisibilityPage, StoreError> {
    let limit = validated_limit(request)?;
    let cursor = WorkflowCursor::decode(request)?;

    let mut count: u64 = 0;
    let mut after_cursor = Vec::new();
    for candidate in candidates {
        if candidate.record.namespace != request.namespace
            || !request.filter.matches(&candidate.record.summary())
        {
            continue;
        }
        count += 1;
        if cursor
            .as_ref()
            .is_none_or(|cursor| candidate.key > cursor.after_key)
        {
            after_cursor.push(candidate);
        }
    }
    after_cursor.sort_by(|left, right| left.key.cmp(&right.key));

    let has_more = after_cursor.len() > limit;
    after_cursor.truncate(limit);
    let next_cursor = match (has_more, after_cursor.last()) {
        (true, Some(last)) => Some(WorkflowCursor::mint(request, &last.key)?),
        _ => None,
    };
    Ok(VisibilityPage {
        items: after_cursor
            .into_iter()
            .map(|candidate| candidate.record)
            .collect(),
        next_cursor,
        count,
    })
}

/// The request's limit as a `usize`, refusing zero: a page of nothing is a
/// malformed request, not an empty answer.
///
/// # Errors
///
/// [`StoreError::InvalidQuery`] when `limit` is zero.
pub fn validated_limit(request: &WorkflowListRequest) -> Result<usize, StoreError> {
    if request.limit == 0 {
        return Err(StoreError::InvalidQuery(String::from(
            "limit must be at least 1",
        )));
    }
    usize::try_from(request.limit)
        .map_err(|_| StoreError::InvalidQuery(String::from("limit exceeds this platform's range")))
}