aion-server 0.30.0

//! The shared list handler: the one place every list surface — `POST
//! /workflows/list`, gRPC `ListWorkflows`, the MCP `list_runs` tool — answers
//! the workflow list contract from.

use aion_proto::{
    ProtoListWorkflowsRequest, ProtoListWorkflowsResponse, WireError, convert::encode_core_value,
};

use super::payload::decode_list_request;
use crate::{CallerIdentity, NamespaceGuard, NamespaceOperation, ServerError};

/// Handles a list-workflows request: one page of the contract, read from the
/// visibility projection through the scoped engine.
///
/// The caller is authorized against `request.namespace` BEFORE the envelope
/// is decoded, so a denied caller learns nothing from a malformed request.
/// The decoded contract request must name that same namespace — the page is
/// then read under the namespace authorization resolved, never under one the
/// envelope smuggled in. A malformed request, a zero limit, or a cursor
/// minted under a different query is `invalid_input`; the projection never
/// reads history here.
///
/// # Errors
///
/// Returns a stable [`WireError`] when namespace scoping fails, the request
/// envelope is missing or malformed, the store refuses the query, or the page
/// cannot be encoded.
///
/// `provenance` is what the serving install says about ITSELF (ADR-016),
/// stamped onto the page as given — see
/// [`describe`](super::describe::describe) for why it is an argument.
pub async fn list(
    guard: &NamespaceGuard,
    caller: &CallerIdentity,
    request: ProtoListWorkflowsRequest,
    provenance: aion_core::ReadProvenance,
) -> Result<ProtoListWorkflowsResponse, WireError> {
    let scoped = guard
        .scope(caller, &NamespaceOperation::list(&request))
        .await
        .map_err(|error| error.to_wire_error())?;
    let mut list_request = decode_list_request(request.request.as_ref())?;
    if list_request.namespace != request.namespace {
        return Err(WireError::invalid_input(format!(
            "list request names namespace `{}` but the call is scoped to `{}`",
            list_request.namespace, request.namespace
        )));
    }
    list_request.namespace = scoped.namespace().to_owned();

    let engine = scoped.engine().map_err(|error| error.to_wire_error())?;
    let mut page = engine
        .list_workflows(&list_request)
        .await
        .map_err(|error| ServerError::from(error).to_wire_error())?;
    page.provenance = Some(provenance);
    let page = encode_core_value(scoped.namespace().to_owned(), None, &page)?;
    Ok(ProtoListWorkflowsResponse { page: Some(page) })
}

#[cfg(test)]
mod tests {
    use aion_core::{
        RunId, SortDirection, WorkflowId, WorkflowListFilter, WorkflowListPage,
        WorkflowListRequest, WorkflowSort, WorkflowSortField, WorkflowStatus,
    };
    use aion_proto::{
        WireErrorCode,
        convert::{ProtoPayload, decode_core_value, encode_core_value},
    };
    use aion_store::visibility::VisibilityRecord;
    use chrono::Utc;

    use super::super::test_support::{
        NAMESPACE, append_started, context, denied_guard, run_id, workflow_id,
    };
    use super::*;
    use crate::{
        NamespaceResolver, StaticScheduleNamespaces, StaticWorkflowNamespaces,
        config::NamespaceMode,
    };

    fn row(workflow_id: WorkflowId, run_id: RunId, workflow_type: &str) -> VisibilityRecord {
        VisibilityRecord {
            namespace: NAMESPACE.to_owned(),
            workflow_id,
            run_id,
            workflow_type: workflow_type.to_owned(),
            status: WorkflowStatus::Running,
            started_at: Utc::now(),
            updated_at: Utc::now(),
            ended_at: None,
            parent: None,
            display_name: None,
            kind: None,
            failed_step: None,
            failure_reason: None,
            search_attributes: std::collections::HashMap::new(),
            outstanding_leases: Vec::new(),
            package_version: None,
        }
    }

    fn list_request(namespace: &str, filter: WorkflowListFilter) -> WorkflowListRequest {
        WorkflowListRequest {
            namespace: namespace.to_owned(),
            filter,
            sort: WorkflowSort {
                field: WorkflowSortField::StartedAt,
                direction: SortDirection::Desc,
            },
            cursor: None,
            limit: 10,
        }
    }

