bridge: Add Kafka as an input

… that is, support converting Kafka messages into Svix API calls.

bridge/Cargo.toml

@@ -9,6 +9,7 @@ members = [
     "svix-bridge-types",
     "svix-bridge",
     "svix-bridge-plugin-queue",
+    "svix-bridge-plugin-kafka",
 ]
 
 [profile.dev.package]

bridge/svix-bridge-plugin-kafka/Cargo.toml

@@ -0,0 +1,18 @@
+[package]
+name = "svix-bridge-plugin-kafka"
+version = "0.1.0"
+edition = "2021"
+
+[dependencies]
+rdkafka = { version = "0.36.0", features = ["cmake-build", "ssl", "tracing"] }
+serde = { version = "1.0", features = ["derive"] }
+serde_json = "1.0.117"
+svix-bridge-types = { path = "../svix-bridge-types" }
+thiserror = "1.0.61"
+tokio = { version = "1.28.1", features = ["time"] }
+tracing = "0.1.40"
+
+[dev-dependencies]
+ctor = "0.2.8"
+tracing-subscriber = { version = "0.3.18", features = ["env-filter"] }
+wiremock = "0.5.18"

bridge/svix-bridge-plugin-kafka/LICENSE

@@ -0,0 +1,20 @@
+Copyright (c) 2024 Svix Inc.
+
+Permission is hereby granted, free of charge, to any person obtaining
+a copy of this software and associated documentation files (the
+"Software"), to deal in the Software without restriction, including
+without limitation the rights to use, copy, modify, merge, publish,
+distribute, sublicense, and/or sell copies of the Software, and to
+permit persons to whom the Software is furnished to do so, subject to
+the following conditions:
+
+The above copyright notice and this permission notice shall be
+included in all copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
+EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
+MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
+NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE
+LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION
+OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
+WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.

bridge/svix-bridge-plugin-kafka/src/config.rs

@@ -0,0 +1,97 @@
+use rdkafka::{consumer::StreamConsumer, error::KafkaResult, ClientConfig};
+use serde::Deserialize;
+use svix_bridge_types::{SenderInput, SenderOutputOpts, TransformationConfig};
+
+use crate::{input::KafkaConsumer, Result};
+
+#[derive(Clone, Deserialize)]
+pub struct KafkaInputOpts {
+    /// Comma-separated list of addresses.
+    ///
+    /// Example: `localhost:9094`
+    #[serde(rename = "kafka_bootstrap_brokers")]
+    pub bootstrap_brokers: String,
+
+    /// The consumer group ID, used to track the stream offset between restarts
+    /// (due to host maintenance, upgrades, crashes, etc.).
+    #[serde(rename = "kafka_group_id")]
+    pub group_id: String,
+
+    /// The topic to listen to.
+    #[serde(rename = "kafka_topic")]
+    pub topic: String,
+
+    /// The value for 'security.protocol' in the kafka config.
+    #[serde(flatten)]
+    pub security_protocol: KafkaSecurityProtocol,
+
+    /// The 'debug' config value for rdkafka - enables more verbose logging
+    /// for the selected 'contexts'
+    #[serde(rename = "kafka_debug_contexts")]
+    pub debug_contexts: Option<String>,
+}
+
+impl KafkaInputOpts {
+    pub(crate) fn create_consumer(self) -> KafkaResult<StreamConsumer> {
+        let mut config = ClientConfig::new();
+        config
+            .set("group.id", self.group_id)
+            .set("bootstrap.servers", self.bootstrap_brokers)
+            // messages are committed manually after webhook delivery was successful.
+            .set("enable.auto.commit", "false");
+
+        match self.security_protocol {
+            KafkaSecurityProtocol::Plaintext => {
+                config.set("security.protocol", "plaintext");
+            }
+            KafkaSecurityProtocol::Ssl => {
+                config.set("security.protocol", "ssl");
+            }
+            KafkaSecurityProtocol::SaslSsl {
+                sasl_username,
+                sasl_password,
+            } => {
+                config
+                    .set("security.protocol", "sasl_ssl")
+                    .set("sasl.mechanisms", "SCRAM-SHA-512")
+                    .set("sasl.username", sasl_username)
+                    .set("sasl.password", sasl_password);
+            }
+        }
+
+        if let Some(debug_contexts) = self.debug_contexts {
+            if !debug_contexts.is_empty() {
+                config.set("debug", debug_contexts);
+            }
+        }
+
+        config.create()
+    }
+}
+
+#[derive(Clone, Debug, Deserialize)]
+#[serde(tag = "kafka_security_protocol", rename_all = "snake_case")]
+pub enum KafkaSecurityProtocol {
+    Plaintext,
+    Ssl,
+    SaslSsl {
+        #[serde(rename = "kafka_sasl_username")]
+        sasl_username: String,
+        #[serde(rename = "kafka_sasl_password")]
+        sasl_password: String,
+    },
+}
+
+pub fn into_sender_input(
+    name: String,
+    opts: KafkaInputOpts,
+    transformation: Option<TransformationConfig>,
+    output: SenderOutputOpts,
+) -> Result<Box<dyn SenderInput>> {
+    Ok(Box::new(KafkaConsumer::new(
+        name,
+        opts,
+        transformation,
+        output,
+    )?))
+}

bridge/svix-bridge-plugin-kafka/src/error.rs

@@ -0,0 +1,35 @@
+use std::str;
+
+use rdkafka::error::KafkaError;
+use svix_bridge_types::svix::error::Error as SvixClientError;
+
+#[derive(Debug, thiserror::Error)]
+pub enum Error {
+    #[error("kafka error")]
+    Kafka(#[from] KafkaError),
+
+    #[error("svix client error")]
+    SvixClient(#[from] SvixClientError),
+
+    #[error("JSON deserialization failed")]
+    Deserialization(#[source] serde_json::Error),
+
+    #[error("non-UTF8 payload")]
+    NonUtf8Payload(#[source] str::Utf8Error),
+
+    #[error("kafka message is missing payload")]
+    MissingPayload,
+
+    #[error("transformation error: {error}")]
+    Transformation { error: String },
+}
+
+impl Error {
+    pub(crate) fn transformation(error: impl Into<String>) -> Self {
+        Self::Transformation {
+            error: error.into(),
+        }
+    }
+}
+
+pub type Result<T, E = Error> = std::result::Result<T, E>;