aion-server 0.31.0

//! HTTP handlers for the read-only deployed-AWL studio surface.
//!
//! Two GETs and nothing else. There is no POST, PUT, PATCH, or DELETE on this
//! route family — not disabled, not guarded: absent, so a mutation attempt is
//! a 405 from the router rather than a handler decision that could regress.
//! `deployed_routes_reject_every_mutating_method` pins that.

use axum::{
    Json,
    extract::{Path, State},
    response::{IntoResponse, Response},
};

use super::auth::HttpCaller;
use super::error::{HttpWireError, Refusal};
use crate::ServerState;
use crate::awl::deployed::{DeployedDocument, DeployedError, DeployedVersion};

/// `GET /awl/deployed` — every loaded workflow version with the honest state
/// of its archived AWL source.
pub(crate) async fn list_deployed(
    State(state): State<ServerState>,
    HttpCaller(caller): HttpCaller,
) -> Result<Json<Vec<DeployedVersion>>, Refusal> {
    let engine = authorized_engine(&state, &caller).map_err(|error| refusal(&error))?;
    crate::awl::deployed::list_versions(&engine)
        .await
        .map(Json)
        .map_err(|error| Refusal::of(DeployedHttpError(error)))
}

/// `GET /awl/deployed/{workflow_type}/{content_hash}` — the archived AWL
/// document of one deployed version, with its schema files and projection.
pub(crate) async fn get_deployed_document(
    State(state): State<ServerState>,
    HttpCaller(caller): HttpCaller,
    Path((workflow_type, content_hash)): Path<(String, String)>,
) -> Result<Json<DeployedDocument>, Refusal> {
    let engine = authorized_engine(&state, &caller).map_err(|error| refusal(&error))?;
    crate::awl::deployed::read_document(&engine, &workflow_type, &content_hash)
        .await
        .map(Json)
        .map_err(|error| Refusal::of(DeployedHttpError(error)))
}

/// `GET /awl/deployed/{workflow_type}/{content_hash}/doc` — the documentation
/// model of one deployed revision.
///
/// The SAME model `aion awl doc --json` prints over the same archive, from
/// the same derivation (`aion_awl::doc::from_package`). An unreadable or
/// absent source is a 200 that NAMES the state and still serves the
/// committed half: the caller asked about a revision that exists, and
/// "unreadable" is an answer, where a 404 would be a lie.
pub(crate) async fn get_deployed_doc(
    State(state): State<ServerState>,
    HttpCaller(caller): HttpCaller,
    Path((workflow_type, content_hash)): Path<(String, String)>,
) -> Result<Json<aion_awl::doc::DocumentDoc>, Refusal> {
    let engine = authorized_engine(&state, &caller).map_err(|error| refusal(&error))?;
    crate::awl::deployed::read_doc(&engine, &workflow_type, &content_hash)
        .await
        .map(Json)
        .map_err(|error| Refusal::of(DeployedHttpError(error)))
}

/// Authorizes a deployed-surface read and borrows the engine.
///
/// The gate is the DEPLOY guard, not the studio's plain authenticated check:
/// this surface reads the deployed catalog and the source of deployed code,
/// which is what `GET /deploy/versions` gates the same way. Using the weaker
/// studio check would let any authenticated caller enumerate deployed versions
/// through the studio that the deploy API refuses to show them.
pub(super) fn authorized_engine(
    state: &ServerState,
    caller: &crate::CallerIdentity,
) -> Result<std::sync::Arc<aion::Engine>, crate::ServerError> {
    let guard = state.deploy_guard();
    guard.authorize(caller)?;
    guard.engine().map(std::sync::Arc::clone)
}

/// Renders a server refusal through the standard wire-error mapping.
pub(super) fn refusal(error: &crate::ServerError) -> Refusal {
    Refusal::of(HttpWireError(error.to_wire_error()))
}

pub(crate) struct DeployedHttpError(pub(crate) DeployedError);

