feat: Introduce "Webhook" sink (#51)

Introduces a new sink that outputs events to a remote endpoint using HTTP calls.

Co-authored-by: Mark Stopka <mark.stopka@perlur.cloud>

Cargo.toml

@@ -30,6 +30,10 @@ serde_json = "1.0.74"
 serde_derive = "1.0.130"
 strum = "0.23"
 strum_macros = "0.23"
+minicbor = "0.12.1"
+
+# feature: webhook
+reqwest = { version = "0.11", optional = true, features = ["blocking", "json"] }
 
 # feature: tuisink
 tui = { version = "0.16", optional = true, default-features = false, features = ["crossterm"] }
@@ -46,11 +50,11 @@ murmur3 = { version = "0.5.1", optional = true }
 
 # required for CI to complete successfully
 openssl = { version = "0.10", optional = true, features = ["vendored"] }
-minicbor = "0.12.1"
 
 [features]
 default = []
-fingerprint = ["murmur3"]
+webhook = ["reqwest"]
+tuisink = ["tui"]
 kafkasink = ["kafka", "openssl"]
 elasticsink = ["elasticsearch", "tokio"]
-tuisink = ["tui"]
+fingerprint = ["murmur3"]

README.md

@@ -68,7 +68,7 @@ If the available out-of-the-box features don't satisfiy your particular use-case
   - [ ] AWS SQS queue
   - [ ] AWS Lambda call
   - [ ] GCP PubSub
-  - [ ] webhook (http post)
+  - [x] webhook (http post)
   - [x] terminal (append-only, tail-like)
   - [ ] TUI
 - Filters

src/bin/oura/daemon.rs

@@ -11,6 +11,9 @@ use oura::sources::n2c::Config as N2CConfig;
 use oura::sources::n2n::Config as N2NConfig;
 use serde_derive::Deserialize;
 
+#[cfg(feature = "webhook")]
+use oura::sinks::webhook::Config as WebhookConfig;
+
 #[cfg(feature = "kafkasink")]
 use oura::sinks::kafka::Config as KafkaConfig;
 
@@ -68,6 +71,9 @@ impl FilterConfig for Filter {
 enum Sink {
     Terminal(TerminalConfig),
 
+    #[cfg(feature = "webhook")]
+    Webhook(WebhookConfig),
+
     #[cfg(feature = "kafkasink")]
     Kafka(KafkaConfig),
 
@@ -80,6 +86,9 @@ impl SinkConfig for Sink {
         match self {
             Sink::Terminal(c) => c.bootstrap(input),
 
+            #[cfg(feature = "webhook")]
+            Sink::Webhook(c) => c.bootstrap(input),
+
             #[cfg(feature = "kafkasink")]
             Sink::Kafka(c) => c.bootstrap(input),
 

src/filters/fingerpint.rs

@@ -132,7 +132,7 @@ fn build_fingerprint(event: &Event, seed: u32) -> Result<String, Error> {
             .with_prefix("coll")
             .append_slice(tx_id)?
             .append_to_string(index)?,
-        EventData::NativeScript => b
+        EventData::NativeScript {} => b
             .with_slot(&event.context.slot)
             .with_prefix("scpt")
             .append_optional(&event.context.tx_hash)?,

src/framework.rs

@@ -208,7 +208,7 @@ pub enum EventData {
         tx_id: String,
         index: u64,
     },
-    NativeScript,
+    NativeScript {},
     PlutusScript {
         data: String,
     },

src/sinks/mod.rs

@@ -1,5 +1,8 @@
 pub mod terminal;
 
+#[cfg(feature = "webhook")]
+pub mod webhook;
+
 #[cfg(feature = "tuisink")]
 pub mod tui;
 

src/sinks/terminal/format.rs

@@ -106,7 +106,7 @@ impl LogLine {
                 source,
                 max_width,
             },
-            EventData::NativeScript => LogLine {
+            EventData::NativeScript {} => LogLine {
                 prefix: "NATIVE",
                 color: Color::White,
                 content: "{{ ... }}".to_string(),

src/sinks/webhook/mod.rs

@@ -0,0 +1,4 @@
+mod run;
+mod setup;
+
+pub use setup::*;

src/sinks/webhook/run.rs

@@ -0,0 +1,77 @@
+use std::time::Duration;
+
+use reqwest::blocking::Client;
+use serde::Serialize;
+
+use crate::framework::{Error, Event, StageReceiver};
+
+use super::ErrorPolicy;
+
+#[derive(Serialize)]
+struct RequestBody {
+    #[serde(flatten)]
+    event: Event,
+    variant: String,
+    timestamp: Option<u64>,
+}
+
+impl From<Event> for RequestBody {
+    fn from(event: Event) -> Self {
+        let timestamp = event.context.timestamp.map(|x| x * 1000);
+        let variant = event.data.to_string();
+
+        RequestBody {
+            event,
+            timestamp,
+            variant,
+        }
+    }
+}
+
+fn execute_fallible_request(
+    client: &Client,
+    url: &str,
+    body: &RequestBody,
+    policy: &ErrorPolicy,
+    retry_quota: usize,
+    backoff_delay: Duration,
+) -> Result<(), Error> {
+    let request = client.post(url).json(body).build()?;
+
+    let result = client
+        .execute(request)
+        .and_then(|res| res.error_for_status());
+
+    match (result, policy, retry_quota) {
+        (Ok(_), _, _) => {
+            log::info!("successful http request to webhook");
+            Ok(())
+        }
+        (Err(x), ErrorPolicy::Exit, 0) => Err(x.into()),
+        (Err(x), ErrorPolicy::Continue, 0) => {
+            log::warn!("failed to send webhook request: {:?}", x);
+            Ok(())
+        }
+        (Err(x), _, quota) => {
+            log::warn!("failed attempt to execute webhook request: {:?}", x);
+            std::thread::sleep(backoff_delay);
+            execute_fallible_request(client, url, body, policy, quota - 1, backoff_delay)
+        }
+    }
+}
+
+pub(crate) fn request_loop(
+    input: StageReceiver,
+    client: &Client,
+    url: &str,
+    error_policy: &ErrorPolicy,
+    max_retries: usize,
+    backoff_delay: Duration,
+) -> Result<(), Error> {
+    loop {
+        let event = input.recv().unwrap();
+        let body = RequestBody::from(event);
+
+        execute_fallible_request(client, url, &body, error_policy, max_retries, backoff_delay)?;
+    }
+}

src/sinks/webhook/setup.rs

@@ -0,0 +1,91 @@
+use std::{collections::HashMap, time::Duration};
+
+use reqwest::header::{self, HeaderMap, HeaderName, HeaderValue};
+use serde_derive::Deserialize;
+
+use crate::framework::{BootstrapResult, Error, SinkConfig, StageReceiver};
+
+use super::run::request_loop;
+
+static APP_USER_AGENT: &str = concat!(env!("CARGO_PKG_NAME"), "/", env!("CARGO_PKG_VERSION"));
+
+#[derive(Debug, Deserialize, Clone)]
+pub enum ErrorPolicy {
+    Continue,
+    Exit,
+}
+
+#[derive(Default, Debug, Deserialize)]
+pub struct Config {
+    url: String,
+    authorization: Option<String>,
+    headers: Option<HashMap<String, String>>,
+    timeout: Option<u64>,
+    error_policy: Option<ErrorPolicy>,
+    max_retries: Option<usize>,
+    backoff_delay: Option<u64>,
+}
+
+fn build_headers_map(config: &Config) -> Result<HeaderMap, Error> {
+    let mut headers = HeaderMap::new();
+
+    headers.insert(
+        header::CONTENT_TYPE,
+        HeaderValue::try_from("application/json")?,
+    );
+
+    if let Some(auth_value) = &config.authorization {
+        let auth_value = HeaderValue::try_from(auth_value)?;
+        headers.insert(header::AUTHORIZATION, auth_value);
+    }
+
+    if let Some(custom) = &config.headers {
+        for (name, value) in custom.iter() {
+            let name = HeaderName::try_from(name)?;
+            let value = HeaderValue::try_from(value)?;
+            headers.insert(name, value);
+        }
+    }
+
+    Ok(headers)
+}
+
+const DEFAULT_MAX_RETRIES: usize = 20;
+const DEFAULT_BACKOFF_DELAY: u64 = 5_000;
+
+impl SinkConfig for Config {
+    fn bootstrap(&self, input: StageReceiver) -> BootstrapResult {
+        let client = reqwest::blocking::ClientBuilder::new()
+            .user_agent(APP_USER_AGENT)
+            .default_headers(build_headers_map(self)?)
+            .timeout(Duration::from_millis(self.timeout.unwrap_or(30000)))
+            .build()?;
+
+        let url = self.url.clone();
+
+        let error_policy = self
+            .error_policy
+            .as_ref()
+            .cloned()
+            .unwrap_or(ErrorPolicy::Exit);
+
+        let max_retries = self.max_retries.unwrap_or(DEFAULT_MAX_RETRIES);
+
+        let backoff_delay =
+            Duration::from_millis(self.backoff_delay.unwrap_or(DEFAULT_BACKOFF_DELAY));
+
+        let handle = std::thread::spawn(move || {
+            request_loop(
+                input,
+                &client,
+                &url,
+                &error_policy,
+                max_retries,
+                backoff_delay,
+            )
+            .expect("request loop failed")
+        });
+
+        Ok(handle)
+    }
+}