bridge: reduce duplicate consumer code as much as possible (#926)

Adds wrapper types around the backend-specific types we get from
generic-queue.

## Motivation

Adding new backends to the consumer bridge plugin today is a bit of a
chore, encouraging parts of the ack/nack and consumer loop flow to be
duplicated. The generics exposed in the signatures of items from
`generic-queue` tend to make code reuse tricky.

## Solution


This is mostly focused on moving the decision-making bits of the flow up
to the trait's default implementation for methods like `receive` and
`process` since these were essentially duplicate from backend to
backend.

New wrapper types are used to hide/erase the distinct backend typing
from the consumer flow.

Additionally, the code related to each backend has been relegated to a
dedicated module (which should help when adding new backends since you
can copy the module entirely as a starting point).

More refactoring could be done, perhaps adding `enum_dispatch` to reduce
runtime cost and improve the layout of the code generally, but even
without this seems like a good first step.

Author: svix-onelson

webhook-bridge/svix-webhook-bridge-plugin-queue-consumer/src/gcp_pubsub/mod.rs

@@ -1,16 +1,11 @@
 use crate::config::{GCPPubSubConsumerConfig, GCPPubSubInputOpts};
 use crate::error::Error;
-use crate::PLUGIN_VERS;
-use crate::{create_svix_message, CreateMessageRequest};
-use crate::{run_inner, Consumer, PLUGIN_NAME};
-use generic_queue::gcp_pubsub::{
-    GCPPubSubConfig, GCPPubSubDelivery, GCPPubSubQueueBackend, GCPPubSubQueueConsumer,
-};
-use generic_queue::{Delivery, TaskQueueBackend, TaskQueueReceive};
-use std::time::Duration;
+use crate::ConsumerWrapper;
+use crate::{run_inner, Consumer};
+use generic_queue::gcp_pubsub::{GCPPubSubConfig, GCPPubSubQueueBackend};
+use generic_queue::TaskQueueBackend;
 use svix::api::Svix;
 use svix_webhook_bridge_types::{async_trait, JsObject, Plugin, TransformerTx};
-use tracing::instrument;
 
 pub struct GCPPubSubConsumerPlugin {
     input_options: GCPPubSubInputOpts,
@@ -34,95 +29,42 @@ impl GCPPubSubConsumerPlugin {
             transformation,
         }
     }
+}
 
-    /// Pulls N messages off the queue and feeds them to [`Self::process`].
-    #[instrument(skip_all,
-        fields(
-            otel.kind = "CONSUMER",
-            messaging.system = "gcp-pubsub",
-            messaging.operation = "receive",
-            messaging.source = &self.input_options.subscription_id,
-            svixagent_plugin.name = PLUGIN_NAME,
-            svixagent_plugin.vers = PLUGIN_VERS,
-        )
-    )]
-    async fn receive(&self, consumer: &mut GCPPubSubQueueConsumer) -> std::io::Result<()> {
-        let deliveries = consumer
-            .receive_all(1, Duration::from_millis(10))
-            .await
-            .map_err(Error::from)?;
-        tracing::trace!("received: {}", deliveries.len());
-        for delivery in deliveries {
-            self.process(delivery).await?;
-        }
-        Ok(())
+#[async_trait]
+impl Consumer for GCPPubSubConsumerPlugin {
+    fn source(&self) -> &str {
+        &self.input_options.subscription_id
     }
 
-    /// Parses the delivery as JSON and feeds it into [`create_svix_message`].
-    /// Will nack the delivery if either the JSON parse, transformation, or the request to svix fails.
-    #[instrument(skip_all, fields(messaging.operation = "process"))]
-    async fn process(&self, delivery: GCPPubSubDelivery<JsObject>) -> std::io::Result<()> {
-        let payload = match Delivery::<JsObject>::payload(&delivery) {
-            Ok(p) => p,
-            Err(e) => {
-                tracing::warn!("{e}");
-                delivery.nack().await.map_err(Error::from)?;
-                return Ok(());
-            }
-        };
+    fn system(&self) -> &str {
+        "gcp-pubsub"
+    }
 
-        let payload = if let Some(script) = &self.transformation {
-            match self.transform(script.clone(), payload).await {
-                Err(e) => {
-                    tracing::error!("nack: {e}");
-                    delivery.nack().await.map_err(Error::from)?;
-                    return Ok(());
-                }
-                Ok(x) => x,
-            }
-        } else {
-            payload
-        };
+    fn transformer_tx(&self) -> &Option<TransformerTx> {
+        &self.transformer_tx
+    }
 
-        match create_svix_message(&self.svix_client, payload).await {
-            Ok(_) => {
-                tracing::trace!("ack");
-                delivery.ack().await.map_err(Error::from)?
-            }
-            Err(e) => {
-                tracing::error!("nack: {e}");
-                delivery.nack().await.map_err(Error::from)?
-            }
-        }
-        Ok(())
+    fn transformation(&self) -> &Option<String> {
+        &self.transformation
     }
-}
 
-#[async_trait]
-impl Consumer for GCPPubSubConsumerPlugin {
-    fn transformer_tx(&self) -> Option<&TransformerTx> {
-        self.transformer_tx.as_ref()
+    fn svix_client(&self) -> &Svix {
+        &self.svix_client
     }
-    async fn consume(&self) -> std::io::Result<()> {
-        let mut consumer =
-            <GCPPubSubQueueBackend as TaskQueueBackend<CreateMessageRequest>>::consuming_half(
-                GCPPubSubConfig {
-                    subscription_id: self.input_options.subscription_id.clone(),
-                    credentials_file: self.input_options.credentials_file.clone(),
-                    // Topics are for producers so we don't care
-                    topic: String::new(),
-                },
-            )
-            .await
-            .map_err(Error::from)?;
-        tracing::debug!(
-            "gcp pubsub consuming: {}",
-            &self.input_options.subscription_id
-        );
 
-        loop {
-            self.receive(&mut consumer).await?;
-        }
+    async fn consumer(&self) -> std::io::Result<ConsumerWrapper> {
+        let consumer = <GCPPubSubQueueBackend as TaskQueueBackend<JsObject>>::consuming_half(
+            GCPPubSubConfig {
+                subscription_id: self.input_options.subscription_id.clone(),
+                credentials_file: self.input_options.credentials_file.clone(),
+                // Topics are for producers so we don't care
+                topic: String::new(),
+            },
+        )
+        .await
+        .map_err(Error::from)?;
+        Ok(ConsumerWrapper::GCPPubSub(consumer))
     }
 }
 
@@ -139,6 +81,6 @@ impl Plugin for GCPPubSubConsumerPlugin {
         self.transformer_tx = tx;
     }
     async fn run(&self) -> std::io::Result<()> {
-        run_inner(self, "gcp subsub", &self.input_options.subscription_id).await
+        run_inner(self).await
     }
 }