feat(slack): add trogon-source-slack webhook receiver

Signed-off-by: Yordis Prieto <yordis.prieto@gmail.com>

Author: yordis

rsworkspace/Cargo.lock

@@ -6,6 +6,7 @@ pub enum ServiceName {
     AcpNatsStdio,
     AcpNatsWs,
     TrogonSourceGithub,
+    TrogonSourceSlack,
 }
 
 impl ServiceName {
@@ -14,6 +15,7 @@ impl ServiceName {
             Self::AcpNatsStdio => "acp-nats-stdio",
             Self::AcpNatsWs => "acp-nats-ws",
             Self::TrogonSourceGithub => "trogon-source-github",
+            Self::TrogonSourceSlack => "trogon-source-slack",
         }
     }
 }
@@ -32,7 +34,14 @@ mod tests {
     fn as_str_returns_expected_values() {
         assert_eq!(ServiceName::AcpNatsStdio.as_str(), "acp-nats-stdio");
         assert_eq!(ServiceName::AcpNatsWs.as_str(), "acp-nats-ws");
-        assert_eq!(ServiceName::TrogonSourceGithub.as_str(), "trogon-source-github");
+        assert_eq!(
+            ServiceName::TrogonSourceGithub.as_str(),
+            "trogon-source-github"
+        );
+        assert_eq!(
+            ServiceName::TrogonSourceSlack.as_str(),
+            "trogon-source-slack"
+        );
     }
 
     #[test]
@@ -43,5 +52,9 @@ mod tests {
             format!("{}", ServiceName::TrogonSourceGithub),
             "trogon-source-github"
         );
+        assert_eq!(
+            format!("{}", ServiceName::TrogonSourceSlack),
+            "trogon-source-slack"
+        );
     }
 }

rsworkspace/crates/acp-telemetry/src/service_name.rs

@@ -1,6 +1,7 @@
 #[cfg(not(coverage))]
 pub mod client;
 pub mod message;
+pub mod publish;
 pub mod traits;
 
 #[cfg(feature = "test-support")]
@@ -15,6 +16,7 @@ pub use message::{
     JsAck, JsAckWith, JsDispatchMessage, JsDoubleAck, JsDoubleAckWith, JsMessageRef,
     JsRequestMessage,
 };
