aion-server 0.31.0

//! `GET /workflows/{workflow_id}/document/{content_hash}` — the deployed AWL
//! document a run is executing, read under the RUN'S permission.
//!
//! Mounted on the workflow family unconditionally: it does not follow the
//! `[deploy].enabled` switch, because a server that is not a deploy target
//! still runs packages, and an operator watching one of its runs needs the
//! graph the run is executing. Authorization is the describe scope
//! ([`crate::api::handlers::authorize_run_document`]), never the deploy grant.
//!
//! The URL is immutable — a content hash names exactly one archive forever —
//! so the answer carries `ETag: "<hash>"` and `Cache-Control: private,
//! immutable`, and a conditional request whose `If-None-Match` names the hash
//! is answered `304` after the SAME authorization, without opening the
//! archive. Cache validation never skips the permission check.

use aion_core::WorkflowId;
use aion_proto::{ProtoWorkflowId, WireError};
use axum::{
    Json,
    extract::{Path, Query, State, rejection::QueryRejection},
    http::{HeaderMap, HeaderValue, StatusCode, header},
    response::{IntoResponse, Response},
};
use serde::Deserialize;

use super::auth::HttpCaller;
use super::error::HttpWireError;
use crate::ServerState;
use crate::api::handlers;

/// A year, the longest lifetime the HTTP caching RFC lets a validator
/// express; `immutable` tells a conforming cache never to revalidate within
/// it. The document a hash names cannot change, so both are true statements.
const CACHE_CONTROL: &str = "private, max-age=31536000, immutable";

/// The query half of the address: the namespace to scope the read to, the
/// same field every workflow-family body carries.
#[derive(Debug, Deserialize)]
pub(crate) struct RunDocumentQuery {
    namespace: String,
}

/// `GET /workflows/{workflow_id}/document/{content_hash}?namespace=…`.
pub(crate) async fn get_run_document(
    State(state): State<ServerState>,
    HttpCaller(caller): HttpCaller,
    Path((workflow_id, content_hash)): Path<(String, String)>,
    query: Result<Query<RunDocumentQuery>, QueryRejection>,
    headers: HeaderMap,
) -> Result<Response, HttpWireError> {
    let Query(query) = query.map_err(|rejection| {
        HttpWireError(WireError::invalid_input(format!(
            "the `namespace` query parameter is required: {rejection}"
        )))
    })?;
    let workflow_id: WorkflowId = ProtoWorkflowId { uuid: workflow_id }
        .try_into()
        .map_err(HttpWireError)?;
    let access = handlers::authorize_run_document(
        state.namespace_guard(),
        &caller,
        &query.namespace,
        &workflow_id,
        &content_hash,
    )
    .await
    .map_err(HttpWireError)?;

    let etag = HeaderValue::from_str(&format!("\"{}\"", access.content_hash)).map_err(|error| {
        HttpWireError(WireError::backend(format!(
            "content hash is not a valid ETag: {error}"
        )))
    })?;
    let cache_control = HeaderValue::from_static(CACHE_CONTROL);
    if if_none_match_names(&headers, &etag) {
        return Ok((
            StatusCode::NOT_MODIFIED,
            [(header::ETAG, etag), (header::CACHE_CONTROL, cache_control)],
        )
            .into_response());
    }
    let document = access.read().await.map_err(HttpWireError)?;
    Ok((
        StatusCode::OK,
        [(header::ETAG, etag), (header::CACHE_CONTROL, cache_control)],
        Json(document),
    )
        .into_response())
}

