Bridge: decode svix events poller's "bridge config" from base64

There's a new authtoken endpoint for creating fresh events
subscriptions. In the response is a base64 encoded json structure
including the authtoken to use, the subscription id, and the app id -
everything we need to build an appropriate URL for fetching events from
the stream.

bridge/svix-bridge/Cargo.toml

@@ -6,6 +6,7 @@ publish = false
 
 [dependencies]
 anyhow = "1"
+base64 = "0.13.1"
 clap = { version = "4.2.4", features = ["env", "derive"] }
 axum = { version = "0.6", features = ["macros"] }
 enum_dispatch = "0.3"

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

@@ -9,7 +9,7 @@ use std::{
 };
 
 use anyhow::anyhow;
-use serde::Deserialize;
+use serde::{de, Deserialize, Deserializer};
 use shellexpand::LookupError;
 #[cfg(feature = "kafka")]
 use svix_bridge_plugin_kafka::{KafkaInputOpts, KafkaOutputOpts};
@@ -253,13 +253,35 @@ impl WebhookReceiverConfig {
     }
 }
 
+#[derive(Clone, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct MessageStreamBridgeConfig {
+    pub token: String,
+    pub app_id: String,
+    pub subscription_id: String,
+}
+
+fn deserialize_message_stream_bridge_config<'de, D>(
+    deserializer: D,
+) -> Result<MessageStreamBridgeConfig, D::Error>
+where
+    D: Deserializer<'de>,
+{
+    let buf = String::deserialize(deserializer)?;
+    let decoded = base64::decode(buf)
+        .map_err(|e| de::Error::custom(format!("failed to decode subscription config: {e:?}")))?;
+    serde_json::from_slice(&decoded)
+        .map_err(|e| de::Error::custom(format!("failed to decode subscription config: {e:?}")))
+}
+
 #[derive(Clone, Deserialize)]
 #[serde(tag = "type", rename_all = "kebab-case")]
 pub enum PollerInputOpts {
     SvixEvents {
-        app_id: String,
-        subscription_id: String,
-        svix_token: String,
+        /// This is the base64 encoded JSON given as `bridgeConfig` in the response from
+        /// `v1.message.events-subscription.create-token`.
+        #[serde(deserialize_with = "deserialize_message_stream_bridge_config")]
+        subscription_token: MessageStreamBridgeConfig,
         #[serde(default)]
         svix_options: Option<SvixOptions>,
     },
@@ -269,11 +291,11 @@ impl PollerInputOpts {
     pub fn svix_client(&self) -> Option<Svix> {
         match self {
             PollerInputOpts::SvixEvents {
-                svix_token,
+                subscription_token,
                 svix_options,
                 ..
             } => Some(Svix::new(
-                svix_token.clone(),
+                subscription_token.token.clone(),
                 svix_options.clone().map(Into::into),
             )),
         }

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

@@ -16,7 +16,9 @@ use tracing::instrument;
 use types::{IntegrationId, IntegrationState, InternalState, SerializableRequest, Unvalidated};
 
 use crate::{
-    config::{PollerInputOpts, PollerReceiverConfig, WebhookReceiverConfig},
+    config::{
+        MessageStreamBridgeConfig, PollerInputOpts, PollerReceiverConfig, WebhookReceiverConfig,
+    },
     webhook_receiver::types::SerializablePayload,
 };
 
@@ -256,8 +258,12 @@ async fn run_inner(poller: &SvixEventsPoller) -> ! {
     let mut sleep_time = NO_SLEEP;
 
     let PollerInputOpts::SvixEvents {
-        app_id,
-        subscription_id,
+        subscription_token:
+            MessageStreamBridgeConfig {
+                app_id,
+                subscription_id,
+                ..
+            },
         ..
     } = &poller.input_opts;