    fn proto(
        namespace: &str,
        request: &WorkflowListRequest,
    ) -> Result<ProtoListWorkflowsRequest, WireError> {
        Ok(ProtoListWorkflowsRequest {
            namespace: namespace.to_owned(),
            request: Some(encode_core_value(namespace, None, request)?),
        })
    }

    fn decode_page(response: &ProtoListWorkflowsResponse) -> Result<WorkflowListPage, WireError> {
        let envelope = response
            .page
            .as_ref()
            .ok_or_else(|| WireError::backend("page missing"))?;
        decode_core_value(envelope)
    }

    /// Regression test (#51): the engine's internal schedule-coordinator
    /// workflow must never surface through the shared list handler (the gRPC
    /// list RPC and `POST /workflows/list` ride it). The coordinator row sits
    /// in the tenant's namespace here to model any path that scopes the
    /// coordinator into a tenant — namespace scoping must not be the only
    /// thing hiding engine internals — and the count hides it too.
    #[tokio::test]
    async fn list_handler_hides_engine_internal_workflows_from_items_and_count()
    -> Result<(), Box<dyn std::error::Error>> {
        let context = context().await?;
        append_started(context.store.as_ref()).await?;
        context
            .visibility_store
            .record_visibility(row(workflow_id(), run_id(), "fixture"))
            .await?;
        context
            .visibility_store
            .record_visibility(row(
                WorkflowId::new(uuid::Uuid::from_u128(0xa10a)),
                RunId::new(uuid::Uuid::from_u128(0xa10b)),
                "aion.schedule_coordinator",
            ))
            .await?;

        let request = proto(
            NAMESPACE,
            &list_request(NAMESPACE, WorkflowListFilter::default()),
        )?;
        let page = decode_page(
            &list(
                &context.guard,
                &context.caller,
                request,
                aion_core::ReadProvenance::default(),
            )
            .await?,
        )?;
        assert_eq!(
            page.items.len(),
            1,
            "list must hide engine-internal workflows"
        );
        assert_eq!(page.count, 1, "the count must hide them too");
        assert_eq!(page.items[0].workflow_type, "fixture");

        // Naming the internal type explicitly is the operator's escape hatch.
        let request = proto(
            NAMESPACE,
            &list_request(
                NAMESPACE,
                WorkflowListFilter {
                    workflow_types: vec![String::from("aion.schedule_coordinator")],
                    ..WorkflowListFilter::default()
                },
            ),
        )?;
        let page = decode_page(
            &list(
                &context.guard,
                &context.caller,
                request,
                aion_core::ReadProvenance::default(),
            )
            .await?,
        )?;
        assert_eq!(page.count, 1);
        assert_eq!(page.items[0].workflow_type, "aion.schedule_coordinator");
        Ok(())
    }

    #[tokio::test]
    async fn list_handler_scopes_then_reads_the_projection()
    -> Result<(), Box<dyn std::error::Error>> {
        let context = context().await?;
        append_started(context.store.as_ref()).await?;
        context
            .visibility_store
            .record_visibility(row(workflow_id(), run_id(), "fixture"))
            .await?;
        let request = proto(
            NAMESPACE,
            &list_request(
                NAMESPACE,
                WorkflowListFilter {
                    workflow_types: vec![String::from("fixture")],
                    statuses: vec![WorkflowStatus::Running],
                    ..WorkflowListFilter::default()
                },
            ),
        )?;

        let page = decode_page(
            &list(
                &context.guard,
                &context.caller,
                request,
                aion_core::ReadProvenance::new(5),
            )
            .await?,
        )?;

        assert_eq!(page.items.len(), 1);
        // ADR-016 / WA-010 R4: the install's own count rides the page exactly
        // as the boundary supplied it — never the engine's default 0, which
        // would read as "never recorded here".
        assert_eq!(page.provenance, Some(aion_core::ReadProvenance::new(5)));
        assert_eq!(page.items[0].workflow_id, workflow_id());
        assert_eq!(page.items[0].run_id, run_id());
        assert_eq!(page.next_cursor, None);
        assert_eq!(page.count, 1);
        Ok(())
    }

    /// The envelope cannot re-target the page: a request whose inner namespace
    /// differs from the one the caller was authorized against is refused as
    /// invalid input, never silently read under either.
    #[tokio::test]
    async fn list_handler_refuses_an_envelope_naming_another_namespace()
    -> Result<(), Box<dyn std::error::Error>> {
        let context = context().await?;
        let request = proto(
            NAMESPACE,
            &list_request("tenant-b", WorkflowListFilter::default()),
        )?;
        let error = list(
            &context.guard,
            &context.caller,
            request,
            aion_core::ReadProvenance::default(),
        )
        .await;
        assert_eq!(
            error.err().map(|error| error.code),
            Some(WireErrorCode::InvalidInput)
        );
        Ok(())
    }

