Add workflow template submission mutation to graph-proxy

graph-proxy/src/graphql/mod.rs

@@ -5,8 +5,7 @@ mod workflows;
 
 use self::{workflow_templates::WorkflowTemplatesQuery, workflows::WorkflowsQuery};
 use async_graphql::{
-    EmptyMutation, EmptySubscription, InputObject, MergedObject, Schema, SchemaBuilder,
-    SimpleObject,
+    EmptySubscription, InputObject, MergedObject, Schema, SchemaBuilder, SimpleObject,
 };
 use async_graphql_axum::{GraphQLRequest, GraphQLResponse};
 use axum::extract::State;
@@ -16,19 +15,24 @@ use axum_extra::{
 };
 use lazy_static::lazy_static;
 use std::fmt::Display;
+use workflow_templates::WorkflowTemplatesMutation;
 
 /// The root schema of the service
-pub type RootSchema = Schema<Query, EmptyMutation, EmptySubscription>;
+pub type RootSchema = Schema<Query, Mutation, EmptySubscription>;
 
 /// A schema builder for the service
-pub fn root_schema_builder() -> SchemaBuilder<Query, EmptyMutation, EmptySubscription> {
-    Schema::build(Query::default(), EmptyMutation, EmptySubscription).enable_federation()
+pub fn root_schema_builder() -> SchemaBuilder<Query, Mutation, EmptySubscription> {
+    Schema::build(Query::default(), Mutation::default(), EmptySubscription).enable_federation()
 }
 
 /// The root query of the service
 #[derive(Debug, Clone, Default, MergedObject)]
 pub struct Query(WorkflowsQuery, WorkflowTemplatesQuery);
 
+/// The root mutation of the service
+#[derive(Debug, Clone, Default, MergedObject)]
+pub struct Mutation(WorkflowTemplatesMutation);
+
 /// Handles HTTP requests as GraphQL according to the provided [`Schema`]
 pub async fn graphql_handler(
     State(schema): State<RootSchema>,

graph-proxy/src/graphql/workflow_templates.rs

@@ -1,4 +1,4 @@
-use super::CLIENT;
+use super::{workflows::Workflow, VisitInput, CLIENT};
 use crate::ArgoServerUrl;
 use anyhow::anyhow;
 use argo_workflows_openapi::APIResult;
@@ -334,3 +334,80 @@ impl WorkflowTemplatesQuery {
         Ok(connection)
     }
 }
+
+/// Mutations related to [`WorkflowTemplate`]s
+#[derive(Debug, Clone, Default)]
+pub struct WorkflowTemplatesMutation;
+
+#[Object]
+impl WorkflowTemplatesMutation {
+    #[instrument(skip(self, ctx))]
+    async fn submit_workflow_template(
+        &self,
+        ctx: &Context<'_>,
+        name: String,
+        visit: VisitInput,
+        parameters: Json<HashMap<String, Value>>,
+    ) -> anyhow::Result<Workflow> {
+        let server_url = ctx.data_unchecked::<ArgoServerUrl>().deref();
+        let auth_token = ctx.data_unchecked::<Option<Authorization<Bearer>>>();
+        let mut url = server_url.clone();
+        let namespace = visit.to_string();
+        url.path_segments_mut()
+            .unwrap()
+            .extend(["api", "v1", "workflows", &namespace, "submit"]);
+        debug!("Submitting workflow template at {url}");
+        let request = if let Some(auth_token) = auth_token {
+            CLIENT.post(url).bearer_auth(auth_token.token())
+        } else {
+            CLIENT.post(url)
+        }
+        .json(
+            &argo_workflows_openapi::IoArgoprojWorkflowV1alpha1WorkflowSubmitRequest {
+                namespace: Some(namespace),
+                resource_kind: Some("ClusterWorkflowTemplate".to_string()),
+                resource_name: Some(name),
+                submit_options: Some(
+                    argo_workflows_openapi::IoArgoprojWorkflowV1alpha1SubmitOpts {
+                        annotations: None,
+                        dry_run: None,
+                        entry_point: None,
+                        generate_name: None,
+                        labels: None,
+                        name: None,
+                        owner_reference: None,
+                        parameters: parameters
+                            .0
+                            .into_iter()
+                            .filter_map(|(name, value)| to_argo_parameter(name, value).transpose())
+                            .collect::<Result<Vec<_>, _>>()?,
+                        pod_priority_class_name: None,
+                        priority: None,
+                        server_dry_run: None,
+                        service_account: None,
+                    },
+                ),
+            },
+        );
+        let workflow = request
+            .send()
+            .await?
+            .json::<APIResult<argo_workflows_openapi::IoArgoprojWorkflowV1alpha1Workflow>>()
+            .await?
+            .into_result()?;
+        Ok(Workflow::new(workflow, visit.into())?)
+    }
+}
+
+/// Convert a paramter into the format expected by the Argo Workflows API
+fn to_argo_parameter(name: String, value: Value) -> Result<Option<String>, serde_json::Error> {
+    Ok(match value {
+        Value::Null => Ok(None),
+        Value::Bool(bool) => Ok(Some(bool.to_string())),
+        Value::Number(number) => Ok(Some(number.to_string())),
+        Value::String(string) => Ok(Some(string)),
+        Value::Array(vec) => serde_json::to_string(&vec).map(Some),
+        Value::Object(map) => serde_json::to_string(&map).map(Some),
+    }?
+    .map(|parameter| format!("{name}={parameter}")))
+}

graph-proxy/src/graphql/workflows.rs

@@ -16,7 +16,7 @@ use tracing::{debug, instrument};
 /// An error encountered when parsing the Argo Server API Workflow response
 #[derive(Debug, thiserror::Error)]
 #[allow(clippy::missing_docs_in_private_items)]
-enum WorkflowParsingError {
+pub(super) enum WorkflowParsingError {
     #[error("status.phase was not a recognised value")]
     UnrecognisedPhase,
     #[error("status.start_time was expected but was not present")]
@@ -31,7 +31,7 @@ enum WorkflowParsingError {
 
 /// A Workflow consisting of one or more [`Task`]s
 #[derive(Debug, SimpleObject)]
-struct Workflow {
+pub(super) struct Workflow {
     /// Metadata containing name, proposal code, proposal number and visit of a workflow
     #[graphql(flatten)]
     metadata: Arc<Metadata>,
@@ -49,7 +49,7 @@ struct Metadata {
 
 #[allow(clippy::missing_docs_in_private_items)]
 impl Workflow {
-    fn new(
+    pub(super) fn new(
         value: IoArgoprojWorkflowV1alpha1Workflow,
         visit: Visit,
     ) -> Result<Self, WorkflowParsingError> {