server: Add support for creating apps in CMG (#2095)

Add support for creating app in CMG

part of: svix/monorepo-private#11626

Author: svix-mman

server/openapi.json

@@ -1931,6 +1931,11 @@
             },
             "MessageIn": {
                 "properties": {
+                    "application": {
+                        "$ref": "#/components/schemas/ApplicationIn",
+                        "description": "Optionally creates a new application alongside the message.\n\nIf the application id or uid that is used in the path already exists, this argument is ignored.",
+                        "nullable": true
+                    },
                     "channels": {
                         "description": "List of free-form identifiers that endpoints can filter by",
                         "example": [
@@ -5291,6 +5296,19 @@
                         },
                         "style": "form"
                     },
+                    {
+                        "in": "path",
+                        "name": "app_id",
+                        "required": true,
+                        "schema": {
+                            "example": "unique-app-identifier",
+                            "maxLength": 256,
+                            "minLength": 1,
+                            "pattern": "^[a-zA-Z0-9\\-_.]+$",
+                            "type": "string"
+                        },
+                        "style": "simple"
+                    },
                     {
                         "description": "The request's idempotency key",
                         "in": "header",

server/svix-server/src/v1/endpoints/application.rs

@@ -12,15 +12,15 @@ use axum::{
 use chrono::{DateTime, Utc};
 use futures::FutureExt;
 use schemars::JsonSchema;
-use sea_orm::{ActiveModelTrait, ActiveValue::Set, TransactionTrait};
+use sea_orm::{ActiveModelTrait, ActiveValue::Set, DatabaseConnection, TransactionTrait};
 use serde::{Deserialize, Serialize};
 use svix_server_derive::{aide_annotate, ModelOut};
 use validator::{Validate, ValidationError};
 
 use crate::{
     core::{
         permissions,
-        types::{metadata::Metadata, ApplicationId, ApplicationUid},
+        types::{metadata::Metadata, ApplicationId, ApplicationUid, OrganizationId},
     },
     db::models::{application, applicationmetadata},
     error::{http_error_on_conflict, HttpError, Result, Traceable},
@@ -270,11 +270,21 @@ async fn create_application(
         };
     }
 
-    let app = application::ActiveModel::new(org_id.clone());
+    let (app, metadata) = create_app_from_app_in(db, data, org_id).await?;
+
+    Ok(JsonStatusUpsert::Created((app, metadata).into()))
+}
+
+pub async fn create_app_from_app_in(
+    db: &DatabaseConnection,
+    app_in: ApplicationIn,
+    org_id: OrganizationId,
+) -> Result<(application::Model, applicationmetadata::Model)> {
+    let app = application::ActiveModel::new(org_id);
     let metadata = applicationmetadata::ActiveModel::new(app.id.clone().unwrap(), None);
 
     let mut model = (app, metadata);
-    data.update_model(&mut model);
+    app_in.update_model(&mut model);
     let (app, metadata) = model;
 
     let (app, metadata) = db
@@ -288,7 +298,7 @@ async fn create_application(
         })
         .await?;
 
-    Ok(JsonStatusUpsert::Created((app, metadata).into()))
+    Ok((app, metadata))
 }
 
 /// Get an application.

server/svix-server/src/v1/endpoints/endpoint/mod.rs

@@ -31,9 +31,9 @@ use crate::{
         cryptography::Encryption,
         permissions,
         types::{
-            metadata::Metadata, EndpointHeaders, EndpointHeadersPatch, EndpointId, EndpointSecret,
-            EndpointSecretInternal, EndpointUid, EventChannelSet, EventTypeName, EventTypeNameSet,
-            MessageStatus,
+            metadata::Metadata, ApplicationIdOrUid, EndpointHeaders, EndpointHeadersPatch,
+            EndpointId, EndpointSecret, EndpointSecretInternal, EndpointUid, EventChannelSet,
+            EventTypeName, EventTypeNameSet, MessageStatus,
         },
     },
     db::models::{
@@ -802,10 +802,20 @@ async fn send_example(
         uid: None,
         payload_retention_period: 90,
         extra_params: None,
+        application: None,
     };
 
-    let create_message =
-        create_message_inner(db, queue_tx, cache, false, Some(endpoint.id), msg_in, app).await?;
+    let create_message = create_message_inner(
+        db,
+        queue_tx,
+        cache,
+        false,
+        Some(endpoint.id),
+        msg_in,
+        app.org_id,
+        ApplicationIdOrUid(app.id.0),
+    )
+    .await?;
 
     Ok(Json(create_message))
 }

server/svix-server/src/v1/endpoints/message.rs

@@ -25,17 +25,21 @@ use crate::{
         message_app::CreateMessageApp,
         permissions,
         types::{
-            EndpointId, EventChannel, EventChannelSet, EventTypeName, EventTypeNameSet,
-            MessageAttemptTriggerType, MessageId, MessageUid,
+            ApplicationIdOrUid, EndpointId, EventChannel, EventChannelSet, EventTypeName,
+            EventTypeNameSet, MessageAttemptTriggerType, MessageId, MessageUid, OrganizationId,
         },
     },
     db::models::{application, message, messagecontent},
-    error::{http_error_on_conflict, Error, HttpError, Result},
+    error::{http_error_on_conflict, Error, HttpError, Result, ValidationErrorItem},
     queue::{MessageTaskBatch, TaskQueueProducer},
-    v1::utils::{
-        filter_and_paginate_time_limited, openapi_tag, validation_error, ApplicationMsgPath,
-        EventTypesQueryParams, JsonStatus, ListResponse, ModelIn, ModelOut, PaginationDescending,
-        PaginationLimit, ReversibleIterator, ValidatedJson, ValidatedQuery,
+    v1::{
+        endpoints::application::{create_app_from_app_in, ApplicationIn},
+        utils::{
+            filter_and_paginate_time_limited, openapi_tag, validation_error, validation_errors,
+            ApplicationMsgPath, ApplicationPath, EventTypesQueryParams, JsonStatus, ListResponse,
+            ModelIn, ModelOut, PaginationDescending, PaginationLimit, ReversibleIterator,
+            ValidatedJson, ValidatedQuery,
+        },
     },
     AppState,
 };
@@ -127,6 +131,14 @@ pub struct MessageIn {
     #[serde(rename = "transformationsParams")]
     #[schemars(skip)]
     pub extra_params: Option<MessageInExtraParams>,
+
+    /// Optionally creates a new application alongside the message.
+    ///
+    /// If the application id or uid that is used in the path already exists,
+    /// this argument is ignored.
+    #[validate]
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub application: Option<ApplicationIn>,
 }
 
 impl MessageIn {
@@ -335,23 +347,64 @@ async fn create_message(
     ValidatedQuery(CreateMessageQueryParams { with_content }): ValidatedQuery<
         CreateMessageQueryParams,
     >,
-    permissions::OrganizationWithApplication { app }: permissions::OrganizationWithApplication,
+    Path(ApplicationPath { app_id }): Path<ApplicationPath>,
+    permissions::Organization { org_id }: permissions::Organization,
     ValidatedJson(data): ValidatedJson<MessageIn>,
 ) -> Result<JsonStatus<202, MessageOut>> {
     Ok(JsonStatus(
-        create_message_inner(db, queue_tx, cache, with_content, None, data, app).await?,
+        create_message_inner(
+            db,
+            queue_tx,
+            cache,
+            with_content,
+            None,
+            data,
+            org_id,
+            app_id,
+        )
+        .await?,
     ))
 }
 
+#[allow(clippy::too_many_arguments)]
 pub(crate) async fn create_message_inner(
     db: &DatabaseConnection,
     queue_tx: TaskQueueProducer,
     cache: Cache,
     with_content: bool,
     force_endpoint: Option<EndpointId>,
     data: MessageIn,
-    app: application::Model,
+    org_id: OrganizationId,
+    app_id: ApplicationIdOrUid,
 ) -> Result<MessageOut> {
+    app_id.validate().map_err(|e| {
+        HttpError::unprocessable_entity(validation_errors(
+            vec!["path".to_owned(), "app_id_or_uid".to_owned()],
+            e,
+        ))
+    })?;
+
+    let app_from_path_app_id =
+        application::Entity::secure_find_by_id_or_uid(org_id.clone(), app_id.to_owned())
+            .one(db)
+            .await?;
+
+    let app = match (&data.application, app_from_path_app_id) {
+        (None, None) => {
+            return Err(
+                HttpError::not_found(None, Some("Application not found".to_string())).into(),
+            );
+        }
+
+        (_, Some(app_from_path_param)) => app_from_path_param,
+        (Some(cmg_app), None) => {
+            validate_create_app_uid(&app_id, cmg_app)?;
+            let (app, _metadata) = create_app_from_app_in(db, cmg_app.to_owned(), org_id).await?;
+
+            app
+        }
+    };
+
     let create_message_app = CreateMessageApp::layered_fetch(
         &cache,
         db,
@@ -407,6 +460,37 @@ pub(crate) async fn create_message_inner(
     Ok(msg_out)
 }
 
+fn validate_create_app_uid(app_id_or_uid: &ApplicationIdOrUid, data: &ApplicationIn) -> Result<()> {
+    // If implicit app creation is requested then the UID must be set
+    // in the request body, and it must match the UID given in the path
+    if let Some(uid) = &data.uid {
+        if uid.0 != app_id_or_uid.0 {
+            return Err(HttpError::unprocessable_entity(vec![ValidationErrorItem {
+                loc: vec![
+                    "body".to_string(),
+                    "application".to_string(),
+                    "uid".to_string(),
+                ],
+                msg: "Application UID in the path and body must match".to_string(),
+                ty: "application_uid_mismatch".to_string(),
+            }])
+            .into());
+        }
+    } else {
+        return Err(HttpError::unprocessable_entity(vec![ValidationErrorItem {
+            loc: vec![
+                "body".to_string(),
+                "application".to_string(),
+                "uid".to_string(),
+            ],
+            msg: "Application UID not set in body".to_string(),
+            ty: "application_uid_missing".to_string(),
+        }])
+        .into());
+    }
+    Ok(())
+}
+
 #[derive(Debug, Deserialize, Validate, JsonSchema)]
 pub struct GetMessageQueryParams {
     /// When `true` message payloads are included in the response

server/svix-server/src/v1/utils/mod.rs

@@ -500,7 +500,7 @@ pub fn validation_error(code: Option<&'static str>, msg: Option<&'static str>) -
 
 /// Recursively searches a [`validator::ValidationErrors`] tree into a linear list of errors to be
 /// sent to the user
-fn validation_errors(
+pub fn validation_errors(
     acc_path: Vec<String>,
     err: validator::ValidationErrors,
 ) -> Vec<ValidationErrorItem> {