impl IntoResponse for DeployedHttpError {
    /// One mapping with the run-scoped document read
    /// (`GET /workflows/{id}/document/{hash}`): the status follows the wire
    /// code (`not_found` → 404, backend → 500) and `error_type` names the
    /// reason, so the same archive refusal reads the same on both surfaces.
    fn into_response(self) -> Response {
        HttpWireError(self.0.to_wire_error()).into_response()
    }
}

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

    use aion::EngineBuilder;
    use aion_package::AwlSource;
    use aion_store::{EventStore, InMemoryStore, PackageRecord};
    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, SCHEMA_BYTES, SCHEMA_PATH, manifest, record};
    use crate::config::{DeployConfig, NamespaceMode};
    use crate::test_support::{EngineUnderTest, StateUnderTest};
    use crate::{NamespaceResolver, StaticScheduleNamespaces, StaticWorkflowNamespaces};

    /// A router over a real engine, with `deploy.enabled` per case and
    /// authentication OFF — which makes the caller the operator, who holds the
    /// deploy grant server-side. The archive rows are persisted AFTER the
    /// engine is built, so they are exactly the "persisted but not loaded"
    /// state, which is what this surface has to render honestly.
    async fn deployed_state(
        deploy_enabled: bool,
        auth_enabled: bool,
        rows: Vec<PackageRecord>,
        workspace: Option<&Path>,
    ) -> Result<(StateUnderTest, Arc<dyn EventStore>), 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?,
        ));
        for row in rows {
            store.put_package(row).await?;
        }
        let resolver = NamespaceResolver::from_parts(
            NamespaceMode::SharedEngine,
            Some(engine.handle()),
            Arc::new(StaticWorkflowNamespaces::default()),
            Arc::new(StaticScheduleNamespaces::default()),
        );
        let mut config = runtime_config();
        config.auth.enabled = auth_enabled;
        config.deploy = DeployConfig {
            enabled: deploy_enabled,
            max_archive_bytes: Some(1024 * 1024),
            max_inflated_bytes: Some(4 * 1024 * 1024),
        };
        config.authoring.workspace_dir = workspace.map(Path::to_path_buf);
        Ok((server_state(engine, resolver, config).await?, store))
    }

    fn operator_request(method: &str, uri: &str) -> Result<Request<body::Body>, axum::http::Error> {
        Request::builder()
            .method(method)
            .uri(uri)
            .header("x-aion-subject", "operator")
            .body(body::Body::empty())
    }

    /// The rendered step graph is served as SVG — the SAME picture the
    /// generated docs page inlines — with every node addressable by its model
    /// id, which is what the console's documentation view delegates clicks on.
    #[tokio::test]
    async fn a_deployed_revisions_step_graph_is_served_as_svg_with_addressable_nodes()
    -> Result<(), Box<dyn std::error::Error>> {
        let row = archived_document()?;
        let content_hash = row.content_hash.clone();
        let (state, _) = deployed_state(true, false, vec![row], None).await?;
        let router = workflow_router(state.clone());
        let response = router
            .oneshot(operator_request(
                "GET",
                &format!("/awl/deployed/deployed_probe/{content_hash}/doc/graph.svg"),
            )?)
            .await?;
        assert_eq!(response.status(), StatusCode::OK);
        let content_type = response
            .headers()
            .get(axum::http::header::CONTENT_TYPE)
            .and_then(|value| value.to_str().ok())
            .ok_or("the picture carried no content type")?
            .to_owned();
        assert!(content_type.starts_with("image/svg+xml"), "{content_type}");
        let svg = String::from_utf8(
            body::to_bytes(response.into_body(), usize::MAX)
                .await?
                .to_vec(),
        )?;
        assert!(svg.starts_with("<svg "), "not an svg document: {svg}");
        assert!(svg.contains("class=\"awl-step-graph\""), "{svg}");
        assert!(svg.contains("data-node=\"outcome:done\""), "{svg}");
        Ok(())
    }

    /// No source, no graph: the SVG route refuses with the same stated
    /// absence the document read gives, never an empty picture that would
    /// read as "this workflow has no steps".
    #[tokio::test]
    async fn a_version_without_source_has_no_step_graph_and_says_so()
    -> Result<(), Box<dyn std::error::Error>> {
        let row = record(manifest("gleam_authored"), None, 1_700_000_000)?;
        let content_hash = row.content_hash.clone();
        let (state, _) = deployed_state(true, false, vec![row], None).await?;
        let router = workflow_router(state.clone());
        let response = router
            .oneshot(operator_request(
                "GET",
                &format!("/awl/deployed/gleam_authored/{content_hash}/doc/graph.svg"),
            )?)
            .await?;
        assert_eq!(response.status(), StatusCode::NOT_FOUND);
        let body: serde_json::Value = read_json(response).await?;
        assert_eq!(body["error_type"], "DeployedAwlSourceAbsent");
        Ok(())
    }

    /// The workspace half serves the same picture for a buffer an author is
    /// editing, and refuses — as a stated refusal, not an image — a buffer the
    /// checker cannot derive a model from.
    #[tokio::test]
    async fn a_workspace_documents_step_graph_is_served_and_an_unreadable_one_refused()
    -> Result<(), Box<dyn std::error::Error>> {
        let workspace = tempfile::tempdir()?;
        std::fs::write(workspace.path().join("deployed_probe.awl"), DOCUMENT)?;
        let (state, _) = deployed_state(true, false, Vec::new(), Some(workspace.path())).await?;
        let router = workflow_router(state.clone());
        let derived = router
            .clone()
            .oneshot(json_request(
                "/awl/doc/graph.svg",
                &serde_json::json!({ "source": DOCUMENT, "path": "deployed_probe.awl" }),
            )?)
            .await?;
        assert_eq!(derived.status(), StatusCode::OK);
        let svg = String::from_utf8(
            body::to_bytes(derived.into_body(), usize::MAX)
                .await?
                .to_vec(),
        )?;
        assert!(svg.contains("data-node=\"outcome:done\""), "{svg}");

        let refused = router
            .oneshot(json_request(
                "/awl/doc/graph.svg",
                // The SAME admitted path with a buffer the checker cannot read:
                // the refusal under test is the document's, not the path gate's.
                &serde_json::json!({ "source": "workflow", "path": "deployed_probe.awl" }),
            )?)
            .await?;
        assert_eq!(refused.status(), StatusCode::BAD_REQUEST);
        let body: serde_json::Value = read_json(refused).await?;
        assert_eq!(body["error_type"], "AwlDocumentRefused");
        Ok(())
    }

    fn json_request(
        uri: &str,
        value: &serde_json::Value,
    ) -> Result<Request<body::Body>, Box<dyn std::error::Error>> {
        Ok(Request::builder()
            .method("POST")
            .uri(uri)
            .header("x-aion-subject", "operator")
            .header("content-type", "application/json")
            .body(body::Body::from(serde_json::to_vec(value)?))?)
    }

    /// An archive carrying a document and the schema it imports.
    ///
    /// The manifest's input schema is deliberately DISTINCT from its output
    /// schema (which stays the fixture default `{"type":"object"}`): the wire
    /// pin below asserts on the input-only `required` marker, so a surface
    /// that served the wrong contract half could not pass by coincidence.
    fn archived_document() -> Result<PackageRecord, Box<dyn std::error::Error>> {
        let mut declared = manifest("deployed_probe");
        declared.input_schema = serde_json::json!({
            "type": "object",
            "properties": { "order_id": { "type": "string" } },
            "required": ["order_id"],
        });
        record(
            declared,
            Some(AwlSource::new(
                "deployed_probe.awl",
                DOCUMENT,
                [(SCHEMA_PATH.to_owned(), SCHEMA_BYTES.to_vec())],
            )),
            1_700_000_000,
        )
    }

    /// Every file under `root`, by relative path, with its bytes.
    fn snapshot(root: &Path) -> std::io::Result<BTreeMap<String, Vec<u8>>> {
        let mut found = BTreeMap::new();
        let mut pending = vec![root.to_path_buf()];
        while let Some(directory) = pending.pop() {
            for entry in std::fs::read_dir(&directory)? {
                let path = entry?.path();
                if path.is_dir() {
                    pending.push(path);
                } else {
                    let key = path.strip_prefix(root).map_or_else(
                        |_| path.to_string_lossy().into_owned(),
                        |relative| relative.to_string_lossy().into_owned(),
                    );
                    found.insert(key, std::fs::read(&path)?);
                }
            }
        }
        Ok(found)
    }

    /// 🔴 THE READ-ONLY PIN AT THE TRANSPORT. Both endpoints answer (survival),
    /// and neither the operator's workspace nor the package store differs by a
    /// single byte afterwards (absence). The workspace half is the ruled
    /// invariant made testable: a deployed view must never write into the
    /// directory the operator is editing in.
    #[tokio::test]
    async fn reading_the_deployed_surface_changes_neither_workspace_nor_store()
    -> Result<(), Box<dyn std::error::Error>> {
        let workspace = crate::test_support::private_tempdir()?;
        std::fs::write(
            workspace.path().join("deployed_probe.awl"),
            b"//! A DIFFERENT document the operator is editing.\nworkflow deployed_probe\n",
        )?;
        let row = archived_document()?;
        let content_hash = row.content_hash.clone();
        let (state, store) =
            deployed_state(true, false, vec![row.clone()], Some(workspace.path())).await?;
        let router = workflow_router(state.clone());

        let workspace_before = snapshot(workspace.path())?;
        let store_before = store.list_packages().await?;

        let listing = router
            .clone()
            .oneshot(operator_request("GET", "/awl/deployed")?)
            .await?;
        assert_eq!(listing.status(), StatusCode::OK);
        let versions: serde_json::Value = read_json(listing).await?;
        let versions = versions.as_array().ok_or("listing was not an array")?;
        assert_eq!(versions.len(), 1);
        assert_eq!(versions[0]["workflow_type"], "deployed_probe");
        assert_eq!(versions[0]["content_hash"], content_hash);
        assert_eq!(versions[0]["source"]["state"], "available");
        assert_eq!(versions[0]["source"]["document_name"], "deployed_probe.awl");
        assert_eq!(versions[0]["source"]["schema_count"], 1);
        assert_eq!(versions[0]["loaded"], false);

        let detail = router
            .clone()
            .oneshot(operator_request(
                "GET",
                &format!("/awl/deployed/deployed_probe/{content_hash}"),
            )?)
            .await?;
        assert_eq!(detail.status(), StatusCode::OK);
        let document: serde_json::Value = read_json(detail).await?;
        assert_eq!(
            document["source"], DOCUMENT,
            "the deployed surface must serve the ARCHIVED document, not the \
             same-named one in the operator's workspace"
        );
        assert_eq!(document["projection"]["ok"], true);
        // The console's form seam: the committed START schema reaches the
        // wire, told apart from the output schema by its input-only
        // `required: ["order_id"]` marker (the fixture keeps the output at the
        // bare object schema, so a surface serving the wrong half fails here).
        assert_eq!(document["input_schema"]["required"][0], "order_id");
        // The primary entry of a contract-bearing archive answers a PRESENT
        // signal list; this fixture's manifest-derived contract commits an
        // empty one, which is `[]` — never `null`, which means "not committed
        // to this entry".
        assert_eq!(
            document["signals"].as_array().map(Vec::len),
            Some(0),
            "this fixture's contract commits an empty signal list: {:?}",
            document["signals"]
        );

        assert_eq!(
            snapshot(workspace.path())?,
            workspace_before,
            "reading the deployed surface wrote into the operator's workspace"
        );
        assert_eq!(
            store.list_packages().await?,
            store_before,
            "reading the deployed surface rewrote the package store"
        );
        Ok(())
    }

    /// 🔴 NO MUTATION PATH EXISTS. The routes carry a single `get`, so every
    /// other method is a 405 the router itself answers — there is no handler
    /// for a write to reach, guarded or otherwise.
    #[tokio::test]
    async fn deployed_routes_reject_every_mutating_method() -> Result<(), Box<dyn std::error::Error>>
    {
        let row = archived_document()?;
        let content_hash = row.content_hash.clone();
        let (state, store) = deployed_state(true, false, vec![row], None).await?;
        let router = workflow_router(state.clone());
        let before = store.list_packages().await?;

        let detail = format!("/awl/deployed/deployed_probe/{content_hash}");
        let doc = format!("{detail}/doc");
        let graph = format!("{detail}/doc/graph.svg");
        for method in ["POST", "PUT", "PATCH", "DELETE"] {
            for uri in [
                "/awl/deployed",
                detail.as_str(),
                doc.as_str(),
                graph.as_str(),
            ] {
                let response = router
                    .clone()
                    .oneshot(operator_request(method, uri)?)
                    .await?;
                assert_eq!(
                    response.status(),
                    StatusCode::METHOD_NOT_ALLOWED,
                    "{method} {uri} reached something"
                );
            }
        }
        assert_eq!(
            store.list_packages().await?,
            before,
            "a refused method still changed the store"
        );
        Ok(())
    }

    /// The surface follows the deploy switch: with `[deploy]` off it is a
    /// plain 404, not an empty listing that would read as "nothing deployed".
    #[tokio::test]
    async fn the_deployed_surface_is_dark_when_deploy_is_disabled()
    -> Result<(), Box<dyn std::error::Error>> {
        let row = archived_document()?;
        let content_hash = row.content_hash.clone();
        let (state, _) = deployed_state(false, false, vec![row], None).await?;
        let router = workflow_router(state.clone());

        for uri in [
            "/awl/deployed".to_owned(),
            format!("/awl/deployed/deployed_probe/{content_hash}"),
        ] {
            let response = router
                .clone()
                .oneshot(operator_request("GET", &uri)?)
                .await?;
            assert_eq!(
                response.status(),
                StatusCode::NOT_FOUND,
                "{uri} answered with the deploy surface disabled"
            );
        }
        Ok(())
    }

    /// A version that carries no archived source is a 404 whose body STATES
    /// the absence and both of its causes — never a 200 with an empty
    /// document, and never a 500 that reads as breakage.
    #[tokio::test]
    async fn a_version_without_archived_source_answers_with_the_absence_stated()
    -> Result<(), Box<dyn std::error::Error>> {
        let row = record(manifest("gleam_authored"), None, 1_700_000_000)?;
        let content_hash = row.content_hash.clone();
        let (state, _) = deployed_state(true, false, vec![row], None).await?;
        let router = workflow_router(state.clone());

        let listing = router
            .clone()
            .oneshot(operator_request("GET", "/awl/deployed")?)
            .await?;
        let versions: serde_json::Value = read_json(listing).await?;
        assert_eq!(versions[0]["source"]["state"], "absent");

        let detail = router
            .oneshot(operator_request(
                "GET",
                &format!("/awl/deployed/gleam_authored/{content_hash}"),
            )?)
            .await?;
        assert_eq!(detail.status(), StatusCode::NOT_FOUND);
        let body: serde_json::Value = read_json(detail).await?;
        assert_eq!(body["error_type"], "DeployedAwlSourceAbsent");
        let message = body["message"]
            .as_str()
            .ok_or("refusal carried no message")?;
        assert!(message.contains("Gleam"), "{message}");
        assert!(
            message.contains("before deploys archived their source"),
            "{message}"
        );
        Ok(())
    }

    /// A GET carrying (or withholding) the deploy grant, expressed through
    /// whichever credential path is compiled: the JWT `deploy` claim under
    /// `feature = "auth"`, the development `x-aion-deploy` header otherwise.
    /// One shape, so the grant pin runs on BOTH builds rather than only the
    /// one the declared gate does not compile.
    fn deploy_grant_request(
        uri: &str,
        granted: bool,
    ) -> Result<Request<body::Body>, Box<dyn std::error::Error>> {
        #[cfg(feature = "auth")]
        let token = if granted {
            crate::auth::test_support::mint_token_with_deploy("alice", "tenant-a", true)?
        } else {
            crate::auth::test_support::mint_token("alice", "tenant-a")?
        };
        #[cfg(not(feature = "auth"))]
        let token = super::super::test_support::TOKEN.to_owned();

        let builder = Request::builder()
            .method("GET")
            .uri(uri)
            .header("authorization", format!("Bearer {token}"))
            .header("x-aion-subject", "alice")
            .header("x-aion-namespaces", "tenant-a");
        // A cfg-gated shadow rather than a mutation: under `auth` the grant
        // already rode in on the token above, so there is no reassignment for
        // a `mut` to be about and `--all-features` clippy refuses one.
        #[cfg(not(feature = "auth"))]
        let builder = if granted {
            builder.header("x-aion-deploy", "true")
        } else {
            builder
        };
        Ok(builder.body(body::Body::empty())?)
    }

    /// The gate is the DEPLOY grant, not the studio's plain authenticated
    /// check: a caller the deploy API would refuse must not enumerate deployed
    /// versions — or read deployed source — through the studio instead. The
    /// granted arm is the control: without it a surface that refused everyone
    /// would pass.
    #[tokio::test]
    async fn a_caller_without_the_deploy_grant_is_refused() -> Result<(), Box<dyn std::error::Error>>
    {
        let row = archived_document()?;
        let content_hash = row.content_hash.clone();
        let (state, _) = deployed_state(true, true, vec![row], None).await?;
        let router = workflow_router(state.clone());
        let detail = format!("/awl/deployed/deployed_probe/{content_hash}");

        for uri in ["/awl/deployed", detail.as_str()] {
            let denied = router
                .clone()
                .oneshot(deploy_grant_request(uri, false)?)
                .await?;
            assert_eq!(
                denied.status(),
                StatusCode::FORBIDDEN,
                "{uri} served deployed state to a caller with no deploy grant"
            );

            let granted = router
                .clone()
                .oneshot(deploy_grant_request(uri, true)?)
                .await?;
            assert_eq!(
                granted.status(),
                StatusCode::OK,
                "{uri} refused a caller who does hold the deploy grant"
            );
        }
        Ok(())
    }
}