+pub use publish::{PublishOutcome, publish_event};
 pub use traits::{
     JetStreamConsumer, JetStreamContext, JetStreamCreateConsumer, JetStreamGetStream,
     JetStreamPublisher, JsMessageOf,

rsworkspace/crates/trogon-nats/src/jetstream/mod.rs

@@ -0,0 +1,101 @@
+use std::fmt;
+use std::time::Duration;
+
+use bytes::Bytes;
+use tracing::error;
+
+use crate::jetstream::JetStreamPublisher;
+
+#[derive(Debug)]
+pub enum PublishOutcome<E: fmt::Display> {
+    Published,
+    PublishFailed(E),
+    AckFailed(E),
+    AckTimedOut(Duration),
+}
+
+impl<E: fmt::Display> PublishOutcome<E> {
+    pub fn is_ok(&self) -> bool {
+        matches!(self, PublishOutcome::Published)
+    }
+
+    pub fn log_on_error(&self, source_name: &str) {
+        match self {
+            PublishOutcome::Published => {}
+            PublishOutcome::PublishFailed(e) => {
+                error!(error = %e, source = source_name, "Failed to publish event to NATS");
+            }
+            PublishOutcome::AckFailed(e) => {
+                error!(error = %e, source = source_name, "NATS ack failed");
+            }
+            PublishOutcome::AckTimedOut(timeout) => {
+                error!(?timeout, source = source_name, "NATS ack timed out");
+            }
+        }
+    }
+}
+
+pub async fn publish_event<P: JetStreamPublisher>(
+    js: &P,
+    subject: String,
+    headers: async_nats::HeaderMap,
+    body: Bytes,
+    ack_timeout: Duration,
+) -> PublishOutcome<P::PublishError> {
+    let ack_future = match js.publish_with_headers(subject, headers, body).await {
+        Ok(f) => f,
+        Err(e) => return PublishOutcome::PublishFailed(e),
+    };
+
+    match tokio::time::timeout(ack_timeout, ack_future).await {
+        Ok(Ok(_)) => PublishOutcome::Published,
+        Ok(Err(e)) => PublishOutcome::AckFailed(e),
+        Err(_) => PublishOutcome::AckTimedOut(ack_timeout),
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[cfg(feature = "test-support")]
+    mod with_mocks {
+        use super::*;
+        use crate::jetstream::MockJetStreamPublisher;
+
+        #[tokio::test]
+        async fn publish_event_returns_published_on_success() {
+            let publisher = MockJetStreamPublisher::new();
+            let result = publish_event(
+                &publisher,
+                "test.subject".to_string(),
+                async_nats::HeaderMap::new(),
+                Bytes::from_static(b"payload"),
+                Duration::from_secs(10),
+            )
+            .await;
+            assert!(result.is_ok());
+        }
+
+        #[tokio::test]
+        async fn publish_event_returns_publish_failed_on_error() {
+            let publisher = MockJetStreamPublisher::new();
+            publisher.fail_next_js_publish();
+            let result = publish_event(
+                &publisher,
+                "test.subject".to_string(),
+                async_nats::HeaderMap::new(),
+                Bytes::from_static(b"payload"),
+                Duration::from_secs(10),
+            )
+            .await;
+            assert!(matches!(result, PublishOutcome::PublishFailed(_)));
+        }
+    }
+
+    #[test]
+    fn log_on_error_does_nothing_for_published() {
+        let outcome: PublishOutcome<String> = PublishOutcome::Published;
+        outcome.log_on_error("test");
+    }
+}

rsworkspace/crates/trogon-nats/src/jetstream/publish.rs

@@ -15,8 +15,8 @@ use axum::{
 use std::future::Future;
 use std::pin::Pin;
 use tower_http::limit::RequestBodyLimitLayer;
-use tracing::{error, info, instrument, warn};
-use trogon_nats::jetstream::{JetStreamContext, JetStreamPublisher};
+use tracing::{info, instrument, warn};
+use trogon_nats::jetstream::{JetStreamContext, JetStreamPublisher, PublishOutcome, publish_event};
 
 #[cfg(not(coverage))]
 #[derive(Debug)]
@@ -53,52 +53,13 @@ impl From<std::io::Error> for ServeError {
     }
 }
 
-enum PublishOutcome<E: fmt::Display> {
-    Published,
-    PublishFailed(E),
-    AckFailed(E),
-    AckTimedOut(Duration),
-}
-
-impl<E: fmt::Display> PublishOutcome<E> {
-    fn into_status(self) -> StatusCode {
-        match self {
-            PublishOutcome::Published => {
-                info!("Published GitHub event to NATS");
-                StatusCode::OK
-            }
-            PublishOutcome::PublishFailed(e) => {
-                error!(error = %e, "Failed to publish GitHub event to NATS");
-                StatusCode::INTERNAL_SERVER_ERROR
-            }
-            PublishOutcome::AckFailed(e) => {
-                error!(error = %e, "NATS ack failed");
-                StatusCode::INTERNAL_SERVER_ERROR
-            }
-            PublishOutcome::AckTimedOut(timeout) => {
-                error!(?timeout, "NATS ack timed out");
-                StatusCode::INTERNAL_SERVER_ERROR
-            }
-        }
-    }
-}
-
-async fn publish_event<P: JetStreamPublisher>(
-    js: &P,
-    subject: String,
-    headers: async_nats::HeaderMap,
-    body: Bytes,
-    ack_timeout: Duration,
-) -> PublishOutcome<P::PublishError> {
-    let ack_future = match js.publish_with_headers(subject, headers, body).await {
-        Ok(f) => f,
-        Err(e) => return PublishOutcome::PublishFailed(e),
-    };
-
-    match tokio::time::timeout(ack_timeout, ack_future).await {
-        Ok(Ok(_)) => PublishOutcome::Published,
-        Ok(Err(e)) => PublishOutcome::AckFailed(e),
-        Err(_) => PublishOutcome::AckTimedOut(ack_timeout),
+fn outcome_to_status<E: fmt::Display>(outcome: PublishOutcome<E>) -> StatusCode {
+    if outcome.is_ok() {
+        info!("Published GitHub event to NATS");
+        StatusCode::OK
+    } else {
+        outcome.log_on_error("github");
+        StatusCode::INTERNAL_SERVER_ERROR
     }
 }
 
@@ -238,15 +199,16 @@ async fn handle_webhook_inner<P: JetStreamPublisher>(
     nats_headers.insert(NATS_HEADER_EVENT, event.as_str());
     nats_headers.insert(NATS_HEADER_DELIVERY, delivery.as_str());
 
-    publish_event(
+    let outcome = publish_event(
         &state.js,
         subject,
         nats_headers,
         body,
         state.nats_ack_timeout,
     )
-    .await
-    .into_status()
+    .await;
+
+    outcome_to_status(outcome)
 }
 
 #[cfg(test)]

rsworkspace/crates/trogon-source-github/src/server.rs

@@ -0,0 +1,34 @@
+[package]
+name = "trogon-source-slack"
+version = "0.1.0"
+edition = "2024"
+
+[lints]
+workspace = true
+
+[[bin]]
+name = "trogon-source-slack"
+path = "src/main.rs"
+
+[dependencies]
+acp-telemetry = { workspace = true }
+async-nats = { workspace = true, features = ["jetstream"] }
+axum = { workspace = true }
+bytes = { workspace = true }
+bytesize = "2.3.1"
+hex = "0.4"
+hmac = "0.12"
+sha2 = "0.10"
+tokio = { workspace = true, features = ["full"] }
+tower-http = { workspace = true, features = ["limit"] }
+serde_json = { workspace = true }
+tracing = { workspace = true }
+trogon-nats = { workspace = true }
+trogon-std = { workspace = true }
+
+[dev-dependencies]
+tokio = { workspace = true, features = ["macros", "rt-multi-thread"] }
+tower = "0.5"
+tracing-subscriber = { workspace = true }
+trogon-nats = { workspace = true, features = ["test-support"] }
+trogon-std = { workspace = true, features = ["test-support"] }