Implement send-example route (#976)

svix-daniel · web-flow · commit 3a4343393914 · 2023-06-28T18:22:39.000-04:00
This change implements the `send-example` route which allows dispatching
example events to a given endpoint.
diff --git a/server/openapi.json b/server/openapi.json
@@ -950,6 +950,20 @@
                 ],
                 "type": "object"
             },
+            "EventExampleIn": {
+                "properties": {
+                    "eventType": {
+                        "example": "user.signup",
+                        "maxLength": 256,
+                        "pattern": "^[a-zA-Z0-9\\-_.]+$",
+                        "type": "string"
+                    }
+                },
+                "required": [
+                    "eventType"
+                ],
+                "type": "object"
+            },
             "EventTypeIn": {
                 "properties": {
                     "archived": {
@@ -4574,8 +4588,35 @@
         },
         "/api/v1/app/{app_id}/endpoint/{endpoint_id}/send-example/": {
             "post": {
-                "description": "Send an example message for event",
+                "description": "Send an example message for an event",
+                "operationId": "v1.endpoint.send-example",
                 "parameters": [
+                    {
+                        "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"
+                    },
+                    {
+                        "in": "path",
+                        "name": "endpoint_id",
+                        "required": true,
+                        "schema": {
+                            "example": "unique-ep-identifier",
+                            "maxLength": 256,
+                            "minLength": 1,
+                            "pattern": "^[a-zA-Z0-9\\-_.]+$",
+                            "type": "string"
+                        },
+                        "style": "simple"
+                    },
                     {
                         "description": "The request's idempotency key",
                         "in": "header",
@@ -4586,11 +4627,89 @@
                         "style": "simple"
                     }
                 ],
+                "requestBody": {
+                    "content": {
+                        "application/json": {
+                            "schema": {
+                                "$ref": "#/components/schemas/EventExampleIn"
+                            }
+                        }
+                    },
+                    "required": true
+                },
                 "responses": {
-                    "204": {
-                        "description": "no content"
+                    "200": {
+                        "content": {
+                            "application/json": {
+                                "schema": {
+                                    "$ref": "#/components/schemas/MessageOut"
+                                }
+                            }
+                        },
+                        "description": ""
+                    },
+                    "401": {
+                        "content": {
+                            "application/json": {
+                                "schema": {
+                                    "$ref": "#/components/schemas/HttpErrorOut"
+                                }
+                            }
+                        },
+                        "description": "Unauthorized"
+                    },
+                    "403": {
+                        "content": {
+                            "application/json": {
+                                "schema": {
+                                    "$ref": "#/components/schemas/HttpErrorOut"
+                                }
+                            }
+                        },
+                        "description": "Forbidden"
+                    },
+                    "404": {
+                        "content": {
+                            "application/json": {
+                                "schema": {
+                                    "$ref": "#/components/schemas/HttpErrorOut"
+                                }
+                            }
+                        },
+                        "description": "Not Found"
+                    },
+                    "409": {
+                        "content": {
+                            "application/json": {
+                                "schema": {
+                                    "$ref": "#/components/schemas/HttpErrorOut"
+                                }
+                            }
+                        },
+                        "description": "Conflict"
+                    },
+                    "422": {
+                        "content": {
+                            "application/json": {
+                                "schema": {
+                                    "$ref": "#/components/schemas/HTTPValidationError"
+                                }
+                            }
+                        },
+                        "description": "Validation Error"
+                    },
+                    "429": {
+                        "content": {
+                            "application/json": {
+                                "schema": {
+                                    "$ref": "#/components/schemas/HttpErrorOut"
+                                }
+                            }
+                        },
+                        "description": "Too Many Requests"
                     }
                 },
+                "summary": "Send Event Type Example Message",
                 "tags": [
                     "Endpoint"
                 ]
diff --git a/server/svix-server/src/db/models/eventtype.rs b/server/svix-server/src/db/models/eventtype.rs
@@ -102,6 +102,21 @@ pub fn schema_example() -> serde_json::Value {
 pub struct Schema(HashMap<String, Json>);
 json_wrapper!(Schema);
 
+impl Schema {
+    pub fn example(&self) -> Option<&serde_json::Value> {
+        self.0
+            .get("1")
+            .and_then(|version| match version {
+                serde_json::Value::Object(obj) => obj.get("examples"),
+                _ => None,
+            })
+            .and_then(|examples| match examples {
+                serde_json::Value::Array(arr) => arr.iter().next(),
+                _ => None,
+            })
+    }
+}
+
 impl JsonSchema for Schema {
     fn schema_name() -> String {
         stringify!(Schema).to_string()
diff --git a/server/svix-server/src/queue/mod.rs b/server/svix-server/src/queue/mod.rs
@@ -94,18 +94,21 @@ impl MessageTask {
 pub struct MessageTaskBatch {
     pub msg_id: MessageId,
     pub app_id: ApplicationId,
+    pub force_endpoint: Option<EndpointId>,
     pub trigger_type: MessageAttemptTriggerType,
 }
 
 impl MessageTaskBatch {
     pub fn new_task(
         msg_id: MessageId,
         app_id: ApplicationId,
+        force_endpoint: Option<EndpointId>,
         trigger_type: MessageAttemptTriggerType,
     ) -> QueueTask {
         QueueTask::MessageBatch(Self {
             msg_id,
             app_id,
+            force_endpoint,
             trigger_type,
         })
     }
diff --git a/server/svix-server/src/v1/endpoints/endpoint/mod.rs b/server/svix-server/src/v1/endpoints/endpoint/mod.rs
@@ -12,20 +12,20 @@ use crate::{
         permissions,
         types::{
             metadata::Metadata, BaseId, EndpointId, EndpointSecretInternal, EndpointUid,
-            EventChannelSet, EventTypeNameSet, MessageEndpointId, MessageStatus,
+            EventChannelSet, EventTypeName, EventTypeNameSet, MessageEndpointId, MessageStatus,
         },
     },
     ctx,
-    db::models::messagedestination,
+    db::models::{eventtype, messagedestination},
     error::{self, HttpError},
     v1::utils::{
-        api_not_implemented, openapi_desc, openapi_tag,
+        openapi_tag,
         patch::{
             patch_field_non_nullable, patch_field_nullable, UnrequiredField,
             UnrequiredNullableField,
         },
         validate_no_control_characters, validate_no_control_characters_unrequired,
-        validation_error, ApplicationEndpointPath, ModelIn,
+        validation_error, ApplicationEndpointPath, ModelIn, ValidatedJson,
     },
     AppState,
 };
@@ -53,6 +53,8 @@ use crate::db::models::endpoint;
 
 use self::secrets::generate_secret;
 
+use super::message::{create_message_inner, MessageIn, MessageOut, RawPayload};
+
 pub fn validate_event_types_ids(
     event_types_ids: &EventTypeNameSet,
 ) -> std::result::Result<(), ValidationError> {
@@ -730,7 +732,81 @@ async fn endpoint_stats(
     }))
 }
 
-const SEND_EXAMPLE_DESCRIPTION: &str = "Send an example message for event";
+#[derive(Deserialize, JsonSchema, Validate)]
+#[serde(rename_all = "camelCase")]
+struct EventExampleIn {
+    event_type: EventTypeName,
+}
+
+const SVIX_PING_EVENT_TYPE_NAME: &str = "svix.ping";
+const SVIX_PING_EVENT_TYPE_PAYLOAD: &str = r#"{"success": true}"#;
+
+/// Send an example message for an event
+#[aide_annotate(
+    op_id = "v1.endpoint.send-example",
+    op_summary = "Send Event Type Example Message"
+)]
+async fn send_example(
+    state: State<AppState>,
+    Path(ApplicationEndpointPath { endpoint_id, .. }): Path<ApplicationEndpointPath>,
+    permissions::OrganizationWithApplication { app }: permissions::OrganizationWithApplication,
+    ValidatedJson(data): ValidatedJson<EventExampleIn>,
+) -> error::Result<Json<MessageOut>> {
+    let State(AppState {
+        ref db,
+        queue_tx,
+        cache,
+        ..
+    }) = state;
+
+    let endpoint = ctx!(
+        endpoint::Entity::secure_find_by_id_or_uid(app.id.clone(), endpoint_id)
+            .one(db)
+            .await
+    )?
+    .ok_or_else(|| HttpError::not_found(None, None))?;
+
+    let example = if data.event_type == EventTypeName(SVIX_PING_EVENT_TYPE_NAME.to_owned()) {
+        SVIX_PING_EVENT_TYPE_PAYLOAD.to_string()
+    } else {
+        let event_type = ctx!(
+            eventtype::Entity::secure_find_by_name(app.org_id.clone(), data.event_type.clone())
+                .one(db)
+                .await
+        )?
+        .ok_or_else(|| HttpError::not_found(None, None))?;
+
+        let example = event_type.schemas.and_then(|schema| {
+            schema
+                .example()
+                .and_then(|ex| serde_json::to_string(ex).ok())
+        });
+
+        match example {
+            Some(example) => example,
+            None => {
+                return Err(HttpError::bad_request(
+                    Some("invalid_scheme".to_owned()),
+                    Some("Unable to generate example message from event-type schema".to_owned()),
+                )
+                .into());
+            }
+        }
+    };
+
+    let msg_in = MessageIn {
+        channels: None,
+        event_type: data.event_type,
+        payload: RawPayload::from_string(example).unwrap(),
+        uid: None,
+        payload_retention_period: 90,
+    };
+
+    let create_message =
+        create_message_inner(db, queue_tx, cache, false, Some(endpoint.id), msg_in, app).await?;
+
+    Ok(Json(create_message))
+}
 
 pub fn router() -> ApiRouter<AppState> {
     let tag = openapi_tag("Endpoint");
@@ -772,7 +848,7 @@ pub fn router() -> ApiRouter<AppState> {
         )
         .api_route_with(
             "/app/:app_id/endpoint/:endpoint_id/send-example/",
-            post_with(api_not_implemented, openapi_desc(SEND_EXAMPLE_DESCRIPTION)),
+            post_with(send_example, send_example_operation),
             &tag,
         )
         .api_route_with(
diff --git a/server/svix-server/src/v1/endpoints/message.rs b/server/svix-server/src/v1/endpoints/message.rs
@@ -3,16 +3,19 @@
 
 use crate::{
     core::{
+        cache::Cache,
         message_app::CreateMessageApp,
         permissions,
         types::{
-            EventChannel, EventChannelSet, EventTypeName, EventTypeNameSet,
+            EndpointId, EventChannel, EventChannelSet, EventTypeName, EventTypeNameSet,
             MessageAttemptTriggerType, MessageId, MessageUid,
         },
     },
-    ctx, err_generic,
+    ctx,
+    db::models::application,
+    err_generic,
     error::{HttpError, Result},
-    queue::MessageTaskBatch,
+    queue::{MessageTaskBatch, TaskQueueProducer},
     v1::utils::{
         apply_pagination_desc, iterator_from_before_or_after, openapi_tag, validation_error,
         ApplicationMsgPath, EventTypesQueryParams, JsonStatus, ListResponse, ModelIn, ModelOut,
@@ -305,6 +308,20 @@ async fn create_message(
     permissions::OrganizationWithApplication { app }: permissions::OrganizationWithApplication,
     ValidatedJson(data): ValidatedJson<MessageIn>,
 ) -> Result<JsonStatus<202, MessageOut>> {
+    Ok(JsonStatus(
+        create_message_inner(db, queue_tx, cache, with_content, None, data, app).await?,
+    ))
+}
+
+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,
+) -> Result<MessageOut> {
     let create_message_app = CreateMessageApp::layered_fetch(
         &cache,
         db,
@@ -334,6 +351,7 @@ async fn create_message(
                 MessageTaskBatch::new_task(
                     msg.id.clone(),
                     app.id.clone(),
+                    force_endpoint,
                     MessageAttemptTriggerType::Scheduled,
                 ),
                 None,
@@ -347,7 +365,7 @@ async fn create_message(
         MessageOut::without_payload(msg)
     };
 
-    Ok(JsonStatus(msg_out))
+    Ok(msg_out)
 }
 
 #[derive(Debug, Deserialize, Validate, JsonSchema)]
diff --git a/server/svix-server/src/worker.rs b/server/svix-server/src/worker.rs
@@ -748,7 +748,7 @@ async fn process_queue_task_inner(
         QueueTask::MessageBatch(task) => {
             let msg = ctx!(message::Entity::find_by_id(task.msg_id).one(db).await)?
                 .ok_or_else(|| err_generic!("Unexpected: message doesn't exist"))?;
-            (msg, None, None, task.trigger_type, 0)
+            (msg, task.force_endpoint, None, task.trigger_type, 0)
         }
     };
 
diff --git a/server/svix-server/tests/e2e_endpoint.rs b/server/svix-server/tests/e2e_endpoint.rs

Original file line number	Diff line number	Diff line change
`@@ -748,7 +748,7 @@ async fn process_queue_task_inner(`
`748`	`748`	`QueueTask::MessageBatch(task) => {`
`749`	`749`	`let msg = ctx!(message::Entity::find_by_id(task.msg_id).one(db).await)?`
`750`	`750`	`.ok_or_else(\|\| err_generic!("Unexpected: message doesn't exist"))?;`
`751`		`- (msg, None, None, task.trigger_type, 0)`
	`751`	`+ (msg, task.force_endpoint, None, task.trigger_type, 0)`
`752`	`752`	`}`
`753`	`753`	`};`
`754`	`754`