/// Whether the request's `If-None-Match` names `etag` (or `*`). A weak
/// validator (`W/"…"`) matches too: the weak comparison is the one the RFC
/// prescribes for `If-None-Match`, and a byte-identical archive is
/// semantically identical by construction.
fn if_none_match_names(headers: &HeaderMap, etag: &HeaderValue) -> bool {
    let Ok(strong) = etag.to_str() else {
        return false;
    };
    headers
        .get_all(header::IF_NONE_MATCH)
        .iter()
        .filter_map(|value| value.to_str().ok())
        .flat_map(|value| value.split(','))
        .map(str::trim)
        .any(|candidate| {
            candidate == "*" || candidate.strip_prefix("W/").unwrap_or(candidate) == strong
        })
}

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

    use aion::EngineBuilder;
    use aion_core::{Event, PackageVersion, RunId, WorkflowId};
    use aion_package::AwlSource;
    use aion_store::{EventStore, InMemoryStore, WriteToken};
    use axum::{body, http::Request, http::StatusCode};
    use tower::ServiceExt;

    use super::super::router::workflow_router;
    use super::super::test_support::{read_json, runtime_config, server_state};
    use crate::awl::deployed::fixtures::{DOCUMENT, manifest, record};
    use crate::config::{DeployConfig, NamespaceMode};
    use crate::test_support::{EngineUnderTest, StateUnderTest};
    use crate::{NamespaceResolver, StaticScheduleNamespaces, StaticWorkflowNamespaces};

    const NAMESPACE: &str = "tenant-a";

    fn workflow_id() -> WorkflowId {
        WorkflowId::new(uuid::Uuid::from_u128(0x51))
    }

    /// State over a real engine with the deploy surface OFF and
    /// authentication OFF (the caller is the single-tenant operator), one
    /// persisted archive, and one run started under it — or, with
    /// `started_under_archive` false, under a hash no archive carries. The
    /// caller holds the state so the engine is shut down when the test ends;
    /// `workflow_router(state.clone())` is the router.
    async fn state_with_run(
        started_under_archive: bool,
    ) -> Result<(StateUnderTest, String), Box<dyn std::error::Error>> {
        let store: Arc<dyn EventStore> = Arc::new(InMemoryStore::default());
        let engine = EngineUnderTest::new(Arc::new(
            EngineBuilder::new()
                .stop_drain_timeout(std::time::Duration::from_secs(5))
                .store_arc(Arc::clone(&store))
                .in_memory_visibility()
                .scheduler_threads(1)
                .build()
                .await?,
        ));
        let row = record(
            manifest("fixture"),
            Some(AwlSource::new(
                "fixture.awl",
                DOCUMENT,
                std::iter::empty::<(String, Vec<u8>)>(),
            )),
            1_700_000_000,
        )?;
        let archive_hash = row.content_hash.clone();
        store.put_package(row).await?;
        let started_under = if started_under_archive {
            archive_hash.clone()
        } else {
            "b".repeat(64)
        };
        store
            .append(
                WriteToken::recorder(),
                &workflow_id(),
                &[Event::WorkflowStarted {
                    envelope: aion_core::EventEnvelope {
                        seq: 1,
                        recorded_at: chrono::Utc::now(),
                        workflow_id: workflow_id(),
                    },
                    workflow_type: "fixture".to_owned(),
                    input: aion_core::Payload::from_json(&serde_json::json!({}))?,
                    run_id: RunId::new(uuid::Uuid::from_u128(1)),
                    parent_run_id: None,
                    parent_workflow_id: None,
                    package_version: PackageVersion::new(started_under.clone()),
                }],
                0,
            )
            .await?;
        let ownership = StaticWorkflowNamespaces::default();
        ownership.record(workflow_id(), NAMESPACE)?;
        let resolver = NamespaceResolver::from_parts(
            NamespaceMode::SharedEngine,
            Some(engine.handle()),
            Arc::new(ownership),
            Arc::new(StaticScheduleNamespaces::default()),
        );
        let mut config = runtime_config();
        config.auth.enabled = false;
        // 🔴 Deploy OFF: the run-scoped read must not follow the deploy switch.
        config.deploy = DeployConfig {
            enabled: false,
            max_archive_bytes: Some(1024 * 1024),
            max_inflated_bytes: Some(4 * 1024 * 1024),
        };
        Ok((server_state(engine, resolver, config).await?, started_under))
    }

    fn request(
        uri: &str,
        if_none_match: Option<&str>,
    ) -> Result<Request<body::Body>, axum::http::Error> {
        let mut builder = Request::builder()
            .method("GET")
            .uri(uri)
            .header("x-aion-subject", "operator");
        if let Some(value) = if_none_match {
            builder = builder.header("if-none-match", value);
        }
        builder.body(body::Body::empty())
    }

    fn document_uri(hash: &str) -> String {
        format!(
            "/workflows/{}/document/{hash}?namespace={NAMESPACE}",
            workflow_id()
        )
    }

    #[tokio::test]
    async fn the_run_document_is_served_with_deploy_off_and_is_cacheable_forever()
    -> Result<(), Box<dyn std::error::Error>> {
        let (state, hash) = state_with_run(true).await?;
        let router = workflow_router(state.clone());
        let response = router
            .clone()
            .oneshot(request(&document_uri(&hash), None)?)
            .await?;
        assert_eq!(response.status(), StatusCode::OK);
        assert_eq!(
            response.headers().get("etag").and_then(|v| v.to_str().ok()),
            Some(format!("\"{hash}\"").as_str())
        );
        assert_eq!(
            response
                .headers()
                .get("cache-control")
                .and_then(|v| v.to_str().ok()),
            Some(super::CACHE_CONTROL)
        );
        let document: serde_json::Value = read_json(response).await?;
        assert_eq!(document["content_hash"], hash);
        assert_eq!(document["workflow_type"], "fixture");
        assert_eq!(document["source"], DOCUMENT);
        assert!(
            document["projection"].is_object(),
            "the check projection rides along: {document}"
        );

        // The deploy-gated catalog is dark on this server; the run read is not
        // a way around the switch because it answers only this run's hash.
        let catalog = router
            .clone()
            .oneshot(request(&format!("/awl/deployed/fixture/{hash}"), None)?)
            .await?;
        assert_eq!(catalog.status(), StatusCode::NOT_FOUND);
        Ok(())
    }

    #[tokio::test]
    async fn a_conditional_request_naming_the_hash_is_not_modified()
    -> Result<(), Box<dyn std::error::Error>> {
        let (state, hash) = state_with_run(true).await?;
        let router = workflow_router(state.clone());
        for validator in [
            format!("\"{hash}\""),
            format!("W/\"{hash}\""),
            format!("\"other\", \"{hash}\""),
            "*".to_owned(),
        ] {
            let response = router
                .clone()
                .oneshot(request(&document_uri(&hash), Some(&validator))?)
                .await?;
            assert_eq!(response.status(), StatusCode::NOT_MODIFIED, "{validator}");
            assert_eq!(
                response.headers().get("etag").and_then(|v| v.to_str().ok()),
                Some(format!("\"{hash}\"").as_str())
            );
        }
        let response = router
            .oneshot(request(&document_uri(&hash), Some("\"stale\""))?)
            .await?;
        assert_eq!(response.status(), StatusCode::OK);
        Ok(())
    }

    /// 🔴 A validator never buys a 304 the caller could not have earned a 200
    /// for: the conditional branch runs AFTER authorization. A hash this
    /// workflow never recorded, and a namespace the workflow is not in, are
    /// refused exactly as an unconditional request is — even when
    /// `If-None-Match` names the hash the caller is asking about.
    #[tokio::test]
    async fn a_conditional_request_is_refused_before_it_can_be_not_modified()
    -> Result<(), Box<dyn std::error::Error>> {
        let (state, hash) = state_with_run(true).await?;
        let router = workflow_router(state.clone());
        let foreign = "c".repeat(64);
        let response = router
            .clone()
            .oneshot(request(
                &document_uri(&foreign),
                Some(&format!("\"{foreign}\"")),
            )?)
            .await?;
        assert_eq!(response.status(), StatusCode::NOT_FOUND);
        let error: serde_json::Value = read_json(response).await?;
        assert_eq!(error["error_type"], "RunPackageNotRecorded");

        let response = router
            .oneshot(request(
                &format!(
                    "/workflows/{}/document/{hash}?namespace=tenant-b",
                    workflow_id()
                ),
                Some(&format!("\"{hash}\"")),
            )?)
            .await?;
        assert_ne!(response.status(), StatusCode::NOT_MODIFIED);
        assert_ne!(response.status(), StatusCode::OK);
        Ok(())
    }

    #[tokio::test]
    async fn refusals_are_typed_404s_and_a_missing_namespace_is_400()
    -> Result<(), Box<dyn std::error::Error>> {
        let (state, hash) = state_with_run(true).await?;
        let router = workflow_router(state.clone());
        let foreign = "c".repeat(64);
        let response = router
            .clone()
            .oneshot(request(&document_uri(&foreign), None)?)
            .await?;
        assert_eq!(response.status(), StatusCode::NOT_FOUND);
        let error: serde_json::Value = read_json(response).await?;
        assert_eq!(error["error_type"], "RunPackageNotRecorded");

        let response = router
            .clone()
            .oneshot(request(
                &format!("/workflows/{}/document/{hash}", workflow_id()),
                None,
            )?)
            .await?;
        assert_eq!(response.status(), StatusCode::BAD_REQUEST);

        let response = router
            .clone()
            .oneshot(request(&document_uri("not-a-hash"), None)?)
            .await?;
        assert_eq!(response.status(), StatusCode::BAD_REQUEST);

        let response = router
            .oneshot(request(
                &format!("/workflows/not-a-uuid/document/{hash}?namespace={NAMESPACE}"),
                None,
            )?)
            .await?;
        assert_eq!(response.status(), StatusCode::BAD_REQUEST);
        Ok(())
    }

    #[tokio::test]
    async fn a_recorded_hash_with_no_archive_here_is_deployed_version_not_found()
    -> Result<(), Box<dyn std::error::Error>> {
        let (state, hash) = state_with_run(false).await?;
        let router = workflow_router(state.clone());
        let response = router.oneshot(request(&document_uri(&hash), None)?).await?;
        assert_eq!(response.status(), StatusCode::NOT_FOUND);
        let error: serde_json::Value = read_json(response).await?;
        assert_eq!(error["error_type"], "DeployedVersionNotFound");
        Ok(())
    }
}