aion-rs 0.29.0

//! The visibility projection: one row per workflow execution, written from
//! history.
//!
//! [`project_visibility`] is THE projector — the Recorder's post-append
//! upsert, the lifecycle handlers' explicit upserts, and boot/adoption
//! reconciliation all build their row here, so no two writers can disagree
//! about what a history projects to.

use std::sync::Arc;

use aion_core::{Event, RunId, WorkflowId, WorkflowSummary};
use aion_store::EventStore;
use aion_store::visibility::{VisibilityRecord, VisibilityStore};

use crate::EngineError;

/// Projects the visibility row for `run_id` from `history`, or `None` when
/// the history holds no `WorkflowStarted` and therefore has nothing to show.
///
/// Every summary field comes from [`WorkflowSummary::from_history`] — the
/// same fold the wire summary uses — so a row and the summary a reader
/// would build from history can never drift. The namespace is the
/// `aion.namespace` start attribute, or [`aion_core::DEFAULT_NAMESPACE`] for
/// a run recorded with no placement — the same fold recovery routes by.
#[must_use]
pub fn project_visibility(history: &[Event], run_id: &RunId) -> Option<VisibilityRecord> {
    let summary = WorkflowSummary::from_history(history)?;
    let search_attributes = aion_core::search_attributes_from_events(history);
    let namespace = aion_core::namespace_from_attributes(&search_attributes);
    Some(VisibilityRecord {
        namespace,
        workflow_id: summary.workflow_id,
        run_id: run_id.clone(),
        workflow_type: summary.workflow_type,
        status: summary.status,
        started_at: summary.started_at,
        updated_at: summary.updated_at,
        ended_at: summary.ended_at,
        parent: summary.parent,
        display_name: summary.display_name,
        kind: summary.kind,
        failed_step: summary.failed_step,
        failure_reason: summary.failure_reason,
        search_attributes,
        outstanding_leases: aion_core::outstanding_leases(history),
        package_version: summary.package_version,
    })
}

/// The run a history currently belongs to: its latest `WorkflowStarted`.
#[must_use]
pub fn current_run_id(history: &[Event]) -> Option<RunId> {
    history.iter().rev().find_map(|event| match event {
        Event::WorkflowStarted { run_id, .. } => Some(run_id.clone()),
        _ => None,
    })
}

/// Rebuilds and upserts the full visibility row for a workflow execution.
///
/// # Errors
///
/// Returns store errors when history cannot be read or the row cannot be
/// written, and a load error if the history has no `WorkflowStarted` to
/// project.
pub async fn upsert_workflow_visibility(
    event_store: Arc<dyn EventStore>,
    visibility_store: Arc<dyn VisibilityStore>,
    workflow_id: &WorkflowId,
    run_id: &RunId,
) -> Result<(), EngineError> {
    let history = event_store.read_history(workflow_id).await?;
    let record = project_visibility(&history, run_id).ok_or_else(|| EngineError::Load {
        reason: format!(
            "workflow `{workflow_id}` history has no WorkflowStarted event for visibility projection"
        ),
    })?;
    visibility_store.record_visibility(record).await?;
    Ok(())
}

