integrate bridge js transformations with webhook-receiver plugin (#933)

Updates the webhook-receiver plugin to accept a `TransformerTx` so it
can leverage the JS Runtime managed by bridge.

This is not the best diff, sadly.

The original webhook-ingester made some design decisions that made this
a little awkward. Specifically, the request would accept a forwarder on
its `forward` method which took responsibility for payload handling.

When transformations are optionally happening, the payload either needs
to transform as a part of the `forward` method, or (my preference) the
payload handling needs to be moved up to the handler level.

I went for the latter - The "forward" method now hangs off the forwarder
and receives a payload. This means the handler can attempt the
transformation before forwarding, which keeps the go/no-go logic all
within the handler body. It also avoids having to somehow pass app state
down into the forwarder so it could perform the transformation itself.

Unfortunately, the go/no-go logic in the handler is also sort of a mess.
In order to tame the additional mess contributed by the transformation,
that chunk of the work has been extracted into a helper function.

---

Ideally, we'd have some actual unit tests around the HTTP handler code,
and the bits invoked by it, but since we don't yet have anything setup
it'd be a bigger lift than I care to tackle right now.

Instead, manual testing was done, noted in the comments below.

Author: svix-onelson

webhook-bridge/svix-webhook-bridge-plugin-webhook-receiver/src/config.rs

@@ -13,6 +13,8 @@ pub struct IntegrationConfig {
     pub name: IntegrationId,
     pub verification: VerificationScheme,
     pub destination: ForwardDestination,
+    #[serde(default)]
+    pub transformation: Option<String>,
 }
 
 /// The [`VerificationScheme`] is an enum with variant for every method for verifying a webhook's

webhook-bridge/svix-webhook-bridge-plugin-webhook-receiver/src/forwarding.rs

@@ -1,5 +1,4 @@
 use crate::config::{GCPPubSubOutputOpts, RabbitMqOutputOpts, RedisOutputOpts, SqsOutputOpts};
-use crate::types::{SerializablePayload, SerializableRequest, Validated};
 use anyhow::Result;
 use axum::async_trait;
 use enum_dispatch::enum_dispatch;
@@ -9,11 +8,12 @@ use generic_queue::redis::{RedisConfig, RedisQueueBackend};
 use generic_queue::sqs::{SqsConfig, SqsQueueBackend};
 use generic_queue::{TaskQueueBackend, TaskQueueSend};
 use std::sync::Arc;
+use svix_webhook_bridge_types::JsObject;
 
 #[async_trait]
 #[enum_dispatch]
 pub trait ForwardingMethod {
-    async fn forward(&self, req: SerializableRequest<Validated>) -> Result<http::StatusCode>;
+    async fn forward(&self, payload: JsObject) -> Result<http::StatusCode>;
 }
 
 #[derive(Clone)]
@@ -101,13 +101,8 @@ impl std::fmt::Debug for GenericQueueForwarder {
 }
 #[async_trait]
 impl ForwardingMethod for GenericQueueForwarder {
-    async fn forward(&self, req: SerializableRequest<Validated>) -> Result<http::StatusCode> {
-        let payload = match req.payload() {
-            SerializablePayload::Standard(data) => serde_json::from_slice(data)?,
-            SerializablePayload::StringSerializable(s) => serde_json::from_str(s)?,
-        };
-
-        self.sender.send(payload).await?;
+    async fn forward(&self, payload: JsObject) -> Result<http::StatusCode> {
+        self.sender.send(payload.into()).await?;
         Ok(http::StatusCode::OK)
     }
 }

webhook-bridge/svix-webhook-bridge-plugin-webhook-receiver/src/lib.rs

@@ -4,9 +4,12 @@ use axum::{
     routing::post,
     Router,
 };
+use forwarding::ForwardingMethod;
 use serde::Deserialize;
 use std::net::SocketAddr;
-use svix_webhook_bridge_types::{async_trait, Plugin};
+use svix_webhook_bridge_types::{
+    async_trait, JsObject, JsReturn, Plugin, TransformerJob, TransformerTx,
+};
 use tracing::instrument;
 use types::{IntegrationId, IntegrationState, InternalState, SerializableRequest, Unvalidated};
 
@@ -24,14 +27,18 @@ pub struct WebhookReceiverPluginConfig {
     pub routes: Vec<IntegrationConfig>,
 }
 
-#[derive(Clone, Debug, Deserialize, PartialEq)]
+#[derive(Clone, Debug)]
 pub struct WebhookReceiverPlugin {
     cfg: WebhookReceiverPluginConfig,
+    transformer_tx: Option<TransformerTx>,
 }
 
 impl WebhookReceiverPlugin {
     pub fn new(cfg: WebhookReceiverPluginConfig) -> Self {
-        Self { cfg }
+        Self {
+            cfg,
+            transformer_tx: None,
+        }
     }
 }
 
@@ -45,11 +52,16 @@ impl TryInto<Box<dyn Plugin>> for WebhookReceiverPluginConfig {
 
 #[async_trait]
 impl Plugin for WebhookReceiverPlugin {
+    fn set_transformer(&mut self, tx: Option<TransformerTx>) {
+        self.transformer_tx = tx;
+    }
+
     async fn run(&self) -> std::io::Result<()> {
         let addr = &self.cfg.listen_addr;
-        let state = InternalState::from_routes(self.cfg.routes.as_slice())
-            .await
-            .map_err(|e| std::io::Error::new(std::io::ErrorKind::Other, e))?;
+        let state =
+            InternalState::from_routes(self.cfg.routes.as_slice(), self.transformer_tx.clone())
+                .await
+                .map_err(|e| std::io::Error::new(std::io::ErrorKind::Other, e))?;
 
         let router = Router::new()
             .route(
@@ -81,18 +93,47 @@ impl Plugin for WebhookReceiverPlugin {
 )]
 async fn route(
     Path(integration_id): Path<IntegrationId>,
-    State(InternalState(id_map)): State<InternalState>,
+    State(InternalState {
+        routes,
+        transformer_tx,
+    }): State<InternalState>,
     req: SerializableRequest<Unvalidated>,
 ) -> http::StatusCode {
     if let Some(IntegrationState {
         verifier,
         forwarder,
-    }) = id_map.get(&integration_id)
+        transformation,
+    }) = routes.get(&integration_id)
     {
         match req.validate(verifier).await {
             Ok(req) => {
+                let payload = match req.payload().as_js_object() {
+                    Ok(payload) => match transformation.clone() {
+                        Some(script) => {
+                            match transform(payload, script, transformer_tx.clone()).await {
+                                Ok(transformed_payload) => transformed_payload,
+                                Err(c) => return c,
+                            }
+                        }
+                        // Keep the original payload as-is if there's no transformation specified.
+                        None => payload,
+                    },
+                    Err(e) => {
+                        tracing::error!("failed to parse payload as json object: {}", e);
+                        return http::StatusCode::BAD_REQUEST;
+                    }
+                };
+
                 tracing::debug!("forwarding request");
-                req.forward(forwarder).await
+                // `forward` method was ambiguous. It looks like there are many traits that offer
+                // this method.
+                match ForwardingMethod::forward(forwarder, payload).await {
+                    Ok(c) => c,
+                    Err(e) => {
+                        tracing::error!("Error forwarding request: {}", e);
+                        http::StatusCode::INTERNAL_SERVER_ERROR
+                    }
+                }
             }
             Err(code) => {
                 tracing::warn!("validation failed: {code}");
@@ -104,3 +145,35 @@ async fn route(
         http::StatusCode::NOT_FOUND
     }
 }
+
+/// Attempts to run the payload through a js transformation.
+async fn transform(
+    payload: JsObject,
+    script: String,
+    tx: Option<TransformerTx>,
+) -> Result<JsObject, http::StatusCode> {
+    let tx = tx.ok_or_else(|| {
+        tracing::error!("transformations are not available");
+        http::StatusCode::INTERNAL_SERVER_ERROR
+    })?;
+
+    let (job, callback) = TransformerJob::new(script.clone(), payload);
+    if let Err(e) = tx.send(job) {
+        tracing::error!("transformations are not available: {}", e);
+        return Err(http::StatusCode::INTERNAL_SERVER_ERROR);
+    }
+
+    match callback.await {
+        // This is the only "good" outcome giving a RHS value for the assignment.
+        // All other match arms should bail with a non-2xx status.
+        Ok(Ok(JsReturn::Object(obj))) => Ok(obj),
+        Ok(Ok(JsReturn::Invalid)) => {
+            tracing::error!("transformation produced invalid payload");
+            Err(http::StatusCode::INTERNAL_SERVER_ERROR)
+        }
+        _ => {
+            tracing::error!("transformation failed");
+            Err(http::StatusCode::INTERNAL_SERVER_ERROR)
+        }
+    }
+}

webhook-bridge/svix-webhook-bridge-plugin-webhook-receiver/src/types.rs

@@ -2,7 +2,7 @@ use crate::config::IntegrationConfig;
 use crate::forwarding::GenericQueueForwarder;
 use crate::{
     config::{ForwardDestination, VerificationScheme},
-    forwarding::{Forwarder, ForwardingMethod},
+    forwarding::Forwarder,
     verification::{NoVerifier, SvixVerifier, VerificationMethod, Verifier},
 };
 use anyhow::Result;
@@ -15,13 +15,21 @@ use axum::{
 use http::{HeaderMap, HeaderValue, Request};
 use serde::{Deserialize, Serialize};
 use std::{collections::HashMap, marker::PhantomData, sync::Arc};
+use svix_webhook_bridge_types::{JsObject, TransformerTx};
 
 #[derive(Clone, Debug)]
 /// The [`InternalState`] is passed to the Axum route and is used to map the "IntegrationId" in the
 /// URL to the configured [`Verifier`] and [`Forwarder`] variants.
-pub struct InternalState(pub HashMap<IntegrationId, IntegrationState>);
+pub struct InternalState {
+    pub routes: HashMap<IntegrationId, IntegrationState>,
+    pub transformer_tx: Option<TransformerTx>,
+}
+
 impl InternalState {
-    pub async fn from_routes(routes: &[IntegrationConfig]) -> Result<Self> {
+    pub async fn from_routes(
+        routes: &[IntegrationConfig],
+        transformer_tx: Option<TransformerTx>,
+    ) -> Result<Self> {
         let mut state_map = HashMap::new();
 
         for cfg in routes {
@@ -57,11 +65,15 @@ impl InternalState {
                 IntegrationState {
                     verifier,
                     forwarder,
+                    transformation: cfg.transformation.clone(),
                 },
             );
         }
 
-        Ok(InternalState(state_map))
+        Ok(InternalState {
+            routes: state_map,
+            transformer_tx,
+        })
     }
 }
 
@@ -99,6 +111,7 @@ impl AsRef<str> for IntegrationId {
 pub struct IntegrationState {
     pub verifier: Verifier,
     pub forwarder: Forwarder,
+    pub transformation: Option<String>,
 }
 
 /// Any arbitrary HTTP request which is not a webhook dispatched by Svix may also have arbitrary
@@ -282,18 +295,6 @@ impl SerializableRequest<Unvalidated> {
     }
 }
 
-impl SerializableRequest<Validated> {
-    pub async fn forward<F: ForwardingMethod>(self, f: &F) -> http::StatusCode {
-        match f.forward(self).await {
-            Ok(c) => c,
-            Err(e) => {
-                tracing::error!("Error forwarding request: {}", e);
-                http::StatusCode::INTERNAL_SERVER_ERROR
-            }
-        }
-    }
-}
-
 #[derive(Clone, Debug)]
 pub enum SerializableHeaderMap {
     Standard(HeaderMap),
@@ -385,4 +386,11 @@ impl SerializablePayload {
             Self::StringSerializable(s) => Ok(Self::StringSerializable(s)),
         }
     }
+
+    pub fn as_js_object(&self) -> Result<JsObject> {
+        Ok(match self {
+            Self::Standard(v) => serde_json::from_slice(v)?,
+            Self::StringSerializable(s) => serde_json::from_str(s)?,
+        })
+    }
 }