aion-store 0.30.0

//! The visibility row, the store contract, and the page it answers with.

use std::collections::HashMap;

use aion_core::{
    OutstandingLease, PackageVersion, RunId, SearchAttributeValue, WorkflowId, WorkflowListRequest,
    WorkflowStatus, WorkflowSummary,
};
use async_trait::async_trait;
use chrono::{DateTime, Utc};
use serde::{Deserialize, Serialize};

use crate::StoreError;

/// The visibility store: rows in, pages out.
///
/// ONE ROW PER WORKFLOW IDENTITY. A workflow that continues as new keeps one
/// history and one row: the successor generation's upsert REPLACES the
/// predecessor's row in place, so a workloop is a single row for its whole
/// life and a list never shows generations. `run_id` stays on the row as
/// data — the current generation — and the generation chain remains readable
/// from history under describe.
///
/// `record_visibility` is the ONLY write; the engine's Recorder calls it after
/// every durable append it projects (lifecycle events re-project the whole
/// row, every other append advances `updated_at`). `list_workflows` answers
/// the list contract from rows alone — a backend never reads history here.
#[async_trait]
pub trait VisibilityStore: Send + Sync + 'static {
    /// Upserts THE row for `record.workflow_id`, replacing any previous
    /// generation's row and moving it under its new sort keys atomically.
    /// The replace is part of the write, not a separate pass: after this
    /// returns, the workflow has exactly one row and it is this one.
    ///
    /// # Errors
    ///
    /// Returns backend or serialization errors encountered while writing.
    async fn record_visibility(&self, record: VisibilityRecord) -> Result<(), StoreError>;

    /// Reads the workflow's row, or `None` when it has no row yet.
    ///
    /// # Errors
    ///
    /// Returns backend or serialization errors encountered while reading.
    async fn get_visibility(
        &self,
        workflow_id: &WorkflowId,
    ) -> Result<Option<VisibilityRecord>, StoreError>;

    /// Removes the row for `workflow_id` IF it belongs to `run_id`, and any
    /// row a store written before the one-row collapse still holds for that
    /// `(workflow_id, run_id)` pair. Returns whether anything was removed.
    ///
    /// The reconciler is the caller: it prunes every generation that is not
    /// the current one, which is how a pre-collapse store converges to one
    /// row per workflow at its next boot without a migration step. The run
    /// guard makes the prune safe against racing a newer generation's upsert:
    /// a remove naming a superseded run never deletes the current row.
    ///
    /// # Errors
    ///
    /// Returns backend or serialization errors encountered while removing.
    async fn remove_visibility(
        &self,
        workflow_id: &WorkflowId,
        run_id: &RunId,
    ) -> Result<bool, StoreError>;

    /// Answers one page of the list contract: `request.filter` applied
    /// BEFORE the limit, ordered by `request.sort` with `workflow_id` as the
    /// total-order tie-break, continuing after `request.cursor`, at most
    /// `request.limit` rows, with the filtered total.
    ///
    /// # Errors
    ///
    /// Returns [`StoreError::InvalidQuery`] for a zero limit or a cursor that
    /// was not minted under this exact `(namespace, filter, sort)`, and
    /// backend or serialization errors encountered while reading.
    async fn list_workflows(
        &self,
        request: &WorkflowListRequest,
    ) -> Result<VisibilityPage, StoreError>;
}

/// One page of rows, before the server turns them into wire summaries.
#[derive(Clone, Debug, PartialEq)]
pub struct VisibilityPage {
    /// The rows, in the requested order.
    pub items: Vec<VisibilityRecord>,
    /// The cursor naming the page's last row, present only when more rows
    /// follow it.
    pub next_cursor: Option<String>,
    /// How many rows match the filter in total, cursor disregarded.
    pub count: u64,
}

