Bridge: receivers now publish to queues with payload field (#982)

This diff is a little more broad than just "put the payload in a field."
It also tries to formalize the requirements of each output case, both
Senders and Receivers.

For Senders, we need a certain shape useful for building a call to
Create Message.
For Receivers, outputs expect to get a JSON object with a `payload`
field on it, representing the data to publish to the queue or what have
you.

These boundary types are now both in the `-types` crate, and used to
ensure both as-is and transformed data conforms where needed.

Existing tests caught the changes on the receiver side (which is
the main thing for this diff).

Additionally, for Senders, I camelCased the `appId` field to make the
create message JSON cased more consistently. This was a remark that came
up during review of an up-coming blog post.

Author: svix-onelson

bridge/README.md

@@ -35,7 +35,7 @@ Transformations are configured as either a single string of JS source code:
 transformation: |
     function handler(input) {
       return {
-        app_id: input.key,
+        appId: input.key,
         message: {
           eventType: input.event_type,
           payload: input.data
@@ -52,7 +52,7 @@ transformation:
   src: |
     function handler(input) {
       return {
-        app_id: input.key,
+        appId: input.key,
         message: {
           eventType: input.event_type,
           payload: input.data
@@ -78,7 +78,7 @@ transformation:
       let payload = JSON.parse(msg.getElementsByTagName("payload")[0].textContent)
 
       return {
-        app_id: msg.attributes.appId,
+        appId: msg.attributes.appId,
         message: {
           eventType: msg.attributes.eventType,
           payload,
@@ -91,7 +91,7 @@ by the configured `format` field.
 
 Note that regardless of the `format`, the return type of `handler` must be an `Object`.
 
-> N.b. at time of writing, `format: string` is currently unsupported for `senders` and `receivers` configured with
+> N.b. at time of writing, `format: string` is unsupported for `senders` and `receivers` configured with
 > a `redis` input or output.
 
 ---
@@ -137,25 +137,30 @@ Each sender and receiver can optionally specify a `transformation`.
 Transformations should define a function called `handler` that accepts an object and returns an object.
 
 Senders should produce JSON following an expected shape:
-```
+```json
 {
-    // This indicates which Svix application to send the message to
-    "app_id": "app_XYZ",
-    
-    // The `message` field has the same requirements as the standard `MessageIn`
-    // used for Create Message API requests
+    "appId": "app_XYZ",
     "message": {
         "eventType": "my.event",
-        "payload": {"abc": 123}
+        "payload": {"code": 123, "message": "something happened..."}
     }
 }
 ```
 
-> The comments in the above JSON are for illustrative purposes only ;)
-> That's not valid JSON! Sorry!
-
 For detail on the `message` field, see: <https://api.svix.com/docs#tag/Message/operation/v1.message.create>
 
+Receivers can accept arbitrary body data but their outputs require a JSON object with a `payload` field representing the
+message to publish.
+
+```json
+{
+    "payload": {"msg": "my cool message, published by svix-bridge!"}
+}
+```
+
+By configuring a transformation, you should be able to consume a variety of `POST` bodies and
+produce a valid output.
+
 See the example configs for how to configure each input and output in more detail:
 - [senders](./svix-bridge.example.senders.yaml)
 - [receivers](./svix-bridge.example.receivers.yaml)

bridge/svix-bridge-plugin-queue/src/lib.rs

@@ -3,13 +3,10 @@ use generic_queue::rabbitmq::{RabbitMqConsumer, RabbitMqDelivery};
 use generic_queue::redis::{RedisStreamConsumer, RedisStreamDelivery, RedisStreamJsonSerde};
 use generic_queue::sqs::{SqsDelivery, SqsQueueConsumer};
 use generic_queue::{Delivery, QueueError, TaskQueueReceive};
-use serde::{Deserialize, Serialize};
 use std::time::{Duration, Instant};
 use svix_bridge_types::{
-    async_trait,
-    svix::api::{MessageIn, PostOptions as PostOptions_, Svix},
-    JsObject, TransformationConfig, TransformerInput, TransformerInputFormat, TransformerJob,
-    TransformerOutput, TransformerTx,
+    async_trait, svix::api::Svix, CreateMessageRequest, JsObject, TransformationConfig,
+    TransformerInput, TransformerInputFormat, TransformerJob, TransformerOutput, TransformerTx,
 };
 use tracing::instrument;
 
@@ -244,7 +241,7 @@ trait Consumer {
                     delivery.nack().await.map_err(Error::from)?;
                     return Ok(());
                 }
-                Ok(x) => x,
+                Ok(x) => serde_json::from_value(serde_json::Value::Object(x))?,
             }
         } else {
             // Parse as JSON when not using a transformation because Create Message requires JSON.
@@ -302,40 +299,18 @@ async fn run_inner(consumer: &(impl Consumer + Send + Sync)) -> std::io::Result<
     }
 }
 
-#[derive(Clone, Default, Deserialize, Serialize)]
-pub struct PostOptions {
-    idempotency_key: Option<String>,
-}
-
-impl From<PostOptions> for PostOptions_ {
-    fn from(value: PostOptions) -> Self {
-        PostOptions_ {
-            idempotency_key: value.idempotency_key,
-        }
-    }
-}
-
-#[derive(Clone, Deserialize, Serialize)]
-pub struct CreateMessageRequest {
-    pub app_id: String,
-    pub message: MessageIn,
-    #[serde(skip_serializing_if = "Option::is_none")]
-    pub post_options: Option<PostOptions>,
-}
-
-async fn create_svix_message(svix: &Svix, value: JsObject) -> std::io::Result<()> {
-    let CreateMessageRequest {
+#[instrument(skip_all, level="error", fields(
+    app_id = app_id,
+    event_type = message.event_type
+))]
+async fn create_svix_message(
+    svix: &Svix,
+    CreateMessageRequest {
         app_id,
         message,
         post_options,
-    }: CreateMessageRequest = serde_json::from_value(value.into())?;
-    let span = tracing::error_span!(
-        "create_svix_message",
-        app_id = app_id,
-        event_type = message.event_type
-    );
-    let _enter = span.enter();
-
+    }: CreateMessageRequest,
+) -> std::io::Result<()> {
     svix.message()
         .create(app_id, message, post_options.map(Into::into))
         .await

bridge/svix-bridge-plugin-queue/src/receiver_output/mod.rs

@@ -9,7 +9,7 @@ use generic_queue::redis::{RedisConfig, RedisQueueBackend};
 use generic_queue::sqs::{SqsConfig, SqsQueueBackend};
 use generic_queue::{TaskQueueBackend, TaskQueueSend};
 use std::sync::Arc;
-use svix_bridge_types::{async_trait, JsObject, ReceiverOutput};
+use svix_bridge_types::{async_trait, ForwardRequest, ReceiverOutput};
 
 type Result<T> = std::result::Result<T, Error>;
 
@@ -120,11 +120,9 @@ impl ReceiverOutput for QueueForwarder {
     fn name(&self) -> &str {
         &self.name
     }
-    async fn handle(&self, payload: JsObject) -> std::io::Result<()> {
+    async fn handle(&self, request: ForwardRequest) -> std::io::Result<()> {
         self.sender
-            // FIXME(#5762): the payload to publish should be coming from a specific field in the
-            //   Object, not the whole thing.
-            .send(serde_json::Value::Object(payload))
+            .send(request.payload)
             .await
             .map_err(crate::Error::from)?;
         Ok(())

bridge/svix-bridge-plugin-queue/tests/gcp_pubsub_consumer.rs

@@ -10,13 +10,11 @@ use google_cloud_pubsub::topic::Topic;
 use std::time::Duration;
 
 use serde_json::json;
-use svix_bridge_plugin_queue::{
-    config::GCPPubSubInputOpts, CreateMessageRequest, GCPPubSubConsumerPlugin,
-};
+use svix_bridge_plugin_queue::{config::GCPPubSubInputOpts, GCPPubSubConsumerPlugin};
 use svix_bridge_types::{
-    svix::api::MessageIn, SenderInput, SenderOutputOpts, SvixOptions, SvixSenderOutputOpts,
-    TransformationConfig, TransformerInput, TransformerInputFormat, TransformerJob,
-    TransformerOutput,
+    svix::api::MessageIn, CreateMessageRequest, SenderInput, SenderOutputOpts, SvixOptions,
+    SvixSenderOutputOpts, TransformationConfig, TransformerInput, TransformerInputFormat,
+    TransformerJob, TransformerOutput,
 };
 use wiremock::matchers::{body_partial_json, method};
 use wiremock::{Mock, MockServer, ResponseTemplate};
@@ -286,7 +284,7 @@ async fn test_consume_transformed_string_ok() {
             };
             // Build a create-message-compatible object, using the string input as a field in the payload.
             let out = json!({
-                "app_id": "app_1234",
+                "appId": "app_1234",
                 "message": {
                     "eventType": "testing.things",
                     "payload": {
@@ -396,7 +394,7 @@ async fn test_missing_event_type_nack() {
     publish(
         &topic,
         &serde_json::to_string(&json!({
-            "app_id": "app_1234",
+            "appId": "app_1234",
             "message": {
                 // No event type
                 "payload": {

bridge/svix-bridge-plugin-queue/tests/rabbitmq_consumer.rs

@@ -6,13 +6,11 @@ use generic_queue::rabbitmq::FieldTable;
 use lapin::{options::QueueDeclareOptions, Channel, Connection, ConnectionProperties, Queue};
 use serde_json::json;
 use std::time::Duration;
-use svix_bridge_plugin_queue::{
-    config::RabbitMqInputOpts, CreateMessageRequest, RabbitMqConsumerPlugin,
-};
+use svix_bridge_plugin_queue::{config::RabbitMqInputOpts, RabbitMqConsumerPlugin};
 use svix_bridge_types::{
-    svix::api::MessageIn, SenderInput, SenderOutputOpts, SvixOptions, SvixSenderOutputOpts,
-    TransformationConfig, TransformerInput, TransformerInputFormat, TransformerJob,
-    TransformerOutput,
+    svix::api::MessageIn, CreateMessageRequest, SenderInput, SenderOutputOpts, SvixOptions,
+    SvixSenderOutputOpts, TransformationConfig, TransformerInput, TransformerInputFormat,
+    TransformerJob, TransformerOutput,
 };
 use wiremock::matchers::{body_partial_json, method};
 use wiremock::{Mock, MockServer, ResponseTemplate};
@@ -270,7 +268,7 @@ async fn test_consume_transformed_string_ok() {
             };
             // Build a create-message-compatible object, using the string input as a field in the payload.
             let out = json!({
-                "app_id": "app_1234",
+                "appId": "app_1234",
                 "message": {
                     "eventType": "testing.things",
                     "payload": {
@@ -390,7 +388,7 @@ async fn test_missing_event_type_nack() {
         &channel,
         queue_name,
         &serde_json::to_vec(&json!({
-            "app_id": "app_1234",
+            "appId": "app_1234",
             "message": {
                 // No event type
                 "payload": {

bridge/svix-bridge-plugin-queue/tests/redis_stream_consumer.rs

@@ -5,11 +5,11 @@ use std::time::Duration;
 
 use redis::{AsyncCommands, Client};
 use serde_json::json;
-use svix_bridge_plugin_queue::{config::RedisInputOpts, CreateMessageRequest, RedisConsumerPlugin};
+use svix_bridge_plugin_queue::{config::RedisInputOpts, RedisConsumerPlugin};
 use svix_bridge_types::{
-    svix::api::MessageIn, SenderInput, SenderOutputOpts, SvixOptions, SvixSenderOutputOpts,
-    TransformationConfig, TransformerInput, TransformerInputFormat, TransformerJob,
-    TransformerOutput,
+    svix::api::MessageIn, CreateMessageRequest, SenderInput, SenderOutputOpts, SvixOptions,
+    SvixSenderOutputOpts, TransformationConfig, TransformerInput, TransformerInputFormat,
+    TransformerJob, TransformerOutput,
 };
 use wiremock::matchers::{body_partial_json, method};
 use wiremock::{Mock, MockServer, ResponseTemplate};
@@ -243,7 +243,7 @@ async fn test_consume_transformed_string_ok() {
             };
             // Build a create-message-compatible object, using the string input as a field in the payload.
             let out = json!({
-                "app_id": "app_1234",
+                "appId": "app_1234",
                 "message": {
                     "eventType": "testing.things",
                     "payload": {
@@ -353,7 +353,7 @@ async fn test_missing_event_type_nack() {
         &client,
         &key,
         &serde_json::to_string(&json!({
-            "app_id": "app_1234",
+            "appId": "app_1234",
             "message": {
                 // No event type
                 "payload": {

bridge/svix-bridge-plugin-queue/tests/sqs_consumer.rs

@@ -7,11 +7,12 @@ use std::time::Duration;
 
 use aws_sdk_sqs::Client;
 use serde_json::json;
-use svix_bridge_plugin_queue::{config::SqsInputOpts, CreateMessageRequest, SqsConsumerPlugin};
+use svix_bridge_plugin_queue::{config::SqsInputOpts, SqsConsumerPlugin};
 use svix_bridge_types::svix::api::MessageIn;
 use svix_bridge_types::{
-    SenderInput, SenderOutputOpts, SvixOptions, SvixSenderOutputOpts, TransformationConfig,
-    TransformerInput, TransformerInputFormat, TransformerJob, TransformerOutput,
+    CreateMessageRequest, SenderInput, SenderOutputOpts, SvixOptions, SvixSenderOutputOpts,
+    TransformationConfig, TransformerInput, TransformerInputFormat, TransformerJob,
+    TransformerOutput,
 };
 use wiremock::matchers::{body_partial_json, method};
 use wiremock::{Mock, MockServer, ResponseTemplate};
@@ -261,7 +262,7 @@ async fn test_consume_transformed_string_ok() {
             };
             // Build a create-message-compatible object, using the string input as a field in the payload.
             let out = json!({
-                "app_id": "app_1234",
+                "appId": "app_1234",
                 "message": {
                     "eventType": "testing.things",
                     "payload": {
@@ -385,7 +386,7 @@ async fn test_missing_event_type_nack() {
         &client,
         &queue_url,
         &serde_json::to_string(&json!({
-            "app_id": "app_1234",
+            "appId": "app_1234",
             "message": {
                 // No event type
                 "payload": {

bridge/svix-bridge-types/src/lib.rs

@@ -1,7 +1,7 @@
 pub use async_trait::async_trait;
 use serde::{Deserialize, Serialize};
 pub use svix;
-use svix::api::SvixOptions as _SvixOptions;
+use svix::api::{MessageIn, PostOptions as PostOptions_, SvixOptions as _SvixOptions};
 use tokio::sync::mpsc;
 use tokio::sync::oneshot;
 
@@ -106,11 +106,8 @@ pub struct TransformerJob {
 
 pub enum TransformerOutput {
     /// A successfully transformed payload.
-    // XXX: not sure if there's a cheaper way to deserialize the output while requiring an Object.
-    // FIXME(#5762): We can define a fixed type here as the expected output for a transformation.
-    //   It think it'll always be json, but the `payload` field can be any json-encoded type.
-    //   That type, whatever it is, will be required as output regardless of if a transformation is
-    //   used or not, probably, replacing `JsObject` as the interchange value.
+    // Both senders and receivers require a map type (Object) but have different requirements which
+    // are best validated after the fact. For now, we validate only that we get a map type back.
     Object(JsObject),
     /// For cases where the JS script executes successfully but produces an unexpected output.
     Invalid,
@@ -149,7 +146,7 @@ pub trait SenderInput: Send {
 #[async_trait]
 pub trait ReceiverOutput: Send + Sync {
     fn name(&self) -> &str;
-    async fn handle(&self, payload: JsObject) -> std::io::Result<()>;
+    async fn handle(&self, request: ForwardRequest) -> std::io::Result<()>;
 }
 
 #[derive(Deserialize, Debug, Clone, Default)]
@@ -214,3 +211,41 @@ pub struct SvixSenderOutputOpts {
     #[serde(default)]
     pub options: Option<SvixOptions>,
 }
+
+#[derive(Clone, Default, Deserialize, Serialize)]
+pub struct PostOptions {
+    idempotency_key: Option<String>,
+}
+
+impl From<PostOptions> for PostOptions_ {
+    fn from(value: PostOptions) -> Self {
+        PostOptions_ {
+            idempotency_key: value.idempotency_key,
+        }
+    }
+}
+
+/// Senders convert messages into Create Message API calls so the JSON pulled out of message queues
+/// or produced by transformations need to conform to this shape.
+#[derive(Clone, Deserialize, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct CreateMessageRequest {
+    pub app_id: String,
+    pub message: MessageIn,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub post_options: Option<PostOptions>,
+}
+
+/// Receivers convert HTTP bodies into messages forwarded to (currently only) message queues, etc.
+/// The `payload` field represents the message body given to the producer, and other fields may be
+/// added in the future allowing transformations to dynamically customize the producer behavior.
+#[derive(Clone, Deserialize, Serialize)]
+pub struct ForwardRequest {
+    /// This is the payload that will be fed into a Receiver Output
+    // XXX: right now I think any arbitrary json value can work, but individual outputs may have
+    // more strict requirements.
+    // The fact this is represented as a field on a json object demands at least that the value can
+    // be represented in json.
+    // FIXME: can we leverage RawValue here?
+    pub payload: serde_json::Value,
+}