add js transforms to bridge (#917)

Embeds a deno runtime in the bridge binary to allow ad hoc reshaping of
messages read from queues before they are sent to Svix.

## Motivation

When consuming JSON from messaging systems to generate new webhooks,
it's common to need to transform the payload before making the Create
Message request to Svix.

## Solution

Embedding a JS runtime into the bridge process allows us to run
user-defined functions on the payloads as they move from input to
output.

With this diff, plugin instances provided by
`svix-webhook-bridge-plugin-queue-consumer` can now evaluate js included
in the plugin config:
```yaml
plugins:
- type: "rabbitmqconsumer"
  input:
    uri: "amqp://guest:guest@localhost:5672/%2f"
    queue_name: "local"
    requeue_on_nack: false
  transformation: | 
    function handler(input) {
      return { 
        app_id: input.key, 
        message: { 
          eventType: input.event_type, 
          payload: input.data 
        }
      };
    }
  output:
    token: "***************"
```

The `transformation` key can now be set on any of these consumers. The
JS fragment should include a default export of a function that accepts
an object and returns an object.

These functions can be used to reshape the payload as necessary.

Author: svix-onelson

webhook-bridge/Cargo.lock

@@ -1,4 +1,10 @@
 [workspace]
+# Earlier versions of deno fail to compile in a workspace because of wgpu-hal
+# The "fix" is to enable resolver 2 at the workspace. Crates with edition 2021
+# use this by default, but workspaces are set independently for some reason.
+# <https://stackoverflow.com/questions/73967574/dependency-fails-to-compile-in-workspace-but-not-in-package>
+resolver = "2"
+
 members = [
     "generic-queue",
     "svix-webhook-bridge-types",

webhook-bridge/Cargo.toml

@@ -33,12 +33,39 @@ Currently this supports the following messaging systems:
 - Redis
 - SQS
 
-Generally instances of this plugin are configured in terms of inputs and outputs, where the input configuration varies
-by the messaging system.
+Generally instances of this plugin are configured in terms of inputs, _optional transformations_, and outputs, where
+the input configuration varies by the messaging system.
 
 The output options control how the Svix client is built and configured.
 The sole required field is `token`.
 
+The optional _transformation_ can be set to a JavaScript fragment which can be used to reshape the messages as they flow through.
+
+```yaml
+
+plugins:
+- type: ...
+  input:
+    # ... snip ...
+
+  # Reshape the messages we get from the queue before they get sent to Svix
+  transformation: | 
+    function handler(input) {
+      return {
+        app_id: input.key,
+        message: {
+          eventType: input.event_type,
+          payload: input.data
+        }
+      };
+    }
+
+  output:
+    # ... snip ...
+```
+
+Transformations should define a function called `handler` that accepts an object and returns an object.
+
 Messages received by these consumers must follow an expected format:
 
 ```

webhook-bridge/README.md

@@ -24,7 +24,7 @@ serde = { version = "1", features = ["derive", "rc"] }
 serde_json = "1"
 thiserror = "1"
 tokio = { version = "1", features = ["full"] }
-tokio-util = { version = "0.7.8", optional = true }
+tokio-util = { version = "0.7", optional = true }
 futures-util = { version = "0.3.28", optional = true }
 tracing = "0.1"
 

webhook-bridge/generic-queue/Cargo.toml

@@ -6,24 +6,28 @@ use svix::api::SvixOptions as _SvixOptions;
 #[derive(Debug, Default, Deserialize)]
 pub struct RabbitMqConsumerConfig {
     pub input: RabbitMqInputOpts,
+    pub transformation: Option<String>,
     pub output: OutputOpts,
 }
 
 #[derive(Debug, Default, Deserialize)]
 pub struct RedisConsumerConfig {
     pub input: RedisInputOpts,
+    pub transformation: Option<String>,
     pub output: OutputOpts,
 }
 
 #[derive(Debug, Default, Deserialize)]
 pub struct SqsConsumerConfig {
     pub input: SqsInputOpts,
+    pub transformation: Option<String>,
     pub output: OutputOpts,
 }
 
 #[derive(Debug, Default, Deserialize)]
 pub struct GCPPubSubConsumerConfig {
     pub input: GCPPubSubInputOpts,
+    pub transformation: Option<String>,
     pub output: OutputOpts,
 }
 

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

@@ -5,6 +5,7 @@ pub enum Error {
     Json(serde_json::Error),
     Queue(QueueError),
     Svix(svix::error::Error),
+    Generic(String),
 }
 
 impl From<svix::error::Error> for Error {
@@ -25,13 +26,20 @@ impl From<QueueError> for Error {
     }
 }
 
+impl From<String> for Error {
+    fn from(value: String) -> Self {
+        Self::Generic(value)
+    }
+}
+
 impl From<Error> for std::io::Error {
     fn from(value: Error) -> Self {
         match value {
             Error::Payload(e) => std::io::Error::new(std::io::ErrorKind::Other, e),
             Error::Json(e) => std::io::Error::new(std::io::ErrorKind::Other, e),
             Error::Queue(e) => std::io::Error::new(std::io::ErrorKind::Other, e),
             Error::Svix(e) => std::io::Error::new(std::io::ErrorKind::Other, e),
+            Error::Generic(e) => std::io::Error::new(std::io::ErrorKind::Other, e),
         }
     }
 }

webhook-bridge/svix-webhook-bridge-plugin-queue-consumer/src/error.rs

@@ -1,27 +1,37 @@
 use crate::config::{GCPPubSubConsumerConfig, GCPPubSubInputOpts};
 use crate::error::Error;
-use crate::PLUGIN_NAME;
 use crate::PLUGIN_VERS;
 use crate::{create_svix_message, CreateMessageRequest};
+use crate::{run_inner, Consumer, PLUGIN_NAME};
 use generic_queue::gcp_pubsub::{
     GCPPubSubConfig, GCPPubSubDelivery, GCPPubSubQueueBackend, GCPPubSubQueueConsumer,
 };
 use generic_queue::{Delivery, TaskQueueBackend, TaskQueueReceive};
-use std::time::{Duration, Instant};
+use std::time::Duration;
 use svix::api::Svix;
-use svix_webhook_bridge_types::{async_trait, Plugin};
+use svix_webhook_bridge_types::{async_trait, JsObject, Plugin, TransformerTx};
 use tracing::instrument;
 
 pub struct GCPPubSubConsumerPlugin {
     input_options: GCPPubSubInputOpts,
     svix_client: Svix,
+    transformer_tx: Option<TransformerTx>,
+    transformation: Option<String>,
 }
 
 impl GCPPubSubConsumerPlugin {
-    pub fn new(GCPPubSubConsumerConfig { input, output }: GCPPubSubConsumerConfig) -> Self {
+    pub fn new(
+        GCPPubSubConsumerConfig {
+            input,
+            transformation,
+            output,
+        }: GCPPubSubConsumerConfig,
+    ) -> Self {
         Self {
             input_options: input,
             svix_client: Svix::new(output.token, output.svix_options.map(Into::into)),
+            transformer_tx: None,
+            transformation,
         }
     }
 
@@ -49,10 +59,10 @@ impl GCPPubSubConsumerPlugin {
     }
 
     /// Parses the delivery as JSON and feeds it into [`create_svix_message`].
-    /// Will nack the delivery if either the JSON parse step, or the request to svix fails.
+    /// Will nack the delivery if either the JSON parse, transformation, or the request to svix fails.
     #[instrument(skip_all, fields(messaging.operation = "process"))]
-    async fn process(&self, delivery: GCPPubSubDelivery<serde_json::Value>) -> std::io::Result<()> {
-        let payload = match Delivery::<serde_json::Value>::payload(&delivery) {
+    async fn process(&self, delivery: GCPPubSubDelivery<JsObject>) -> std::io::Result<()> {
+        let payload = match Delivery::<JsObject>::payload(&delivery) {
             Ok(p) => p,
             Err(e) => {
                 tracing::warn!("{e}");
@@ -61,6 +71,19 @@ impl GCPPubSubConsumerPlugin {
             }
         };
 
+        let payload = if let Some(script) = &self.transformation {
+            match self.transform(script.clone(), payload).await {
+                Err(e) => {
+                    tracing::error!("nack: {e}");
+                    delivery.nack().await.map_err(Error::from)?;
+                    return Ok(());
+                }
+                Ok(x) => x,
+            }
+        } else {
+            payload
+        };
+
         match create_svix_message(&self.svix_client, payload).await {
             Ok(_) => {
                 tracing::trace!("ack");
@@ -73,7 +96,13 @@ impl GCPPubSubConsumerPlugin {
         }
         Ok(())
     }
+}
 
+#[async_trait]
+impl Consumer for GCPPubSubConsumerPlugin {
+    fn transformer_tx(&self) -> Option<&TransformerTx> {
+        self.transformer_tx.as_ref()
+    }
     async fn consume(&self) -> std::io::Result<()> {
         let mut consumer =
             <GCPPubSubQueueBackend as TaskQueueBackend<CreateMessageRequest>>::consuming_half(
@@ -86,7 +115,6 @@ impl GCPPubSubConsumerPlugin {
             )
             .await
             .map_err(Error::from)?;
-
         tracing::debug!(
             "gcp pubsub consuming: {}",
             &self.input_options.subscription_id
@@ -100,43 +128,17 @@ impl GCPPubSubConsumerPlugin {
 
 impl TryInto<Box<dyn Plugin>> for GCPPubSubConsumerConfig {
     type Error = &'static str;
-
     fn try_into(self) -> Result<Box<dyn Plugin>, Self::Error> {
         Ok(Box::new(GCPPubSubConsumerPlugin::new(self)))
     }
 }
 
 #[async_trait]
 impl Plugin for GCPPubSubConsumerPlugin {
+    fn set_transformer(&mut self, tx: Option<TransformerTx>) {
+        self.transformer_tx = tx;
+    }
     async fn run(&self) -> std::io::Result<()> {
-        let mut fails: u64 = 0;
-        let mut last_fail = Instant::now();
-
-        tracing::info!(
-            "gcp pubsub starting: {}",
-            &self.input_options.subscription_id
-        );
-
-        loop {
-            if let Err(e) = self.consume().await {
-                tracing::error!("{e}");
-            }
-
-            tracing::error!(
-                "gcp pubsub disconnected: {}",
-                &self.input_options.subscription_id
-            );
-
-            if last_fail.elapsed() > Duration::from_secs(10) {
-                // reset the fail count if we didn't have a hiccup in the past short while.
-                tracing::trace!("been a while since last fail, resetting count");
-                fails = 0;
-            } else {
-                fails += 1;
-            }
-
-            last_fail = Instant::now();
-            tokio::time::sleep(Duration::from_millis((300 * fails).min(3000))).await;
-        }
+        run_inner(self, "gcp subsub", &self.input_options.subscription_id).await
     }
 }