/// The complete row for one workflow execution.
///
/// Every field a list can sort or filter on is a first-class field here,
/// projected by the engine when the row is written; `search_attributes`
/// remains the full recorded attribute map for readers that need more.
#[derive(Serialize, Deserialize, Clone, Debug, PartialEq)]
pub struct VisibilityRecord {
    /// The namespace the run was placed in, from the `aion.namespace` start
    /// attribute (`aion_core::DEFAULT_NAMESPACE` for a run recorded with no
    /// placement).
    pub namespace: String,
    /// Logical workflow identifier.
    pub workflow_id: WorkflowId,
    /// Concrete run identifier for this workflow execution.
    pub run_id: RunId,
    /// Workflow type recorded when the execution started.
    pub workflow_type: String,
    /// Current projected status.
    pub status: WorkflowStatus,
    /// Recorded instant of the run's `WorkflowStarted`.
    pub started_at: DateTime<Utc>,
    /// Recorded instant of the run's latest event of any kind.
    pub updated_at: DateTime<Utc>,
    /// Recorded instant of the current lease's terminal event, if terminal.
    pub ended_at: Option<DateTime<Utc>>,
    /// The parent workflow, for a child execution.
    pub parent: Option<WorkflowId>,
    /// The operator-facing display name, from the `aion.display_name`
    /// attribute (last write wins).
    pub display_name: Option<String>,
    /// The document kind from the `aion.kind` attribute (`"workloop"`), or
    /// `None` for an ordinary workflow.
    pub kind: Option<String>,
    /// The step (activity type) that failed, only for a terminal-Failed run.
    pub failed_step: Option<String>,
    /// The terminal `WorkflowFailed` message, only for a terminal-Failed run.
    pub failure_reason: Option<String>,
    /// The full recorded search-attribute map.
    pub search_attributes: HashMap<String, SearchAttributeValue>,
    /// Every attempt a worker currently holds, earliest lease first — the
    /// same fold as [`aion_core::outstanding_leases`] over the run's current
    /// lease segment, maintained event by event by the engine's Recorder
    /// through [`aion_core::apply_lease_transition`] (WA-010 R4). The
    /// summary's `current_worker` is the last entry.
    ///
    /// Defaults to empty so a row written before the field existed decodes
    /// as UNATTRIBUTED rather than failing to decode.
    #[serde(default)]
    pub outstanding_leases: Vec<OutstandingLease>,
    /// The content hash of the package this run started under, from the run's
    /// own `WorkflowStarted` (see [`WorkflowSummary::package_version`]).
    ///
    /// Defaults to `None` so a row written before the field existed decodes
    /// as "not carried" rather than failing to decode — and is NEVER filled
    /// in by looking up the type's current version.
    #[serde(default)]
    pub package_version: Option<PackageVersion>,
}

impl VisibilityRecord {
    /// The wire summary this row projects to.
    #[must_use]
    pub fn summary(&self) -> WorkflowSummary {
        WorkflowSummary {
            workflow_id: self.workflow_id.clone(),
            run_id: self.run_id.clone(),
            workflow_type: self.workflow_type.clone(),
            status: self.status,
            started_at: self.started_at,
            updated_at: self.updated_at,
            ended_at: self.ended_at,
            parent: self.parent.clone(),
            failed_step: self.failed_step.clone(),
            failure_reason: self.failure_reason.clone(),
            display_name: self.display_name.clone(),
            kind: self.kind.clone(),
            current_worker: aion_core::current_worker(&self.outstanding_leases),
            package_version: self.package_version.clone(),
        }
    }
}

impl From<VisibilityRecord> for WorkflowSummary {
    fn from(record: VisibilityRecord) -> Self {
        Self {
            workflow_id: record.workflow_id,
            run_id: record.run_id,
            workflow_type: record.workflow_type,
            status: record.status,
            started_at: record.started_at,
            updated_at: record.updated_at,
            ended_at: record.ended_at,
            parent: record.parent,
            failed_step: record.failed_step,
            failure_reason: record.failure_reason,
            display_name: record.display_name,
            kind: record.kind,
            current_worker: aion_core::current_worker(&record.outstanding_leases),
            package_version: record.package_version,
        }
    }
}

#[cfg(test)]
mod tests {
    use std::sync::Arc;

    use aion_core::{PackageVersion, WorkflowSummary};

    use super::{VisibilityRecord, VisibilityStore};

    #[test]
    fn visibility_store_is_object_safe() {
        let _: Option<Arc<dyn VisibilityStore>> = None;
    }

    /// A row persisted before `package_version` (and `outstanding_leases`)
    /// existed decodes as "not carried": the summary reports `None`, and
    /// nothing fills it in from the type's current version. A row that
    /// carries the hash reports exactly that hash.
    #[test]
    fn a_row_written_before_the_package_field_decodes_as_not_carried()
    -> Result<(), Box<dyn std::error::Error>> {
        let legacy = serde_json::json!({
            "namespace": "default",
            "workflow_id": "00000000-0000-0000-0000-000000000001",
            "run_id": "00000000-0000-0000-0000-000000000002",
            "workflow_type": "checkout",
            "status": "Running",
            "started_at": "2023-11-14T22:13:21Z",
            "updated_at": "2023-11-14T22:13:21Z",
            "ended_at": null,
            "parent": null,
            "display_name": null,
            "kind": null,
            "failed_step": null,
            "failure_reason": null,
            "search_attributes": {}
        });
        let record: VisibilityRecord = serde_json::from_value(legacy.clone())?;
        assert_eq!(record.package_version, None);
        assert_eq!(record.summary().package_version, None);

        let mut carried = legacy;
        carried["package_version"] = serde_json::Value::String("c".repeat(64));
        let record: VisibilityRecord = serde_json::from_value(carried)?;
        assert_eq!(
            record.summary().package_version,
            Some(PackageVersion::new("c".repeat(64)))
        );
        assert_eq!(
            WorkflowSummary::from(record).package_version,
            Some(PackageVersion::new("c".repeat(64)))
        );
        Ok(())
    }
}