WIP Bridge: js-module sender input

The general idea here is to let folks define a js module to use as a
data source for a "sender."

This could be general purpose but the initial use case is for polling
APIs that don't expose webhooks.

By using `fetch` to issue HTTP requests, we can track state in the
module itself to decide if the data is worth sending a webhook for.

Example:
<https://github.com/svix-onelson/poller-input-poc/blob/main/fetcher.js>

There are many barriers to integrating the code in the POC linked above
which need to be cleared first.

We need:
- a custom module loader to let us source modules from the
  bridge config instead of js files on disk.
- Glue code to connect the `op_forward` deno extension calls to either a
  transformation or svix sender output.
- existing transformation code in bridge needs to be refactored to allow
  us to use "newer deno" without also introducing a memory leak.

For the latter, <svix/monorepo-private#5670>
aims to solve this.

In addition to the above, deno ops (i.e native extension code) are
supposed to be able to register state with the runtime, but I wasn't
able to get it to work for keeping track of which worker was which
(allowing us to propagate payloads to the appropriate output).

bridge/svix-bridge/src/config/mod.rs

@@ -3,10 +3,13 @@ use std::borrow::Cow;
 use std::collections::HashMap;
 use std::io::{Error, ErrorKind};
 use std::net::SocketAddr;
+use std::path::PathBuf;
 use svix_bridge_plugin_queue::config::{
     into_receiver_output, QueueConsumerConfig, ReceiverOutputOpts as QueueOutOpts,
 };
-use svix_bridge_types::{ReceiverInputOpts, ReceiverOutput, SenderInput, TransformationConfig};
+use svix_bridge_types::{
+    ReceiverInputOpts, ReceiverOutput, SenderInput, SenderOutputOpts, TransformationConfig,
+};
 use tracing::Level;
 
 #[derive(Deserialize)]
@@ -104,6 +107,7 @@ pub enum SenderConfig {
         feature = "sqs"
     ))]
     QueueConsumer(QueueConsumerConfig),
+    JsModule(JsModuleSenderConfig),
 }
 
 impl TryFrom<SenderConfig> for Box<dyn SenderInput> {
@@ -117,6 +121,7 @@ impl TryFrom<SenderConfig> for Box<dyn SenderInput> {
                 feature = "sqs"
             ))]
             SenderConfig::QueueConsumer(backend) => backend.into_sender_input(),
+            SenderConfig::JsModule(inner) => inner.into_sender_input(),
         }
     }
 }
@@ -154,5 +159,29 @@ impl ReceiverConfig {
     }
 }
 
+#[derive(Deserialize)]
+pub struct JsModuleSenderConfig {
+    pub name: String,
+    pub input: JsModuleSenderInputOpts,
+    #[serde(default)]
+    pub transformation: Option<TransformationConfig>,
+    pub output: SenderOutputOpts,
+}
+
+impl JsModuleSenderConfig {
+    fn into_sender_input(self) -> Result<Box<dyn SenderInput>, &'static str> {
+        // FIXME: need to make it so we can use latest deno for transformations before we can
+        //   connect the new module code.
+        todo!()
+    }
+}
+
+#[derive(Deserialize)]
+#[serde(tag = "type", rename_all = "lowercase")]
+pub enum JsModuleSenderInputOpts {
+    #[serde(rename = "js-module")]
+    JsModule { module_path: PathBuf },
+}
+
 #[cfg(test)]
 mod tests;

bridge/svix-bridge/src/config/tests.rs

@@ -1,6 +1,9 @@
 use super::Config;
-use crate::config::{LogFormat, LogLevel, SenderConfig};
+use crate::config::{
+    JsModuleSenderConfig, JsModuleSenderInputOpts, LogFormat, LogLevel, SenderConfig,
+};
 use std::collections::HashMap;
+use std::path::PathBuf;
 use svix_bridge_plugin_queue::config::{QueueConsumerConfig, RabbitMqInputOpts, SenderInputOpts};
 use svix_bridge_types::{SenderOutputOpts, SvixSenderOutputOpts};
 
@@ -485,3 +488,43 @@ fn test_variable_substitution_repeated_lookups() {
         panic!("sender did not match expected pattern");
     }
 }
+
+#[test]
+fn test_js_module_sender_input_ok() {
+    let src = r#"
+    senders:
+        - name: "js-module-example"
+          input:
+            type: "js-module"
+            # FIXME: custom module loader needed to use yaml keys for src
+            module_path: "./my-module.js"
+          transformation: |
+            function handler(input) {
+              return {
+                appId: "xxxxx",
+                message: {
+                  eventType: "lipsum.word-lengths.changed",
+                  payload: { lengths: input.lengths }
+                }
+              };
+            }
+          output:
+            type: "svix"
+            token: "x"
+    "#;
+    let cfg = Config::from_src(src, Some(HashMap::new()).as_ref()).unwrap();
+
+    if let SenderConfig::JsModule(JsModuleSenderConfig {
+        input: JsModuleSenderInputOpts::JsModule { module_path, .. },
+        transformation,
+        output: SenderOutputOpts::Svix(SvixSenderOutputOpts { token, .. }),
+        ..
+    }) = &cfg.senders[0]
+    {
+        assert_eq!(module_path, &PathBuf::from("./my-module.js"));
+        assert!(transformation.is_some());
+        assert_eq!(token, "x");
+    } else {
+        panic!("sender did not match expected pattern");
+    }
+}