/// Reconciles every visibility row with authoritative event history.
///
/// Boot, shard adoption, and the periodic repair loop call this; the read
/// path never does. Each workflow's history is projected and compared with
/// its stored row by key, so a consistent store costs one read per row and
/// no list.
///
/// # Errors
///
/// Returns store errors while reading histories or rows, and load errors for
/// a history with no `WorkflowStarted` to project.
pub async fn reconcile_visibility(
    event_store: Arc<dyn EventStore>,
    visibility_store: Arc<dyn VisibilityStore>,
) -> Result<(), EngineError> {
    for workflow_id in event_store.list_workflow_ids().await? {
        let history = event_store.read_history(&workflow_id).await?;
        let run_id = current_run_id(&history).ok_or_else(|| EngineError::Load {
            reason: format!(
                "workflow `{workflow_id}` history has no WorkflowStarted event for visibility \
                 projection"
            ),
        })?;
        let Some(projected) = project_visibility(&history, &run_id) else {
            return Err(EngineError::Load {
                reason: format!(
                    "workflow `{workflow_id}` history has no WorkflowStarted event for \
                     visibility projection"
                ),
            });
        };
        let stored = visibility_store
            .get_visibility(&workflow_id, &run_id)
            .await?;
        if stored.as_ref() != Some(&projected) {
            visibility_store.record_visibility(projected).await?;
        }
    }
    Ok(())
}

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

    use aion_core::{
        DISPLAY_NAME_ATTRIBUTE, Event, EventEnvelope, NAMESPACE_ATTRIBUTE, PackageVersion, Payload,
        RunId, SearchAttributeValue, WorkflowError, WorkflowId, WorkflowStatus,
    };
    use aion_store::visibility::VisibilityStore;
    use aion_store::{EventStore, InMemoryStore, WritableEventStore, WriteToken};
    use chrono::{TimeZone, Utc};

    use super::{current_run_id, project_visibility, reconcile_visibility};

    type TestResult = Result<(), Box<dyn Error>>;

    fn envelope(workflow_id: &WorkflowId, seq: u64) -> Result<EventEnvelope, Box<dyn Error>> {
        let base = Utc
            .with_ymd_and_hms(2026, 1, 1, 0, 0, 0)
            .single()
            .ok_or("test timestamp should be unambiguous")?;
        Ok(EventEnvelope {
            seq,
            recorded_at: base + chrono::Duration::seconds(i64::try_from(seq)?),
            workflow_id: workflow_id.clone(),
        })
    }

    fn payload() -> Result<Payload, Box<dyn Error>> {
        Ok(Payload::from_json(&serde_json::json!({}))?)
    }

    fn workflow_started(
        workflow_id: &WorkflowId,
        run_id: &RunId,
        parent: Option<WorkflowId>,
    ) -> Result<Event, Box<dyn Error>> {
        Ok(Event::WorkflowStarted {
            envelope: envelope(workflow_id, 1)?,
            workflow_type: String::from("order_processing"),
            input: payload()?,
            run_id: run_id.clone(),
            parent_run_id: None,
            parent_workflow_id: parent,
            package_version: PackageVersion::new("a".repeat(64)),
        })
    }

    fn attributes(
        workflow_id: &WorkflowId,
        seq: u64,
        pairs: &[(&str, &str)],
    ) -> Result<Event, Box<dyn Error>> {
        Ok(Event::SearchAttributesUpdated {
            envelope: envelope(workflow_id, seq)?,
            workflow_id: workflow_id.clone(),
            attributes: pairs
                .iter()
                .map(|(key, value)| {
                    (
                        (*key).to_owned(),
                        SearchAttributeValue::String((*value).to_owned()),
                    )
                })
                .collect::<HashMap<_, _>>(),
        })
    }

    #[test]
    fn a_history_without_a_start_projects_nothing() -> TestResult {
        let wf_id = WorkflowId::new_v4();
        let orphan = vec![Event::WorkflowCompleted {
            envelope: envelope(&wf_id, 1)?,
            result: payload()?,
        }];
        assert!(project_visibility(&[], &RunId::new_v4()).is_none());
        assert!(project_visibility(&orphan, &RunId::new_v4()).is_none());
        assert!(current_run_id(&orphan).is_none());
        Ok(())
    }

    #[test]
    fn a_running_history_projects_every_field() -> TestResult {
        let wf_id = WorkflowId::new_v4();
        let run_id = RunId::new_v4();
        let parent = WorkflowId::new_v4();
        let history = vec![
            workflow_started(&wf_id, &run_id, Some(parent.clone()))?,
            attributes(
                &wf_id,
                2,
                &[
                    (NAMESPACE_ATTRIBUTE, "tenant-a"),
                    (DISPLAY_NAME_ATTRIBUTE, "Nightly close"),
                    ("region", "eu-west-1"),
                ],
            )?,
            Event::SignalReceived {
                envelope: envelope(&wf_id, 3)?,
                name: String::from("wake"),
                payload: payload()?,
            },
        ];

        let record = project_visibility(&history, &run_id).ok_or("a started history projects")?;
        assert_eq!(record.namespace, "tenant-a");
        assert_eq!(record.workflow_id, wf_id);
        assert_eq!(record.run_id, run_id);
        assert_eq!(record.workflow_type, "order_processing");
        assert_eq!(record.status, WorkflowStatus::Running);
        assert_eq!(record.started_at, envelope(&wf_id, 1)?.recorded_at);
        assert_eq!(
            record.updated_at,
            envelope(&wf_id, 3)?.recorded_at,
            "updated_at is the LAST event of any kind, not the last lifecycle event"
        );
        assert_eq!(record.ended_at, None);
        assert_eq!(record.parent, Some(parent));
        assert_eq!(record.display_name.as_deref(), Some("Nightly close"));
        assert_eq!(record.kind, None);
        assert_eq!(record.failed_step, None);
        assert_eq!(record.failure_reason, None);
        assert_eq!(
            record.package_version,
            Some(PackageVersion::new("a".repeat(64))),
            "the row carries the hash of the package the run started under"
        );
        assert_eq!(
            record.search_attributes.get("region"),
            Some(&SearchAttributeValue::String(String::from("eu-west-1")))
        );
        assert_eq!(current_run_id(&history), Some(run_id));
        Ok(())
    }

    #[test]
    fn a_failed_history_projects_the_terminal_and_an_unplaced_run_is_in_the_default_namespace()
    -> TestResult {
        let wf_id = WorkflowId::new_v4();
        let run_id = RunId::new_v4();
        let terminal = envelope(&wf_id, 2)?;
        let ended = terminal.recorded_at;
        let history = vec![
            workflow_started(&wf_id, &run_id, None)?,
            Event::WorkflowFailed {
                envelope: terminal,
                error: WorkflowError {
                    message: String::from("boom"),
                    details: None,
                },
            },
        ];
        let record = project_visibility(&history, &run_id).ok_or("a started history projects")?;
        assert_eq!(record.namespace, aion_core::DEFAULT_NAMESPACE);
        assert_eq!(record.status, WorkflowStatus::Failed);
        assert_eq!(record.ended_at, Some(ended));
        assert_eq!(record.updated_at, ended);
        assert_eq!(record.failure_reason.as_deref(), Some("boom"));
        Ok(())
    }

    #[test]
    fn a_reopened_history_has_no_end_and_the_current_run_is_the_latest_start() -> TestResult {
        let wf_id = WorkflowId::new_v4();
        let first = RunId::new_v4();
        let history = vec![
            workflow_started(&wf_id, &first, None)?,
            Event::WorkflowCompleted {
                envelope: envelope(&wf_id, 2)?,
                result: payload()?,
            },
            Event::WorkflowReopened {
                envelope: envelope(&wf_id, 3)?,
                run_id: first.clone(),
                reopened: Vec::new(),
            },
        ];
        let record = project_visibility(&history, &first).ok_or("a started history projects")?;
        assert_eq!(record.status, WorkflowStatus::Running);
        assert_eq!(record.ended_at, None);
        assert_eq!(record.updated_at, envelope(&wf_id, 3)?.recorded_at);
        assert_eq!(current_run_id(&history), Some(first));
        Ok(())
    }

    /// Reconciliation writes a missing row, rewrites a stale one, and leaves
    /// a consistent one alone — proven through a store that counts writes.
    #[tokio::test]
    async fn reconcile_writes_only_rows_that_differ_from_history() -> TestResult {
        let backing = Arc::new(InMemoryStore::default());
        let events: Arc<dyn EventStore> = Arc::clone(&backing) as Arc<dyn EventStore>;
        let visibility: Arc<dyn VisibilityStore> = Arc::clone(&backing) as Arc<dyn VisibilityStore>;
        let wf_id = WorkflowId::new_v4();
        let run_id = RunId::new_v4();
        backing
            .append(
                WriteToken::recorder(),
                &wf_id,
                &[workflow_started(&wf_id, &run_id, None)?],
                0,
            )
            .await?;

        // Missing row: written.
        reconcile_visibility(Arc::clone(&events), Arc::clone(&visibility)).await?;
        let row = visibility
            .get_visibility(&wf_id, &run_id)
            .await?
            .ok_or("reconcile writes the missing row")?;
        assert_eq!(row.status, WorkflowStatus::Running);

        // Stale row: history moved on, the row is rewritten to match.
        backing
            .append(
                WriteToken::recorder(),
                &wf_id,
                &[Event::WorkflowCompleted {
                    envelope: envelope(&wf_id, 2)?,
                    result: payload()?,
                }],
                1,
            )
            .await?;
        reconcile_visibility(Arc::clone(&events), Arc::clone(&visibility)).await?;
        let row = visibility
            .get_visibility(&wf_id, &run_id)
            .await?
            .ok_or("the row survives reconcile")?;
        assert_eq!(row.status, WorkflowStatus::Completed);
        assert_eq!(row.ended_at, Some(envelope(&wf_id, 2)?.recorded_at));
        assert_eq!(row.updated_at, envelope(&wf_id, 2)?.recorded_at);

        // Consistent row: reconcile is a pure read.
        let before = row.clone();
        reconcile_visibility(events, Arc::clone(&visibility)).await?;
        assert_eq!(
            visibility.get_visibility(&wf_id, &run_id).await?,
            Some(before)
        );
        Ok(())
    }
}