Bridge: support raw inputs for transformations

Adds a bunch of plumbing to allow the message payloads from consumers to
be passed to transformations as strings rather than pre-parsing as JSON
in rust.

One concrete use case for this is if the system publishing those
messages to the queue is sending XML or perhaps base64 encoded data.

When configured to specify the `format` of a transformation, you can set
it to `string` rather than the default `json`. Then, when the
transformation receives that string as an argument, it can do whatever
parsing is required in order to return a valid `Object`.

Tests have been added up and down the stack to show that strings are
handled correctly, and that the new config data is propagated.

Notable omissions:

- redis relies on JSON values internally and can't easily support "raw"
  inputs. We can revisit this when we switch to `omniqueue` since I
  believe it handles this better.

Author: svix-onelson

bridge/README.md

@@ -27,6 +27,75 @@ publish the payload on to a specified "output."
 
 Both **senders** and **receivers** are defined in terms of their input, and optional JavaScript transformation, and their output.
 
+---
+
+Transformations are configured as either a single string of JS source code:
+
+```yaml
+transformation: |
+    function handler(input) {
+      return {
+        app_id: input.key,
+        message: {
+          eventType: input.event_type,
+          payload: input.data
+        }
+      };
+    }
+```
+In this situation, the input value will be pre-parsed as JSON.
+
+You can also configure transformations using the verbose form, allowing control of the `format`:
+```yaml
+transformation:
+  format: "json"
+  src: |
+    function handler(input) {
+      return {
+        app_id: input.key,
+        message: {
+          eventType: input.event_type,
+          payload: input.data
+        }
+      };
+    }
+```
+The `format` field can be set to either `string` (ie. the _raw payload_) or `json` (the default behavior).
+
+For example, using the string `format` you can parse data out of the input using whatever method you like:
+
+```yaml
+transformation:
+  # Let's say the input is an XML message like:
+  # `<msg appId="app_123" eventType="a.b"><payload>{"x": 123}</payload></msg>`
+
+  format: "string"
+  src: |
+    function handler(input) {
+      let parser = new DOMParser();
+      let doc = parser.parseFromString(input, "text/xml");
+      let msg = doc.firstChild;
+      let payload = JSON.parse(msg.getElementsByTagName("payload")[0].textContent)
+
+      return {
+        app_id: msg.attributes.appId,
+        message: {
+          eventType: msg.attributes.eventType,
+          payload,
+        }
+      };
+    }
+```
+Transformations must define a function named `handler` which receives a single argument, the type of which is controlled
+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
+> a `redis` input or output.
+
+---
+
 Currently the supported Sender inputs and Receiver outputs are the following
 messaging systems:
 

bridge/generic-queue/src/gcp_pubsub.rs

@@ -186,6 +186,10 @@ impl<T: DeserializeOwned + Send + Serialize + Sync> Delivery<T> for GCPPubSubDel
         serde_json::from_slice(&self.message.message.data).map_err(Into::into)
     }
 
+    fn raw_payload(&self) -> Result<&str, QueueError> {
+        std::str::from_utf8(&self.message.message.data).map_err(QueueError::generic)
+    }
+
     async fn ack(self) -> Result<(), QueueError> {
         self.message.ack().await.map_err(QueueError::generic)
     }

bridge/generic-queue/src/lib.rs

@@ -98,14 +98,15 @@ pub trait TaskQueueBackend<T: Send + Sync> {
 pub trait Delivery<T: Send + Sync> {
     /// Returns a freshly deserialized instance of the contained payload.
     fn payload(&self) -> Result<T, QueueError>;
+    fn raw_payload(&self) -> Result<&str, QueueError>;
 
     /// ACKs this message, which, depending on what backend you are using, may be a NOOP, or it may
-    /// explicity acknowledge the successful processing the message.
+    /// explicitly acknowledge the successful processing the message.
     ///
     /// When ACKed, consumers will not see this exact message again.
     async fn ack(self) -> Result<(), QueueError>;
     /// NACKs this message, which, depending on what backend you are using, may be a NOOP, it may
-    /// explicitly mark a messaege as not acknowledged, or it may reinsert the message back into the
+    /// explicitly mark a message as not acknowledged, or it may reinsert the message back into the
     /// end of the queue.
     ///
     /// When NACKed, consumers of this queue will process the message again at some point.

bridge/generic-queue/src/memory_queue.rs

@@ -55,6 +55,11 @@ impl<T: 'static + Clone + Debug + DeserializeOwned + Send + Serialize + Sync> De
         Ok(self.payload.clone())
     }
 
+    fn raw_payload(&self) -> Result<&str, QueueError> {
+        // this backend is not used in bridge, and we will migrate to omniqueue anyway...
+        unimplemented!("not supported");
+    }
+
     async fn ack(self) -> Result<(), QueueError> {
         Ok(())
     }

bridge/generic-queue/src/rabbitmq.rs

@@ -128,6 +128,9 @@ impl<T: DeserializeOwned + Send + Serialize + Sync> Delivery<T> for RabbitMqDeli
         serde_json::from_slice(&self.body).map_err(Into::into)
     }
 
