server: Add support for creating application in cmg

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/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> {

server/svix-server/tests/it/e2e_message.rs

@@ -2,6 +2,7 @@
 // SPDX-License-Identifier: MIT
 
 use chrono::{Duration, Utc};
+use rand::distributions::DistString;
 use reqwest::StatusCode;
 use sea_orm::{sea_query::Expr, ColumnTrait, EntityTrait, QueryFilter};
 use serde::de::IgnoredAny;
@@ -18,6 +19,10 @@ use svix_server::{
     },
 };
 
+fn rand_str(len: usize) -> String {
+    rand::distributions::Alphanumeric.sample_string(&mut rand::thread_rng(), len)
+}
+
 use crate::utils::{
     common_calls::{create_test_app, create_test_endpoint, create_test_msg_with, message_in},
     run_with_retries, start_svix_server, TestReceiver,
@@ -603,3 +608,111 @@ async fn test_raw_payload() {
     let rec_body = receiver.data_recv.recv().await;
     assert_eq!(msg_payload.to_string(), rec_body.unwrap().to_string());
 }
+
+#[tokio::test]
+async fn test_create_message_with_application() {
+    let (client, _jh) = start_svix_server().await;
+
+    let app_uid = format!("app-created-in-cmg-{}", rand_str(15));
+
+    // cmg without the application field fails
+    let _: IgnoredAny = client
+        .post(
+            &format!("api/v1/app/{app_uid}/msg/"),
+            json!({
+                "eventType": "test.event",
+                "payload": { "test": "value" }
+            }),
+            StatusCode::NOT_FOUND,
+        )
+        .await
+        .unwrap();
+
+    // cmg with application
+    let _: IgnoredAny = client
+        .post(
+            &format!("api/v1/app/{app_uid}/msg/"),
+            json!({
+                "eventType": "test.event",
+                "payload": { "test": "value1" },
+                "application": {
+                    "name": "Test App Created With Message",
+                    "uid": app_uid,
+                }
+            }),
+            StatusCode::ACCEPTED,
+        )
+        .await
+        .unwrap();
+
+    // app was created
+    let app: serde_json::Value = client
+        .get(&format!("api/v1/app/{app_uid}/"), StatusCode::OK)
+        .await
+        .unwrap();
+
+    assert_eq!(app["uid"], app_uid);
+    assert_eq!(app["name"], "Test App Created With Message");
+
+    // Create another message to the now-existing app with the application field
+    // The application field should be ignored since the app already exists
+    let _: IgnoredAny = client
+        .post(
+            &format!("api/v1/app/{app_uid}/msg/"),
+            json!({
+                "eventType": "test.event",
+                "payload": { "test": "value2" },
+                "application": {
+                    "name": "Updated name will be ignored",
+                    "uid": app_uid,
+                }
+            }),
+            StatusCode::ACCEPTED,
+        )
+        .await
+        .unwrap();
+
+    // Verify the app name didn't change
+    let app_after: serde_json::Value = client
+        .get(&format!("api/v1/app/{app_uid}/"), StatusCode::OK)
+        .await
+        .unwrap();
+
+    assert_eq!(app_after["name"], "Test App Created With Message");
+
+    // UID in path must match UID in body
+    let _: IgnoredAny = client
+        .post(
+            "api/v1/app/different-uid/msg/",
+            json!({
+                "eventType": "test.event",
+                "payload": { "test": "value" },
+                "payloadRetentionPeriod": 5,
+                "application": {
+                    "name": "Test App",
+                    "uid": app_uid,  // This doesn't match the path
+                }
+            }),
+            StatusCode::UNPROCESSABLE_ENTITY,
+        )
+        .await
+        .unwrap();
+
+    // UID must be set in body when creating
+    let _: IgnoredAny = client
+        .post(
+            "api/v1/app/new-app-uid/msg/",
+            json!({
+                "eventType": "test.event",
+                "payload": { "test": "value" },
+                "payloadRetentionPeriod": 5,
+                "application": {
+                    "name": "Test App Without UID",
+                    // Missing uid field
+                }
+            }),
+            StatusCode::UNPROCESSABLE_ENTITY,
+        )
+        .await
+        .unwrap();
+}

server/svix-server/tests/it/utils/common_calls.rs

@@ -124,6 +124,7 @@ pub fn message_in<T: Serialize>(event_type: &str, payload: T) -> Result<MessageI
         channels: None,
         uid: None,
         extra_params: None,
+        application: None,
     })
 }