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.

Currently this diff contains only support code for config handling.

Some implementation details were worked out separatly as a
[POC in a separate repo.](https://github.com/svix-onelson/poller-input-poc/)

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");
+    }
+}