server: Support sending raw (pre-formatted) payloads (#1930)

Based on #1929.
Closes #1927.

Author: svix-jplatte

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

@@ -799,6 +799,7 @@ async fn send_example(
         payload: RawPayload::from_string(example).unwrap(),
         uid: None,
         payload_retention_period: 90,
+        extra_params: None,
     };
 
     let create_message =

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

@@ -124,6 +124,27 @@ pub struct MessageIn {
     #[serde(default = "default_90")]
     #[schemars(example = "default_90")]
     pub payload_retention_period: i64,
+    #[serde(rename = "transformationsParams")]
+    #[schemars(skip)]
+    pub extra_params: Option<MessageInExtraParams>,
+}
+
+impl MessageIn {
+    fn payload(&self) -> Vec<u8> {
+        if let Some(params) = &self.extra_params {
+            if let Some(raw_payload) = &params.raw_payload {
+                return raw_payload.as_bytes().to_owned();
+            }
+        }
+
+        self.payload.0.get().as_bytes().to_owned()
+    }
+}
+
+#[derive(Clone, Debug, Default, PartialEq, Eq, Deserialize, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct MessageInExtraParams {
+    raw_payload: Option<String>,
 }
 
 fn example_channel_set() -> Vec<&'static str> {
@@ -350,7 +371,7 @@ pub(crate) async fn create_message_inner(
     // Should never happen since you're giving it an existing Application, but just in case
     .ok_or_else(|| Error::generic(format!("Application doesn't exist: {}", app.id)))?;
 
-    let payload = data.payload.to_string().into_bytes();
+    let payload = data.payload();
     let msg = message::ActiveModel {
         app_id: Set(app.id.clone()),
         org_id: Set(app.org_id),

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

@@ -531,3 +531,36 @@ async fn test_message_validation() {
         .await
         .unwrap();
 }
+
+#[tokio::test]
+async fn test_raw_payload() {
+    let (client, _jh) = start_svix_server().await;
+
+    let app_id = create_test_app(&client, "testRawPayload").await.unwrap().id;
+
+    let mut receiver = TestReceiver::start(axum::http::StatusCode::OK);
+
+    create_test_endpoint(&client, &app_id, &receiver.endpoint)
+        .await
+        .unwrap();
+
+    let msg_payload = json!({ "test": "value1" });
+
+    let _: IgnoredAny = client
+        .post(
+            &format!("api/v1/app/{app_id}/msg/"),
+            json!({
+                "eventType": "payload.raw",
+                "payload": {},
+                "transformationsParams": {
+                    "rawPayload": msg_payload.to_string(),
+                },
+            }),
+            StatusCode::ACCEPTED,
+        )
+        .await
+        .unwrap();
+
+    let rec_body = receiver.data_recv.recv().await;
+    assert_eq!(msg_payload.to_string(), rec_body.unwrap().to_string());
+}

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

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