+    fn raw_payload(&self) -> Result<&str, QueueError> {
+        std::str::from_utf8(&self.body).map_err(QueueError::generic)
+    }
     async fn ack(self) -> Result<(), QueueError> {
         self.acker
             .ack(BasicAckOptions { multiple: false })

bridge/generic-queue/src/redis.rs

@@ -180,6 +180,15 @@ impl<
     fn payload(&self) -> Result<T, QueueError> {
         T::from_redis_stream_map(&self.body.map).map_err(Into::into)
     }
+    // FIXME: the redis stream queue impl requires values to be encoded a certain way (eg. inside
+    //   a "payload" key in a map).
+    //   This means producers and consumers need to agree on the encoding.
+    //   This also means we can't just accept raw messages published by other systems (as is the
+    //   expectation with these raw payloads).
+    //   Leaving unsupported until we migrate to `omniqueue`.
+    fn raw_payload(&self) -> Result<&str, QueueError> {
+        unimplemented!("raw payload not supported with redis backend")
+    }
 
     async fn ack(self) -> Result<(), QueueError> {
         let mut conn = self.redis.get().await.map_err(QueueError::generic)?;

bridge/generic-queue/src/sqs.rs

@@ -106,6 +106,10 @@ impl<T: DeserializeOwned + Send + Serialize + Sync> Delivery<T> for SqsDelivery<
         serde_json::from_str(&self.body).map_err(Into::into)
     }
 
+    fn raw_payload(&self) -> Result<&str, QueueError> {
+        Ok(&self.body)
+    }
+
     async fn ack(self) -> Result<(), QueueError> {
         if let Some(receipt_handle) = self.receipt_handle {
             self.ack_client

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

@@ -7,58 +7,78 @@ use crate::{
     GCPPubSubConsumerPlugin, RabbitMqConsumerPlugin, RedisConsumerPlugin, SqsConsumerPlugin,
 };
 use serde::Deserialize;
-use svix_bridge_types::{ReceiverOutput, SenderInput, SenderOutputOpts};
+use svix_bridge_types::{
+    ReceiverOutput, SenderInput, SenderOutputOpts, TransformationConfig, TransformerInputFormat,
+};
 
 #[derive(Deserialize)]
 pub struct QueueConsumerConfig {
     pub name: String,
     pub input: SenderInputOpts,
     #[serde(default)]
-    pub transformation: Option<String>,
+    pub transformation: Option<TransformationConfig>,
     pub output: SenderOutputOpts,
 }
 
 impl QueueConsumerConfig {
-    pub fn into_sender_input(self) -> Box<dyn SenderInput> {
+    pub fn into_sender_input(self) -> Result<Box<dyn SenderInput>, &'static str> {
         match self.input {
-            SenderInputOpts::GCPPubSub(input) => Box::new(GCPPubSubConsumerPlugin::new(
-                self.name,
-                input,
-                self.transformation,
-                self.output,
-            )),
-            SenderInputOpts::RabbitMQ(input) => Box::new(RabbitMqConsumerPlugin::new(
+            SenderInputOpts::GCPPubSub(input) => Ok(Box::new(GCPPubSubConsumerPlugin::new(
                 self.name,
                 input,
                 self.transformation,
                 self.output,
-            )),
-            SenderInputOpts::Redis(input) => Box::new(RedisConsumerPlugin::new(
+            ))),
+            SenderInputOpts::RabbitMQ(input) => Ok(Box::new(RabbitMqConsumerPlugin::new(
                 self.name,
                 input,
                 self.transformation,
                 self.output,
-            )),
-            SenderInputOpts::SQS(input) => Box::new(SqsConsumerPlugin::new(
+            ))),
+            SenderInputOpts::Redis(input) => {
+                if let Some(xform) = &self.transformation {
+                    if xform.format() != TransformerInputFormat::Json {
+                        return Err("redis only supports json formatted transformations");
+                    }
+                }
+                Ok(Box::new(RedisConsumerPlugin::new(
+                    self.name,
+                    input,
+                    self.transformation,
+                    self.output,
+                )))
+            }
+            SenderInputOpts::SQS(input) => Ok(Box::new(SqsConsumerPlugin::new(
                 self.name,
                 input,
                 self.transformation,
                 self.output,
-            )),
+            ))),
         }
     }
 }
 
 pub async fn into_receiver_output(
     name: String,
     opts: ReceiverOutputOpts,
+    // Annoying to have to pass this, but certain backends (redis) only work with certain transformations (json).
+    transformation: &Option<TransformationConfig>,
 ) -> Result<Box<dyn ReceiverOutput>, crate::Error> {
     let forwarder = match opts {
         ReceiverOutputOpts::GCPPubSub(opts) => {
             QueueForwarder::from_gcp_pupsub_cfg(name, opts).await?
         }
         ReceiverOutputOpts::RabbitMQ(opts) => QueueForwarder::from_rabbitmq_cfg(name, opts).await?,
-        ReceiverOutputOpts::Redis(opts) => QueueForwarder::from_redis_cfg(name, opts).await?,
+        ReceiverOutputOpts::Redis(opts) => {
+            if let Some(t) = transformation {
+                if t.format() != TransformerInputFormat::Json {
+                    return Err(crate::Error::Generic(
+                        "redis only supports json formatted transformations".to_string(),
+                    ));
+                }
+            }
+            QueueForwarder::from_redis_cfg(name, opts).await?
+        }
         ReceiverOutputOpts::SQS(opts) => QueueForwarder::from_sqs_cfg(name, opts).await?,
     };
     Ok(Box::new(forwarder))
@@ -84,3 +104,73 @@ pub enum ReceiverOutputOpts {
     Redis(RedisOutputOpts),
     SQS(SqsOutputOpts),
 }
+
+#[cfg(test)]
+mod tests {
+    use super::into_receiver_output;
+    use super::QueueConsumerConfig;
+    use crate::config::{ReceiverOutputOpts, SenderInputOpts};
+    use crate::redis::{RedisInputOpts, RedisOutputOpts};
+    use svix_bridge_types::{
+        SenderOutputOpts, SvixSenderOutputOpts, TransformationConfig, TransformerInputFormat,
+    };
+
+    // FIXME: can't support raw payload access for redis because it requires JSON internally.
+    //   Revisit after `omniqueue` adoption.
+    #[test]
+    fn redis_sender_with_string_transformation_is_err() {
+        let cfg = QueueConsumerConfig {
+            name: "redis-with-string-transformation".to_string(),
+            input: SenderInputOpts::Redis(RedisInputOpts {
+                dsn: "".to_string(),
+                max_connections: 0,
+                reinsert_on_nack: false,
+                queue_key: "".to_string(),
+                consumer_group: "".to_string(),
+                consumer_name: "".to_string(),
+            }),
+            transformation: Some(TransformationConfig::Explicit {
+                format: TransformerInputFormat::String,
+                src: String::new(),
+            }),
+            output: SenderOutputOpts::Svix(SvixSenderOutputOpts {
+                token: "".to_string(),
+                options: None,
+            }),
+        };
+
+        assert_eq!(
+            cfg.into_sender_input()
+                .err()
+                .expect("invalid config didn't result in error"),
+            "redis only supports json formatted transformations"
+        )
+    }
+
+    // FIXME: can't support raw payload access for redis because it requires JSON internally.
+    //   Revisit after `omniqueue` adoption.
+    #[tokio::test]
+    async fn test_redis_receiver_string_transform_is_err() {
+        let redis_out = ReceiverOutputOpts::Redis(RedisOutputOpts {
+            dsn: "".to_string(),
+            max_connections: 0,
+            queue_key: "".to_string(),
+        });
+
+        // Explicit String fails
+        let res = into_receiver_output(
+            "".to_string(),
+            redis_out,
+            &Some(TransformationConfig::Explicit {
+                src: String::new(),
+                format: TransformerInputFormat::String,
+            }),
+        )
+        .await;
+        assert!(matches!(
+            res.err()
+                .expect("invalid config didn't result in error"),
+            crate::error::Error::Generic(msg) if msg == "redis only supports json formatted transformations"
+        ));
+    }
+}

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

@@ -6,7 +6,8 @@ use generic_queue::TaskQueueBackend;
 use serde::Deserialize;
 use std::path::PathBuf;
 use svix_bridge_types::{
-    async_trait, svix::api::Svix, JsObject, SenderInput, SenderOutputOpts, TransformerTx,
+    async_trait, svix::api::Svix, JsObject, SenderInput, SenderOutputOpts, TransformationConfig,
+    TransformerTx,
 };
 
 #[derive(Debug, Default, Deserialize)]
@@ -20,14 +21,14 @@ pub struct GCPPubSubConsumerPlugin {
     input_options: GCPPubSubInputOpts,
     svix_client: Svix,
     transformer_tx: Option<TransformerTx>,
-    transformation: Option<String>,
+    transformation: Option<TransformationConfig>,
 }
 
 impl GCPPubSubConsumerPlugin {
     pub fn new(
         name: String,
         input: GCPPubSubInputOpts,
-        transformation: Option<String>,
+        transformation: Option<TransformationConfig>,
         output: SenderOutputOpts,
     ) -> Self {
         Self {
@@ -58,7 +59,7 @@ impl Consumer for GCPPubSubConsumerPlugin {
         &self.transformer_tx
     }
 
-    fn transformation(&self) -> &Option<String> {
+    fn transformation(&self) -> &Option<TransformationConfig> {
         &self.transformation
     }