    /// A cursor minted under one query and replayed under another is refused
    /// as `invalid_input` (the store's typed refusal, mapped at the wire) — the
    /// console treats it as "restart from the first page".
    #[tokio::test]
    async fn list_handler_refuses_a_cursor_from_another_query()
    -> Result<(), Box<dyn std::error::Error>> {
        let context = context().await?;
        append_started(context.store.as_ref()).await?;
        for n in 1..=3_u128 {
            context
                .visibility_store
                .record_visibility(row(
                    WorkflowId::new(uuid::Uuid::from_u128(n)),
                    RunId::new(uuid::Uuid::from_u128(n + 100)),
                    "fixture",
                ))
                .await?;
        }
        let mut first = list_request(NAMESPACE, WorkflowListFilter::default());
        first.limit = 2;
        let page = decode_page(
            &list(
                &context.guard,
                &context.caller,
                proto(NAMESPACE, &first)?,
                aion_core::ReadProvenance::default(),
            )
            .await?,
        )?;
        let cursor = page.next_cursor.ok_or("a second page must exist")?;

        let mut other = list_request(NAMESPACE, WorkflowListFilter::default());
        other.sort.direction = SortDirection::Asc;
        other.cursor = Some(cursor);
        let error = list(
            &context.guard,
            &context.caller,
            proto(NAMESPACE, &other)?,
            aion_core::ReadProvenance::default(),
        )
        .await;
        assert_eq!(
            error.err().map(|error| error.code),
            Some(WireErrorCode::InvalidInput)
        );
        Ok(())
    }

    #[tokio::test]
    async fn denied_handler_returns_namespace_denied_before_engine_access()
    -> Result<(), Box<dyn std::error::Error>> {
        let ownership = StaticWorkflowNamespaces::default();
        let resolver = NamespaceResolver::authorization_only(
            NamespaceMode::SharedEngine,
            ownership,
            StaticScheduleNamespaces::default(),
        );
        let guard = NamespaceGuard::new(resolver);
        let caller = CallerIdentity::new("alice", [String::from("tenant-b")]);
        let request = proto(
            NAMESPACE,
            &list_request(NAMESPACE, WorkflowListFilter::default()),
        )?;

        let error = list(
            &guard,
            &caller,
            request,
            aion_core::ReadProvenance::default(),
        )
        .await;

        assert_eq!(
            error.err().map(|error| error.code),
            Some(WireErrorCode::NamespaceDenied)
        );
        Ok(())
    }

    #[tokio::test]
    async fn denied_list_does_not_decode_malformed_request_before_namespace_check()
    -> Result<(), Box<dyn std::error::Error>> {
        let (guard, caller) = denied_guard();
        let request = ProtoListWorkflowsRequest {
            namespace: NAMESPACE.to_owned(),
            request: Some(aion_proto::WireEnvelope {
                namespace: NAMESPACE.to_owned(),
                request_id: None,
                payload: Some(ProtoPayload {
                    content_type: "application/octet-stream".to_owned(),
                    bytes: Vec::new(),
                }),
            }),
        };

        let error = list(
            &guard,
            &caller,
            request,
            aion_core::ReadProvenance::default(),
        )
        .await;

        assert_eq!(
            error.err().map(|error| error.code),
            Some(WireErrorCode::NamespaceDenied)
        );
        Ok(())
    }

    /// A missing envelope is a malformed call, not an empty page: the contract
    /// has no default sort or limit to fall back on.
    #[tokio::test]
    async fn list_handler_refuses_a_missing_request_envelope()
    -> Result<(), Box<dyn std::error::Error>> {
        let context = context().await?;
        let request = ProtoListWorkflowsRequest {
            namespace: NAMESPACE.to_owned(),
            request: None,
        };
        let error = list(
            &context.guard,
            &context.caller,
            request,
            aion_core::ReadProvenance::default(),
        )
        .await;
        assert_eq!(
            error.err().map(|error| error.code),
            Some(WireErrorCode::InvalidInput)
        );
        Ok(())
    }
}