From fb76c0ec3ecd00c894aaa5c3073a7faafd0c8521 Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Thu, 5 Mar 2026 09:16:22 -0500 Subject: [PATCH] Add arc-slack crate: Slack Socket Mode integration for interviewer Implements a complete Slack integration for the interviewer system using Socket Mode (WebSocket-based, no public URL required). Supports all five question types: YesNo, Confirmation, MultipleChoice, MultiSelect, and Freeform (via thread replies with @mention). Modules: config, client, blocks, interaction, socket, dispatch, connection, threads. 72 unit tests + e2e example. Co-Authored-By: Claude Opus 4.6 (1M context) --- .env.example | 5 +- Cargo.lock | 17 ++ Cargo.toml | 2 + crates/arc-slack/Cargo.toml | 29 ++ crates/arc-slack/examples/slack_e2e.rs | 203 ++++++++++++++ crates/arc-slack/src/blocks.rs | 243 +++++++++++++++++ crates/arc-slack/src/client.rs | 250 +++++++++++++++++ crates/arc-slack/src/config.rs | 106 ++++++++ crates/arc-slack/src/connection.rs | 363 +++++++++++++++++++++++++ crates/arc-slack/src/dispatch.rs | 222 +++++++++++++++ crates/arc-slack/src/interaction.rs | 233 ++++++++++++++++ crates/arc-slack/src/lib.rs | 8 + crates/arc-slack/src/socket.rs | 152 +++++++++++ crates/arc-slack/src/threads.rs | 225 +++++++++++++++ 14 files changed, 2057 insertions(+), 1 deletion(-) create mode 100644 crates/arc-slack/Cargo.toml create mode 100644 crates/arc-slack/examples/slack_e2e.rs create mode 100644 crates/arc-slack/src/blocks.rs create mode 100644 crates/arc-slack/src/client.rs create mode 100644 crates/arc-slack/src/config.rs create mode 100644 crates/arc-slack/src/connection.rs create mode 100644 crates/arc-slack/src/dispatch.rs create mode 100644 crates/arc-slack/src/interaction.rs create mode 100644 crates/arc-slack/src/lib.rs create mode 100644 crates/arc-slack/src/socket.rs create mode 100644 crates/arc-slack/src/threads.rs diff --git a/.env.example b/.env.example index daa85d879..4d26d678b 100644 --- a/.env.example +++ b/.env.example @@ -14,4 +14,7 @@ export ARC_JWT_PUBLIC_KEY= export SESSION_SECRET= export GITHUB_APP_CLIENT_SECRET= export GITHUB_APP_WEBHOOK_SECRET= -export GITHUB_APP_PRIVATE_KEY= \ No newline at end of file +export GITHUB_APP_PRIVATE_KEY= + +export ARC_SLACK_APP_TOKEN= +export ARC_SLACK_BOT_TOKEN= \ No newline at end of file diff --git a/Cargo.lock b/Cargo.lock index ff274984a..888dede87 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -286,6 +286,23 @@ dependencies = [ "tracing", ] +[[package]] +name = "arc-slack" +version = "0.1.0" +dependencies = [ + "arc-workflows", + "futures-util", + "reqwest 0.12.28", + "rustls", + "serde", + "serde_json", + "tokio", + "tokio-tungstenite", + "toml", + "tracing", + "tracing-subscriber", +] + [[package]] name = "arc-types" version = "0.1.0" diff --git a/Cargo.toml b/Cargo.toml index df6280596..ae4775383 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -46,6 +46,8 @@ indicatif = "0.18" termimad = "0.34" toml = "0.8" jsonwebtoken = { version = "10", features = ["aws_lc_rs"] } +tokio-tungstenite = { version = "0.26", features = ["rustls-tls-webpki-roots"] } +futures-util = "0.3" daytona-sdk = { git = "https://github.com/brynary/daytona-sdk-rust", package = "daytona-sdk" } daytona-api-client = { git = "https://github.com/brynary/daytona-sdk-rust", package = "daytona-api-client" } diff --git a/crates/arc-slack/Cargo.toml b/crates/arc-slack/Cargo.toml new file mode 100644 index 000000000..64e412fe7 --- /dev/null +++ b/crates/arc-slack/Cargo.toml @@ -0,0 +1,29 @@ +[package] +name = "arc-slack" +edition.workspace = true +version.workspace = true +license.workspace = true +description = "Slack Socket Mode integration for Arc interviewer" + +[lib] +doctest = false + +[dependencies] +arc-workflows = { path = "../arc-workflows" } +futures-util.workspace = true +serde.workspace = true +serde_json.workspace = true +tokio.workspace = true +tokio-tungstenite.workspace = true +reqwest.workspace = true +tracing.workspace = true + +[[example]] +name = "slack-e2e" +path = "examples/slack_e2e.rs" + +[dev-dependencies] +tokio = { workspace = true, features = ["test-util", "macros"] } +toml.workspace = true +tracing-subscriber.workspace = true +rustls = { version = "0.23", default-features = false, features = ["std", "ring"] } diff --git a/crates/arc-slack/examples/slack_e2e.rs b/crates/arc-slack/examples/slack_e2e.rs new file mode 100644 index 000000000..79e33000f --- /dev/null +++ b/crates/arc-slack/examples/slack_e2e.rs @@ -0,0 +1,203 @@ +use std::sync::Arc; + +use arc_slack::blocks::{answered_blocks, question_to_blocks}; +use arc_slack::client::{PostedMessage, SlackClient}; +use arc_slack::connection; +use arc_slack::threads::ThreadRegistry; +use arc_workflows::interviewer::web::WebInterviewer; +use arc_workflows::interviewer::{ + Answer, AnswerValue, Interviewer, Question, QuestionOption, QuestionType, +}; + +struct TestCase { + label: &'static str, + question: Question, +} + +fn test_cases() -> Vec { + let mut mc = Question::new("Pick a language:", QuestionType::MultipleChoice); + mc.options = vec![ + QuestionOption { + key: "rs".to_string(), + label: "Rust".to_string(), + }, + QuestionOption { + key: "ts".to_string(), + label: "TypeScript".to_string(), + }, + QuestionOption { + key: "py".to_string(), + label: "Python".to_string(), + }, + ]; + + let mut ms = Question::new("Select features to enable:", QuestionType::MultiSelect); + ms.options = vec![ + QuestionOption { + key: "auth".to_string(), + label: "Authentication".to_string(), + }, + QuestionOption { + key: "billing".to_string(), + label: "Billing".to_string(), + }, + QuestionOption { + key: "notifications".to_string(), + label: "Notifications".to_string(), + }, + ]; + + vec![ + TestCase { + label: "YesNo", + question: Question::new("Do you approve this deployment?", QuestionType::YesNo), + }, + TestCase { + label: "Confirmation", + question: Question::new( + "This will delete all staging data. Continue?", + QuestionType::Confirmation, + ), + }, + TestCase { + label: "MultipleChoice", + question: mc, + }, + TestCase { + label: "MultiSelect", + question: ms, + }, + TestCase { + label: "Freeform", + question: Question::new("What is the repository URL?", QuestionType::Freeform), + }, + ] +} + +fn format_answer(answer: &Answer) -> String { + match &answer.value { + AnswerValue::Yes => "Yes".to_string(), + AnswerValue::No => "No".to_string(), + AnswerValue::Text(t) => t.clone(), + AnswerValue::Selected(k) => { + if let Some(opt) = &answer.selected_option { + format!("{} ({})", opt.label, k) + } else { + k.clone() + } + } + AnswerValue::Skipped => "Skipped".to_string(), + AnswerValue::Timeout => "Timed out".to_string(), + } +} + +async fn ask_question( + test_case: TestCase, + interviewer: &Arc, + thread_registry: &ThreadRegistry, + slack_client: &SlackClient, + channel: &str, +) { + eprintln!("\n--- {} ---", test_case.label); + + let question_text = test_case.question.text.clone(); + let is_freeform = test_case.question.question_type == QuestionType::Freeform; + let interviewer_clone = Arc::clone(interviewer); + let ask_handle = tokio::spawn(async move { + interviewer_clone.ask(test_case.question).await + }); + + tokio::time::sleep(std::time::Duration::from_millis(100)).await; + + let pending = interviewer.pending_questions(); + let pq = pending + .iter() + .find(|pq| pq.question.text == question_text) + .expect("Question should be pending"); + + let question_id = pq.id.clone(); + let blocks = question_to_blocks(&question_id, &pq.question); + + let posted: PostedMessage = slack_client + .post_message(channel, &blocks, None) + .await + .unwrap_or_else(|e| { + eprintln!("Failed to post message: {e}"); + std::process::exit(1); + }); + + // For freeform questions, register the message ts so thread replies get routed + if is_freeform { + thread_registry.register(&posted.ts, &question_id); + eprintln!("Posted. Reply in thread in Slack..."); + } else { + eprintln!("Posted. Respond in Slack..."); + } + + let answer = ask_handle.await.expect("ask task panicked"); + let answer_text = format_answer(&answer); + eprintln!("Got answer: {answer_text}"); + + // Clean up thread registration + if is_freeform { + thread_registry.remove(&posted.ts); + } + + let updated = answered_blocks(&question_text, &answer_text); + if let Err(e) = slack_client + .update_message(&posted.channel_id, &posted.ts, &updated) + .await + { + eprintln!("Failed to update message: {e}"); + } +} + +#[tokio::main] +async fn main() { + let _ = rustls::crypto::ring::default_provider().install_default(); + + tracing_subscriber::fmt() + .with_env_filter("arc_slack=debug,info") + .init(); + + let bot_token = std::env::var("ARC_SLACK_BOT_TOKEN").expect("ARC_SLACK_BOT_TOKEN required"); + let app_token = std::env::var("ARC_SLACK_APP_TOKEN").expect("ARC_SLACK_APP_TOKEN required"); + let channel = std::env::var("ARC_SLACK_CHANNEL").unwrap_or_else(|_| "#arc-test".to_string()); + + eprintln!("Connecting to Slack Socket Mode..."); + + let wss_url = connection::open_socket_url(&app_token) + .await + .expect("Failed to open socket URL"); + + let slack_client = SlackClient::new(bot_token.clone()); + let interviewer = Arc::new(WebInterviewer::new()); + let thread_registry = Arc::new(ThreadRegistry::new()); + + // Start the event loop in the background + let interviewer_for_loop = Arc::clone(&interviewer); + let thread_registry_for_loop = Arc::clone(&thread_registry); + let slack_client_for_loop = SlackClient::new(bot_token); + tokio::spawn(async move { + connection::run_event_loop( + &wss_url, + &interviewer_for_loop, + &thread_registry_for_loop, + &slack_client_for_loop, + None, + ) + .await + .ok(); + }); + + // Wait for the socket to connect + tokio::time::sleep(std::time::Duration::from_millis(500)).await; + eprintln!("Connected. Running all question types...\n"); + + let cases = test_cases(); + for case in cases { + ask_question(case, &interviewer, &thread_registry, &slack_client, &channel).await; + } + + eprintln!("\nAll question types tested!"); +} diff --git a/crates/arc-slack/src/blocks.rs b/crates/arc-slack/src/blocks.rs new file mode 100644 index 000000000..8fa440cf6 --- /dev/null +++ b/crates/arc-slack/src/blocks.rs @@ -0,0 +1,243 @@ +use arc_workflows::interviewer::{Question, QuestionType}; +use serde_json::{json, Value}; + +fn text_block(text: &str) -> Value { + json!({ + "type": "section", + "text": { + "type": "mrkdwn", + "text": text + } + }) +} + +fn button(label: &str, value: &str, action_id: &str) -> Value { + json!({ + "type": "button", + "text": { + "type": "plain_text", + "text": label + }, + "value": value, + "action_id": action_id + }) +} + +pub fn answered_blocks(question_text: &str, answer_text: &str) -> Vec { + vec![text_block(&format!("~{question_text}~\n*Answer:* {answer_text}"))] +} + +pub fn question_to_blocks(question_id: &str, question: &Question) -> Vec { + let section = text_block(&question.text); + + match question.question_type { + QuestionType::YesNo | QuestionType::Confirmation => { + let actions = json!({ + "type": "actions", + "elements": [ + button("Yes", "yes", &format!("{question_id}:yes")), + button("No", "no", &format!("{question_id}:no")), + ] + }); + vec![section, actions] + } + QuestionType::MultipleChoice => { + let elements: Vec = question + .options + .iter() + .map(|opt| button(&opt.label, &opt.key, &format!("{question_id}:{}", opt.key))) + .collect(); + let actions = json!({ + "type": "actions", + "elements": elements + }); + vec![section, actions] + } + QuestionType::MultiSelect => { + let options: Vec = question + .options + .iter() + .map(|opt| { + json!({ + "text": { "type": "plain_text", "text": opt.label }, + "value": opt.key + }) + }) + .collect(); + let checkboxes = json!({ + "type": "actions", + "block_id": format!("{question_id}:checkboxes"), + "elements": [{ + "type": "checkboxes", + "action_id": format!("{question_id}:select"), + "options": options + }] + }); + let submit = json!({ + "type": "actions", + "elements": [ + button("Submit", "submit", &format!("{question_id}:submit")), + ] + }); + vec![section, checkboxes, submit] + } + QuestionType::Freeform => { + vec![text_block(&format!( + "{}\n_Please reply in thread (mention me with your answer)._", + question.text + ))] + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use arc_workflows::interviewer::QuestionOption; + + #[test] + fn yes_no_produces_two_buttons() { + let q = Question::new("Approve this PR?", QuestionType::YesNo); + let blocks = question_to_blocks("q-1", &q); + let blocks_json: Value = serde_json::to_value(&blocks).unwrap(); + + let section = &blocks_json[0]; + assert_eq!(section["type"], "section"); + assert!(section["text"]["text"] + .as_str() + .unwrap() + .contains("Approve this PR?")); + + let actions = &blocks_json[1]; + assert_eq!(actions["type"], "actions"); + let elements = actions["elements"].as_array().unwrap(); + assert_eq!(elements.len(), 2); + assert_eq!(elements[0]["text"]["text"], "Yes"); + assert_eq!(elements[1]["text"]["text"], "No"); + } + + #[test] + fn confirmation_produces_two_buttons() { + let q = Question::new("Continue?", QuestionType::Confirmation); + let blocks = question_to_blocks("q-2", &q); + let blocks_json: Value = serde_json::to_value(&blocks).unwrap(); + + let actions = &blocks_json[1]; + let elements = actions["elements"].as_array().unwrap(); + assert_eq!(elements.len(), 2); + assert_eq!(elements[0]["text"]["text"], "Yes"); + assert_eq!(elements[1]["text"]["text"], "No"); + } + + #[test] + fn multiple_choice_produces_button_per_option() { + let mut q = Question::new("Pick a language:", QuestionType::MultipleChoice); + q.options = vec![ + QuestionOption { + key: "rs".to_string(), + label: "Rust".to_string(), + }, + QuestionOption { + key: "ts".to_string(), + label: "TypeScript".to_string(), + }, + QuestionOption { + key: "py".to_string(), + label: "Python".to_string(), + }, + ]; + let blocks = question_to_blocks("q-3", &q); + let blocks_json: Value = serde_json::to_value(&blocks).unwrap(); + + let actions = &blocks_json[1]; + let elements = actions["elements"].as_array().unwrap(); + assert_eq!(elements.len(), 3); + assert_eq!(elements[0]["text"]["text"], "Rust"); + assert_eq!(elements[0]["value"], "rs"); + assert_eq!(elements[1]["text"]["text"], "TypeScript"); + assert_eq!(elements[2]["text"]["text"], "Python"); + } + + #[test] + fn freeform_produces_section_prompting_thread_reply() { + let q = Question::new("What's the repo URL?", QuestionType::Freeform); + let blocks = question_to_blocks("q-4", &q); + let blocks_json: Value = serde_json::to_value(&blocks).unwrap(); + + assert_eq!(blocks_json.as_array().unwrap().len(), 1); + let text = blocks_json[0]["text"]["text"].as_str().unwrap(); + assert!(text.contains("What's the repo URL?")); + assert!(text.contains("reply in thread")); + assert!(text.contains("mention me")); + } + + #[test] + fn question_id_embedded_in_action_ids() { + let q = Question::new("Approve?", QuestionType::YesNo); + let blocks = question_to_blocks("q-7", &q); + let blocks_json: Value = serde_json::to_value(&blocks).unwrap(); + + let actions = &blocks_json[1]; + let elements = actions["elements"].as_array().unwrap(); + assert!(elements[0]["action_id"].as_str().unwrap().contains("q-7")); + } + + #[test] + fn answered_blocks_show_question_and_answer() { + let blocks = answered_blocks("Do you approve?", "Yes"); + let json: Value = serde_json::to_value(&blocks).unwrap(); + + assert_eq!(json.as_array().unwrap().len(), 1); + let text = json[0]["text"]["text"].as_str().unwrap(); + assert!(text.contains("Do you approve?")); + assert!(text.contains("Yes")); + } + + #[test] + fn answered_blocks_have_no_actions() { + let blocks = answered_blocks("Pick one:", "Rust"); + let json: Value = serde_json::to_value(&blocks).unwrap(); + + let has_actions = json + .as_array() + .unwrap() + .iter() + .any(|b| b["type"] == "actions"); + assert!(!has_actions); + } + + #[test] + fn multi_select_produces_checkboxes_and_submit_button() { + let mut q = Question::new("Select features:", QuestionType::MultiSelect); + q.options = vec![ + QuestionOption { + key: "a".to_string(), + label: "Auth".to_string(), + }, + QuestionOption { + key: "b".to_string(), + label: "Billing".to_string(), + }, + ]; + let blocks = question_to_blocks("q-5", &q); + let blocks_json: Value = serde_json::to_value(&blocks).unwrap(); + + // Checkboxes in their own block with a block_id + let checkbox_block = &blocks_json[1]; + assert_eq!(checkbox_block["type"], "actions"); + assert!(checkbox_block["block_id"].as_str().unwrap().contains("q-5")); + let cb_elements = checkbox_block["elements"].as_array().unwrap(); + assert_eq!(cb_elements[0]["type"], "checkboxes"); + + // Submit button in a separate actions block + let submit_block = &blocks_json[2]; + assert_eq!(submit_block["type"], "actions"); + let submit_elements = submit_block["elements"].as_array().unwrap(); + assert_eq!(submit_elements[0]["type"], "button"); + assert_eq!(submit_elements[0]["text"]["text"], "Submit"); + assert!(submit_elements[0]["action_id"] + .as_str() + .unwrap() + .contains("q-5")); + } +} diff --git a/crates/arc-slack/src/client.rs b/crates/arc-slack/src/client.rs new file mode 100644 index 000000000..ccca10d9b --- /dev/null +++ b/crates/arc-slack/src/client.rs @@ -0,0 +1,250 @@ +use reqwest::Client; +use serde_json::{json, Value}; +use tracing::debug; + +const SLACK_API_BASE: &str = "https://slack.com/api"; + +#[derive(Debug, Clone)] +pub struct PostedMessage { + pub channel_id: String, + pub ts: String, +} + +pub struct SlackClient { + pub bot_token: String, + http: Client, +} + +impl SlackClient { + pub fn new(bot_token: String) -> Self { + Self { + bot_token, + http: Client::new(), + } + } + + pub async fn post_message( + &self, + channel: &str, + blocks: &[Value], + thread_ts: Option<&str>, + ) -> Result { + let body = build_post_message_body(channel, blocks, thread_ts); + let resp = self + .http + .post(format!("{SLACK_API_BASE}/chat.postMessage")) + .bearer_auth(&self.bot_token) + .json(&body) + .send() + .await + .map_err(|e| SlackApiError::Http(e.to_string()))?; + + let json: Value = resp + .json() + .await + .map_err(|e| SlackApiError::Http(e.to_string()))?; + + let posted = parse_post_message_response(&json)?; + debug!(channel, ts = %posted.ts, "Posted Slack message"); + Ok(posted) + } + + pub async fn update_message( + &self, + channel: &str, + ts: &str, + blocks: &[Value], + ) -> Result<(), SlackApiError> { + let body = build_update_message_body(channel, ts, blocks); + let resp = self + .http + .post(format!("{SLACK_API_BASE}/chat.update")) + .bearer_auth(&self.bot_token) + .json(&body) + .send() + .await + .map_err(|e| SlackApiError::Http(e.to_string()))?; + + let json: Value = resp + .json() + .await + .map_err(|e| SlackApiError::Http(e.to_string()))?; + + if json["ok"].as_bool() != Some(true) { + let error = json["error"].as_str().unwrap_or("unknown_error"); + return Err(SlackApiError::Api(error.to_string())); + } + + Ok(()) + } +} + +#[derive(Debug)] +pub enum SlackApiError { + Http(String), + Api(String), +} + +impl std::fmt::Display for SlackApiError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + Self::Http(e) => write!(f, "Slack HTTP error: {e}"), + Self::Api(e) => write!(f, "Slack API error: {e}"), + } + } +} + +impl std::error::Error for SlackApiError {} + +pub fn parse_post_message_response(response: &Value) -> Result { + if response["ok"].as_bool() != Some(true) { + let error = response["error"].as_str().unwrap_or("unknown_error"); + return Err(SlackApiError::Api(error.to_string())); + } + let channel_id = response["channel"] + .as_str() + .ok_or_else(|| SlackApiError::Api("missing channel in response".to_string()))?; + let ts = response["ts"] + .as_str() + .ok_or_else(|| SlackApiError::Api("missing ts in response".to_string()))?; + Ok(PostedMessage { + channel_id: channel_id.to_string(), + ts: ts.to_string(), + }) +} + +pub fn parse_wss_url(response: &Value) -> Result { + if response["ok"].as_bool() != Some(true) { + let error = response["error"].as_str().unwrap_or("unknown_error"); + return Err(SlackApiError::Api(error.to_string())); + } + response["url"] + .as_str() + .map(|s| s.to_string()) + .ok_or_else(|| SlackApiError::Api("missing url in response".to_string())) +} + +fn build_post_message_body(channel: &str, blocks: &[Value], thread_ts: Option<&str>) -> Value { + let mut body = json!({ + "channel": channel, + "blocks": blocks + }); + if let Some(ts) = thread_ts { + body["thread_ts"] = json!(ts); + } + body +} + +fn build_update_message_body(channel: &str, ts: &str, blocks: &[Value]) -> Value { + json!({ + "channel": channel, + "ts": ts, + "blocks": blocks + }) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn new_client_stores_token() { + let client = SlackClient::new("xoxb-test-token".to_string()); + assert_eq!(client.bot_token, "xoxb-test-token"); + } + + #[test] + fn post_message_request_body_format() { + let body = build_post_message_body( + "#general", + &[serde_json::json!({"type": "section"})], + None, + ); + assert_eq!(body["channel"], "#general"); + assert_eq!(body["blocks"][0]["type"], "section"); + assert!(body["thread_ts"].is_null()); + } + + #[test] + fn post_message_request_body_with_thread() { + let body = build_post_message_body( + "#general", + &[serde_json::json!({"type": "section"})], + Some("1234.5678"), + ); + assert_eq!(body["thread_ts"], "1234.5678"); + } + + #[test] + fn update_message_request_body_format() { + let body = build_update_message_body( + "#general", + "1234.5678", + &[serde_json::json!({"type": "section"})], + ); + assert_eq!(body["channel"], "#general"); + assert_eq!(body["ts"], "1234.5678"); + assert_eq!(body["blocks"][0]["type"], "section"); + } + + #[test] + fn parse_post_message_response_extracts_channel_and_ts() { + let response = serde_json::json!({ + "ok": true, + "channel": "C1234567890", + "ts": "1234.5678" + }); + let posted = parse_post_message_response(&response).unwrap(); + assert_eq!(posted.channel_id, "C1234567890"); + assert_eq!(posted.ts, "1234.5678"); + } + + #[test] + fn parse_post_message_response_api_error() { + let response = serde_json::json!({ + "ok": false, + "error": "not_in_channel" + }); + let err = parse_post_message_response(&response).unwrap_err(); + assert!(err.to_string().contains("not_in_channel")); + } + + #[test] + fn parse_wss_url_success() { + let response = serde_json::json!({ + "ok": true, + "url": "wss://wss-primary.slack.com/link/?ticket=abc123" + }); + let url = parse_wss_url(&response).unwrap(); + assert!(url.starts_with("wss://")); + assert!(url.contains("ticket=abc123")); + } + + #[test] + fn parse_wss_url_api_error() { + let response = serde_json::json!({ + "ok": false, + "error": "invalid_auth" + }); + let err = parse_wss_url(&response).unwrap_err(); + assert!(err.to_string().contains("invalid_auth")); + } + + #[test] + fn parse_wss_url_missing_url() { + let response = serde_json::json!({ + "ok": true + }); + let err = parse_wss_url(&response).unwrap_err(); + assert!(err.to_string().contains("missing url")); + } + + #[test] + fn slack_api_error_display() { + let http_err = SlackApiError::Http("timeout".to_string()); + assert_eq!(http_err.to_string(), "Slack HTTP error: timeout"); + + let api_err = SlackApiError::Api("channel_not_found".to_string()); + assert_eq!(api_err.to_string(), "Slack API error: channel_not_found"); + } +} diff --git a/crates/arc-slack/src/config.rs b/crates/arc-slack/src/config.rs new file mode 100644 index 000000000..3c542e23a --- /dev/null +++ b/crates/arc-slack/src/config.rs @@ -0,0 +1,106 @@ +use serde::Deserialize; + +#[derive(Debug, Clone, Default, Deserialize, PartialEq)] +pub struct SlackConfig { + pub default_channel: Option, +} + +#[derive(Debug, Clone)] +pub struct SlackCredentials { + pub bot_token: String, + pub app_token: String, +} + +pub fn resolve_credentials() -> Option { + let bot_token = std::env::var("ARC_SLACK_BOT_TOKEN").ok()?; + let app_token = std::env::var("ARC_SLACK_APP_TOKEN").ok()?; + Some(SlackCredentials { + bot_token, + app_token, + }) +} + +pub struct SlackRuntimeConfig { + pub config: SlackConfig, + pub credentials: SlackCredentials, +} + +impl SlackRuntimeConfig { + pub fn new(config: SlackConfig, credentials: SlackCredentials) -> Self { + Self { + config, + credentials, + } + } + + pub fn is_enabled(&self) -> bool { + !self.credentials.bot_token.is_empty() && !self.credentials.app_token.is_empty() + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn parse_empty_toml_defaults() { + let config: SlackConfig = toml::from_str("").unwrap(); + assert_eq!(config.default_channel, None); + } + + #[test] + fn parse_with_channel() { + let toml_str = r##"default_channel = "#arc-reviews""##; + let config: SlackConfig = toml::from_str(toml_str).unwrap(); + assert_eq!(config.default_channel.as_deref(), Some("#arc-reviews")); + } + + #[test] + fn resolve_credentials_from_env() { + let creds = SlackCredentials { + bot_token: "xoxb-test".to_string(), + app_token: "xapp-test".to_string(), + }; + assert_eq!(creds.bot_token, "xoxb-test"); + assert_eq!(creds.app_token, "xapp-test"); + } + + #[test] + fn is_enabled_when_both_tokens_present() { + let config = SlackConfig { + default_channel: None, + }; + let creds = SlackCredentials { + bot_token: "xoxb-test".to_string(), + app_token: "xapp-test".to_string(), + }; + let runtime = SlackRuntimeConfig::new(config, creds); + assert!(runtime.is_enabled()); + } + + #[test] + fn is_not_enabled_with_empty_bot_token() { + let config = SlackConfig { + default_channel: None, + }; + let creds = SlackCredentials { + bot_token: String::new(), + app_token: "xapp-test".to_string(), + }; + let runtime = SlackRuntimeConfig::new(config, creds); + assert!(!runtime.is_enabled()); + } + + #[test] + fn is_not_enabled_with_empty_app_token() { + let config = SlackConfig { + default_channel: None, + }; + let creds = SlackCredentials { + bot_token: "xoxb-test".to_string(), + app_token: String::new(), + }; + let runtime = SlackRuntimeConfig::new(config, creds); + assert!(!runtime.is_enabled()); + } +} diff --git a/crates/arc-slack/src/connection.rs b/crates/arc-slack/src/connection.rs new file mode 100644 index 000000000..1a7c09aeb --- /dev/null +++ b/crates/arc-slack/src/connection.rs @@ -0,0 +1,363 @@ +use std::sync::Arc; + +use arc_workflows::interviewer::web::WebInterviewer; +use futures_util::{SinkExt, StreamExt}; +use tokio_tungstenite::tungstenite::Message; +use tracing::{debug, error, info, warn}; + +use crate::client::{parse_wss_url, SlackApiError, SlackClient}; +use crate::dispatch::{dispatch, DispatchAction}; +use crate::socket::{SocketAck, SocketEnvelope}; +use crate::threads::ThreadRegistry; + +#[derive(Debug)] +pub enum ConnectionError { + WebSocket(String), + Protocol(String), + Api(SlackApiError), +} + +impl std::fmt::Display for ConnectionError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + Self::WebSocket(e) => write!(f, "WebSocket error: {e}"), + Self::Protocol(e) => write!(f, "Protocol error: {e}"), + Self::Api(e) => write!(f, "API error: {e}"), + } + } +} + +impl std::error::Error for ConnectionError {} + +#[derive(Debug, PartialEq, Eq)] +pub enum ProcessOutcome { + Continue, + Reconnect, + Closed, +} + +/// Process a single raw WebSocket text message: parse, ack, dispatch. +pub fn process_message( + text: &str, + thread_registry: &ThreadRegistry, +) -> (Option, ProcessOutcome, DispatchAction) { + let envelope: SocketEnvelope = match serde_json::from_str(text) { + Ok(e) => e, + Err(_) => { + warn!("Failed to parse WebSocket message as envelope"); + return (None, ProcessOutcome::Continue, DispatchAction::Ignored); + } + }; + + let ack_json = envelope + .envelope_id + .as_deref() + .map(|id| serde_json::to_string(&SocketAck::new(id)).expect("ack serialization")); + + let action = dispatch(&envelope, thread_registry); + + let outcome = match &action { + DispatchAction::Reconnect => ProcessOutcome::Reconnect, + _ => ProcessOutcome::Continue, + }; + + (ack_json, outcome, action) +} + +/// Fetch a WebSocket URL from Slack's `apps.connections.open` endpoint. +pub async fn open_socket_url(app_token: &str) -> Result { + let http = reqwest::Client::new(); + let resp = http + .post("https://slack.com/api/apps.connections.open") + .bearer_auth(app_token) + .header("Content-Type", "application/x-www-form-urlencoded") + .send() + .await + .map_err(|e| ConnectionError::WebSocket(e.to_string()))?; + + let json: serde_json::Value = resp + .json() + .await + .map_err(|e| ConnectionError::WebSocket(e.to_string()))?; + + parse_wss_url(&json).map_err(ConnectionError::Api) +} + +/// Run the Socket Mode event loop. Connects, reads messages, acks, dispatches. +/// On disconnect, returns so the caller can reconnect. +pub async fn run_event_loop( + wss_url: &str, + interviewer: &Arc, + thread_registry: &ThreadRegistry, + _slack_client: &SlackClient, + _default_channel: Option<&str>, +) -> Result<(), ConnectionError> { + let (ws_stream, _) = tokio_tungstenite::connect_async(wss_url) + .await + .map_err(|e| ConnectionError::WebSocket(e.to_string()))?; + + let (mut write, mut read) = ws_stream.split(); + info!("Socket Mode WebSocket connected"); + + while let Some(msg) = read.next().await { + let msg = match msg { + Ok(m) => m, + Err(e) => { + error!("WebSocket read error: {e}"); + return Err(ConnectionError::WebSocket(e.to_string())); + } + }; + + let text = match msg { + Message::Text(t) => t, + Message::Close(_) => { + info!("WebSocket closed by server"); + return Ok(()); + } + Message::Ping(data) => { + let _ = write.send(Message::Pong(data)).await; + continue; + } + _ => continue, + }; + + let (ack_json, outcome, action) = process_message(&text, thread_registry); + + // Send ack immediately (Slack requires within 3 seconds) + if let Some(ack) = ack_json { + if let Err(e) = write.send(Message::Text(ack.into())).await { + error!("Failed to send ack: {e}"); + } + } + + // Handle dispatch action + match action { + DispatchAction::SubmitAnswer { + question_id, + answer, + } => { + debug!(question_id, "Submitting answer from Slack"); + let _ = interviewer.submit_answer(&question_id, answer); + } + DispatchAction::Connected => { + info!("Socket Mode handshake complete"); + } + DispatchAction::Reconnect | DispatchAction::Ignored => {} + } + + if outcome == ProcessOutcome::Reconnect { + info!("Server requested disconnect, will reconnect"); + return Ok(()); + } + } + + info!("WebSocket stream ended"); + Ok(()) +} + +/// Top-level runner: connects, runs the event loop, and reconnects on disconnect. +pub async fn run( + app_token: &str, + interviewer: Arc, + thread_registry: &ThreadRegistry, + slack_client: &SlackClient, + default_channel: Option<&str>, +) { + let mut backoff = std::time::Duration::from_secs(1); + let max_backoff = std::time::Duration::from_secs(30); + + loop { + let wss_url = match open_socket_url(app_token).await { + Ok(url) => { + backoff = std::time::Duration::from_secs(1); + url + } + Err(e) => { + error!("Failed to open Socket Mode connection: {e}"); + tokio::time::sleep(backoff).await; + backoff = (backoff * 2).min(max_backoff); + continue; + } + }; + + match run_event_loop( + &wss_url, + &interviewer, + thread_registry, + slack_client, + default_channel, + ) + .await + { + Ok(()) => { + info!("Event loop ended, reconnecting..."); + backoff = std::time::Duration::from_secs(1); + } + Err(e) => { + error!("Event loop error: {e}, reconnecting..."); + tokio::time::sleep(backoff).await; + backoff = (backoff * 2).min(max_backoff); + } + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use arc_workflows::interviewer::AnswerValue; + + fn registry() -> ThreadRegistry { + ThreadRegistry::new() + } + + #[test] + fn process_hello_message() { + let text = r#"{"type":"hello","num_connections":1}"#; + let (ack, outcome, action) = process_message(text, ®istry()); + assert!(ack.is_none()); + assert_eq!(outcome, ProcessOutcome::Continue); + assert_eq!(action, DispatchAction::Connected); + } + + #[test] + fn process_interactive_message_acks_and_dispatches() { + let text = r#"{ + "type": "interactive", + "envelope_id": "env-1", + "payload": { + "type": "block_actions", + "actions": [{ + "action_id": "q-1:yes", + "type": "button", + "value": "yes" + }] + } + }"#; + let (ack, outcome, action) = process_message(text, ®istry()); + assert!(ack.is_some()); + assert!(ack.unwrap().contains("env-1")); + assert_eq!(outcome, ProcessOutcome::Continue); + match action { + DispatchAction::SubmitAnswer { + question_id, + answer, + } => { + assert_eq!(question_id, "q-1"); + assert_eq!(answer.value, AnswerValue::Yes); + } + other => panic!("expected SubmitAnswer, got {other:?}"), + } + } + + #[test] + fn process_disconnect_signals_reconnect() { + let text = r#"{"type":"disconnect","reason":"link_disabled"}"#; + let (ack, outcome, action) = process_message(text, ®istry()); + assert!(ack.is_none()); + assert_eq!(outcome, ProcessOutcome::Reconnect); + assert_eq!(action, DispatchAction::Reconnect); + } + + #[test] + fn process_invalid_json_is_ignored() { + let text = "not valid json {{{"; + let (ack, outcome, action) = process_message(text, ®istry()); + assert!(ack.is_none()); + assert_eq!(outcome, ProcessOutcome::Continue); + assert_eq!(action, DispatchAction::Ignored); + } + + #[test] + fn process_events_api_acks_but_ignores() { + let text = r#"{ + "type": "events_api", + "envelope_id": "env-99", + "payload": { + "event": { "type": "app_mention", "text": "hi" } + } + }"#; + let (ack, outcome, action) = process_message(text, ®istry()); + assert!(ack.is_some()); + assert!(ack.unwrap().contains("env-99")); + assert_eq!(outcome, ProcessOutcome::Continue); + assert_eq!(action, DispatchAction::Ignored); + } + + #[test] + fn process_thread_reply_with_registered_question() { + let reg = registry(); + reg.register("1234.5678", "q-10"); + let text = serde_json::json!({ + "type": "events_api", + "envelope_id": "env-50", + "payload": { + "event": { + "type": "message", + "text": "my answer", + "thread_ts": "1234.5678", + "user": "U123" + } + } + }) + .to_string(); + let (ack, outcome, action) = process_message(&text, ®); + assert!(ack.is_some()); + assert_eq!(outcome, ProcessOutcome::Continue); + match action { + DispatchAction::SubmitAnswer { + question_id, + answer, + } => { + assert_eq!(question_id, "q-10"); + assert_eq!(answer.value, AnswerValue::Text("my answer".to_string())); + } + other => panic!("expected SubmitAnswer, got {other:?}"), + } + } + + #[tokio::test] + async fn submit_answer_reaches_web_interviewer() { + let interviewer = Arc::new(WebInterviewer::new()); + let i_clone = Arc::clone(&interviewer); + + let handle = tokio::spawn(async move { + use arc_workflows::interviewer::{Interviewer, Question, QuestionType}; + let q = Question::new("approve?", QuestionType::YesNo); + i_clone.ask(q).await + }); + + tokio::time::sleep(std::time::Duration::from_millis(50)).await; + + let pending = interviewer.pending_questions(); + assert_eq!(pending.len(), 1); + + let question_id = pending[0].id.clone(); + let text = serde_json::json!({ + "type": "interactive", + "envelope_id": "e1", + "payload": { + "type": "block_actions", + "actions": [{ + "action_id": format!("{question_id}:yes"), + "type": "button", + "value": "yes" + }] + } + }) + .to_string(); + let (_, _, action) = process_message(&text, ®istry()); + match action { + DispatchAction::SubmitAnswer { + question_id: qid, + answer, + } => { + assert!(interviewer.submit_answer(&qid, answer)); + } + other => panic!("expected SubmitAnswer, got {other:?}"), + } + + let answer = handle.await.unwrap(); + assert_eq!(answer.value, AnswerValue::Yes); + } +} diff --git a/crates/arc-slack/src/dispatch.rs b/crates/arc-slack/src/dispatch.rs new file mode 100644 index 000000000..fe33c1639 --- /dev/null +++ b/crates/arc-slack/src/dispatch.rs @@ -0,0 +1,222 @@ +use arc_workflows::interviewer::Answer; + +use crate::interaction; +use crate::socket::{classify_envelope, SocketEnvelope, SocketEventKind}; +use crate::threads::{self, ThreadRegistry}; + +#[derive(Debug)] +pub enum DispatchAction { + Connected, + SubmitAnswer { question_id: String, answer: Answer }, + Reconnect, + Ignored, +} + +impl PartialEq for DispatchAction { + fn eq(&self, other: &Self) -> bool { + matches!( + (self, other), + (Self::Connected, Self::Connected) + | (Self::Reconnect, Self::Reconnect) + | (Self::Ignored, Self::Ignored) + ) + } +} + +impl Eq for DispatchAction {} + +pub fn dispatch(envelope: &SocketEnvelope, thread_registry: &ThreadRegistry) -> DispatchAction { + match classify_envelope(envelope) { + SocketEventKind::Hello => DispatchAction::Connected, + SocketEventKind::Interactive => { + let Some(ref payload) = envelope.payload else { + return DispatchAction::Ignored; + }; + match interaction::parse_interaction(payload) { + Some((question_id, answer)) => DispatchAction::SubmitAnswer { + question_id, + answer, + }, + None => DispatchAction::Ignored, + } + } + SocketEventKind::EventsApi => { + let Some(ref payload) = envelope.payload else { + return DispatchAction::Ignored; + }; + let Some((thread_ts, text)) = threads::parse_thread_reply(payload) else { + return DispatchAction::Ignored; + }; + let Some(question_id) = thread_registry.resolve(&thread_ts) else { + return DispatchAction::Ignored; + }; + DispatchAction::SubmitAnswer { + question_id, + answer: Answer::text(text), + } + } + SocketEventKind::Disconnect => DispatchAction::Reconnect, + SocketEventKind::Unknown => DispatchAction::Ignored, + } +} + +#[cfg(test)] +mod tests { + use super::*; + use arc_workflows::interviewer::AnswerValue; + + #[test] + fn hello_produces_connected() { + let registry = ThreadRegistry::new(); + let envelope = SocketEnvelope { + envelope_type: "hello".to_string(), + envelope_id: None, + payload: None, + }; + let action = dispatch(&envelope, ®istry); + assert_eq!(action, DispatchAction::Connected); + } + + #[test] + fn interactive_button_produces_submit_answer() { + let registry = ThreadRegistry::new(); + let envelope = SocketEnvelope { + envelope_type: "interactive".to_string(), + envelope_id: Some("env-1".to_string()), + payload: Some(serde_json::json!({ + "type": "block_actions", + "actions": [{ + "action_id": "q-1:yes", + "type": "button", + "value": "yes" + }] + })), + }; + let action = dispatch(&envelope, ®istry); + match action { + DispatchAction::SubmitAnswer { + question_id, + answer, + } => { + assert_eq!(question_id, "q-1"); + assert_eq!(answer.value, AnswerValue::Yes); + } + other => panic!("expected SubmitAnswer, got {other:?}"), + } + } + + #[test] + fn interactive_with_unparseable_payload_produces_ignored() { + let registry = ThreadRegistry::new(); + let envelope = SocketEnvelope { + envelope_type: "interactive".to_string(), + envelope_id: Some("env-2".to_string()), + payload: Some(serde_json::json!({ + "type": "view_submission" + })), + }; + let action = dispatch(&envelope, ®istry); + assert_eq!(action, DispatchAction::Ignored); + } + + #[test] + fn interactive_with_no_payload_produces_ignored() { + let registry = ThreadRegistry::new(); + let envelope = SocketEnvelope { + envelope_type: "interactive".to_string(), + envelope_id: Some("env-3".to_string()), + payload: None, + }; + let action = dispatch(&envelope, ®istry); + assert_eq!(action, DispatchAction::Ignored); + } + + #[test] + fn disconnect_produces_reconnect() { + let registry = ThreadRegistry::new(); + let envelope = SocketEnvelope { + envelope_type: "disconnect".to_string(), + envelope_id: None, + payload: None, + }; + let action = dispatch(&envelope, ®istry); + assert_eq!(action, DispatchAction::Reconnect); + } + + #[test] + fn events_api_non_thread_produces_ignored() { + let registry = ThreadRegistry::new(); + let envelope = SocketEnvelope { + envelope_type: "events_api".to_string(), + envelope_id: Some("env-4".to_string()), + payload: Some(serde_json::json!({ + "event": { "type": "app_mention", "text": "hello" } + })), + }; + let action = dispatch(&envelope, ®istry); + assert_eq!(action, DispatchAction::Ignored); + } + + #[test] + fn events_api_thread_reply_to_registered_question() { + let registry = ThreadRegistry::new(); + registry.register("1234.5678", "q-10"); + let envelope = SocketEnvelope { + envelope_type: "events_api".to_string(), + envelope_id: Some("env-5".to_string()), + payload: Some(serde_json::json!({ + "event": { + "type": "message", + "text": "https://github.com/org/repo", + "thread_ts": "1234.5678", + "user": "U123" + } + })), + }; + let action = dispatch(&envelope, ®istry); + match action { + DispatchAction::SubmitAnswer { + question_id, + answer, + } => { + assert_eq!(question_id, "q-10"); + assert_eq!( + answer.value, + AnswerValue::Text("https://github.com/org/repo".to_string()) + ); + } + other => panic!("expected SubmitAnswer, got {other:?}"), + } + } + + #[test] + fn events_api_thread_reply_to_unknown_thread_ignored() { + let registry = ThreadRegistry::new(); + let envelope = SocketEnvelope { + envelope_type: "events_api".to_string(), + envelope_id: Some("env-6".to_string()), + payload: Some(serde_json::json!({ + "event": { + "type": "message", + "text": "some reply", + "thread_ts": "9999.0000", + "user": "U123" + } + })), + }; + let action = dispatch(&envelope, ®istry); + assert_eq!(action, DispatchAction::Ignored); + } + + #[test] + fn unknown_type_produces_ignored() { + let registry = ThreadRegistry::new(); + let envelope = SocketEnvelope { + envelope_type: "weird_type".to_string(), + envelope_id: None, + payload: None, + }; + let action = dispatch(&envelope, ®istry); + assert_eq!(action, DispatchAction::Ignored); + } +} diff --git a/crates/arc-slack/src/interaction.rs b/crates/arc-slack/src/interaction.rs new file mode 100644 index 000000000..8a8673aa1 --- /dev/null +++ b/crates/arc-slack/src/interaction.rs @@ -0,0 +1,233 @@ +use arc_workflows::interviewer::Answer; +use serde_json::Value; + +/// Parses a Slack interaction payload and returns (question_id, Answer). +/// +/// Action IDs follow the format `{question_id}:{action}` as set by `blocks::question_to_blocks`. +pub fn parse_interaction(payload: &Value) -> Option<(String, Answer)> { + if payload["type"].as_str()? != "block_actions" { + return None; + } + + let action = payload["actions"].as_array()?.first()?; + let action_id = action["action_id"].as_str()?; + let (question_id, action_key) = action_id.split_once(':')?; + + let action_type = action["type"].as_str().unwrap_or("button"); + + let answer = match action_type { + "button" => match action_key { + "yes" => Answer::yes(), + "no" => Answer::no(), + "submit" => extract_checkbox_selections(question_id, payload), + key => { + let value = action["value"].as_str().unwrap_or(key); + Answer::text(value.to_string()) + } + }, + "checkboxes" => { + // Ignore checkbox toggle events — wait for Submit button + return None; + } + "plain_text_input" => { + let value = action["value"].as_str()?; + Answer::text(value.to_string()) + } + _ => return None, + }; + + Some((question_id.to_string(), answer)) +} + +/// Extract selected checkbox values from `payload.state.values`. +/// The checkbox block has block_id `{question_id}:checkboxes` and +/// action_id `{question_id}:select`. +fn extract_checkbox_selections(question_id: &str, payload: &Value) -> Answer { + let block_id = format!("{question_id}:checkboxes"); + let action_id = format!("{question_id}:select"); + + let selected = payload["state"]["values"][&block_id][&action_id]["selected_options"] + .as_array(); + + match selected { + Some(options) if !options.is_empty() => { + let values: Vec = options + .iter() + .filter_map(|opt| opt["value"].as_str().map(String::from)) + .collect(); + Answer::text(values.join(", ")) + } + _ => Answer::skipped(), + } +} + +#[cfg(test)] +mod tests { + use super::*; + use arc_workflows::interviewer::AnswerValue; + + #[test] + fn parse_yes_button_click() { + let payload = serde_json::json!({ + "type": "block_actions", + "actions": [{ + "action_id": "q-1:yes", + "type": "button", + "value": "yes" + }] + }); + let result = parse_interaction(&payload).unwrap(); + assert_eq!(result.0, "q-1"); + assert_eq!(result.1.value, AnswerValue::Yes); + } + + #[test] + fn parse_no_button_click() { + let payload = serde_json::json!({ + "type": "block_actions", + "actions": [{ + "action_id": "q-2:no", + "type": "button", + "value": "no" + }] + }); + let result = parse_interaction(&payload).unwrap(); + assert_eq!(result.0, "q-2"); + assert_eq!(result.1.value, AnswerValue::No); + } + + #[test] + fn parse_multiple_choice_button() { + let payload = serde_json::json!({ + "type": "block_actions", + "actions": [{ + "action_id": "q-3:rs", + "type": "button", + "value": "rs" + }] + }); + let result = parse_interaction(&payload).unwrap(); + assert_eq!(result.0, "q-3"); + assert_eq!(result.1.value, AnswerValue::Text("rs".to_string())); + } + + #[test] + fn checkbox_toggle_is_ignored() { + let payload = serde_json::json!({ + "type": "block_actions", + "actions": [{ + "action_id": "q-5:select", + "type": "checkboxes", + "selected_options": [ + { "value": "a" }, + { "value": "b" } + ] + }] + }); + assert!(parse_interaction(&payload).is_none()); + } + + #[test] + fn submit_button_reads_checkbox_state() { + let payload = serde_json::json!({ + "type": "block_actions", + "actions": [{ + "action_id": "q-5:submit", + "type": "button", + "value": "submit" + }], + "state": { + "values": { + "q-5:checkboxes": { + "q-5:select": { + "type": "checkboxes", + "selected_options": [ + { "value": "auth" }, + { "value": "billing" } + ] + } + } + } + } + }); + let result = parse_interaction(&payload).unwrap(); + assert_eq!(result.0, "q-5"); + assert_eq!( + result.1.value, + AnswerValue::Text("auth, billing".to_string()) + ); + } + + #[test] + fn submit_button_with_no_checkboxes_selected() { + let payload = serde_json::json!({ + "type": "block_actions", + "actions": [{ + "action_id": "q-5:submit", + "type": "button", + "value": "submit" + }], + "state": { + "values": { + "q-5:checkboxes": { + "q-5:select": { + "type": "checkboxes", + "selected_options": [] + } + } + } + } + }); + let result = parse_interaction(&payload).unwrap(); + assert_eq!(result.0, "q-5"); + assert_eq!(result.1.value, AnswerValue::Skipped); + } + + #[test] + fn parse_plain_text_input() { + let payload = serde_json::json!({ + "type": "block_actions", + "actions": [{ + "action_id": "q-6:input", + "type": "plain_text_input", + "value": "https://github.com/org/repo" + }] + }); + let result = parse_interaction(&payload).unwrap(); + assert_eq!(result.0, "q-6"); + assert_eq!( + result.1.value, + AnswerValue::Text("https://github.com/org/repo".to_string()) + ); + } + + #[test] + fn returns_none_for_empty_actions() { + let payload = serde_json::json!({ + "type": "block_actions", + "actions": [] + }); + assert!(parse_interaction(&payload).is_none()); + } + + #[test] + fn returns_none_for_unknown_type() { + let payload = serde_json::json!({ + "type": "view_submission" + }); + assert!(parse_interaction(&payload).is_none()); + } + + #[test] + fn returns_none_for_malformed_action_id() { + let payload = serde_json::json!({ + "type": "block_actions", + "actions": [{ + "action_id": "no-colon", + "type": "button", + "value": "yes" + }] + }); + assert!(parse_interaction(&payload).is_none()); + } +} diff --git a/crates/arc-slack/src/lib.rs b/crates/arc-slack/src/lib.rs new file mode 100644 index 000000000..5fa7bb737 --- /dev/null +++ b/crates/arc-slack/src/lib.rs @@ -0,0 +1,8 @@ +pub mod blocks; +pub mod client; +pub mod config; +pub mod connection; +pub mod dispatch; +pub mod interaction; +pub mod socket; +pub mod threads; diff --git a/crates/arc-slack/src/socket.rs b/crates/arc-slack/src/socket.rs new file mode 100644 index 000000000..82253796e --- /dev/null +++ b/crates/arc-slack/src/socket.rs @@ -0,0 +1,152 @@ +use serde::{Deserialize, Serialize}; +use serde_json::Value; + +#[derive(Debug, Clone, Deserialize)] +pub struct SocketEnvelope { + #[serde(rename = "type")] + pub envelope_type: String, + pub envelope_id: Option, + pub payload: Option, +} + +#[derive(Debug, Clone, Serialize)] +pub struct SocketAck { + pub envelope_id: String, +} + +impl SocketAck { + pub fn new(envelope_id: &str) -> Self { + Self { + envelope_id: envelope_id.to_string(), + } + } +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum SocketEventKind { + Hello, + Interactive, + EventsApi, + Disconnect, + Unknown, +} + +pub fn classify_envelope(envelope: &SocketEnvelope) -> SocketEventKind { + match envelope.envelope_type.as_str() { + "hello" => SocketEventKind::Hello, + "interactive" => SocketEventKind::Interactive, + "events_api" => SocketEventKind::EventsApi, + "disconnect" => SocketEventKind::Disconnect, + _ => SocketEventKind::Unknown, + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn parse_hello_envelope() { + let json = r#"{"type":"hello","num_connections":1}"#; + let envelope: SocketEnvelope = serde_json::from_str(json).unwrap(); + assert_eq!(envelope.envelope_type, "hello"); + assert!(envelope.envelope_id.is_none()); + } + + #[test] + fn parse_events_api_envelope() { + let json = r#"{ + "type": "events_api", + "envelope_id": "abc-123", + "payload": { + "event": { + "type": "app_mention", + "text": "<@U123> run workflow", + "channel": "C456" + } + } + }"#; + let envelope: SocketEnvelope = serde_json::from_str(json).unwrap(); + assert_eq!(envelope.envelope_type, "events_api"); + assert_eq!(envelope.envelope_id.as_deref(), Some("abc-123")); + assert!(envelope.payload.is_some()); + } + + #[test] + fn parse_interactive_envelope() { + let json = r#"{ + "type": "interactive", + "envelope_id": "def-456", + "payload": { + "type": "block_actions", + "actions": [{ + "action_id": "q-1:yes", + "type": "button", + "value": "yes" + }] + } + }"#; + let envelope: SocketEnvelope = serde_json::from_str(json).unwrap(); + assert_eq!(envelope.envelope_type, "interactive"); + assert_eq!(envelope.envelope_id.as_deref(), Some("def-456")); + let payload = envelope.payload.unwrap(); + assert_eq!(payload["type"], "block_actions"); + } + + #[test] + fn build_ack_message() { + let ack = SocketAck::new("env-123"); + let json = serde_json::to_value(&ack).unwrap(); + assert_eq!(json["envelope_id"], "env-123"); + } + + #[test] + fn classify_hello() { + let envelope = SocketEnvelope { + envelope_type: "hello".to_string(), + envelope_id: None, + payload: None, + }; + assert_eq!(classify_envelope(&envelope), SocketEventKind::Hello); + } + + #[test] + fn classify_interactive() { + let envelope = SocketEnvelope { + envelope_type: "interactive".to_string(), + envelope_id: Some("e1".to_string()), + payload: Some(serde_json::json!({"type": "block_actions"})), + }; + assert_eq!(classify_envelope(&envelope), SocketEventKind::Interactive); + } + + #[test] + fn classify_events_api() { + let envelope = SocketEnvelope { + envelope_type: "events_api".to_string(), + envelope_id: Some("e2".to_string()), + payload: Some(serde_json::json!({"event": {}})), + }; + assert_eq!(classify_envelope(&envelope), SocketEventKind::EventsApi); + } + + #[test] + fn classify_disconnect() { + let envelope = SocketEnvelope { + envelope_type: "disconnect".to_string(), + envelope_id: None, + payload: None, + }; + assert_eq!(classify_envelope(&envelope), SocketEventKind::Disconnect); + } + + #[test] + fn classify_unknown() { + let envelope = SocketEnvelope { + envelope_type: "something_else".to_string(), + envelope_id: None, + payload: None, + }; + assert_eq!(classify_envelope(&envelope), SocketEventKind::Unknown); + } +} diff --git a/crates/arc-slack/src/threads.rs b/crates/arc-slack/src/threads.rs new file mode 100644 index 000000000..d66b07de8 --- /dev/null +++ b/crates/arc-slack/src/threads.rs @@ -0,0 +1,225 @@ +use std::collections::HashMap; +use std::sync::Mutex; + +use serde_json::Value; + +pub struct ThreadRegistry { + ts_to_question: Mutex>, +} + +impl ThreadRegistry { + pub fn new() -> Self { + Self { + ts_to_question: Mutex::new(HashMap::new()), + } + } + + pub fn register(&self, message_ts: &str, question_id: &str) { + self.ts_to_question + .lock() + .expect("thread registry lock poisoned") + .insert(message_ts.to_string(), question_id.to_string()); + } + + pub fn resolve(&self, thread_ts: &str) -> Option { + self.ts_to_question + .lock() + .expect("thread registry lock poisoned") + .get(thread_ts) + .cloned() + } + + pub fn remove(&self, message_ts: &str) { + self.ts_to_question + .lock() + .expect("thread registry lock poisoned") + .remove(message_ts); + } +} + +/// Parse a thread reply from an events_api payload. +/// Returns (thread_ts, reply_text) if this is a thread reply from a human user. +/// Accepts both `message` and `app_mention` event types (some workspaces only +/// deliver `app_mention` to bots). +pub fn parse_thread_reply(payload: &Value) -> Option<(String, String)> { + let event = &payload["event"]; + let event_type = event["type"].as_str()?; + if event_type != "message" && event_type != "app_mention" { + return None; + } + // Ignore bot messages (our own replies) + if event["bot_id"].is_string() || event["subtype"].is_string() { + return None; + } + let thread_ts = event["thread_ts"].as_str()?; + let mut text = event["text"].as_str()?.to_string(); + // Strip the @mention prefix from app_mention events (e.g. "<@U123> my answer" → "my answer") + if event_type == "app_mention" { + if let Some(rest) = text.strip_prefix('<') { + if let Some(after_mention) = rest.split_once('>') { + text = after_mention.1.trim().to_string(); + } + } + } + if text.is_empty() { + return None; + } + Some((thread_ts.to_string(), text)) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn register_and_resolve() { + let registry = ThreadRegistry::new(); + registry.register("1234.5678", "q-1"); + assert_eq!(registry.resolve("1234.5678"), Some("q-1".to_string())); + } + + #[test] + fn resolve_unknown_returns_none() { + let registry = ThreadRegistry::new(); + assert_eq!(registry.resolve("unknown"), None); + } + + #[test] + fn remove_clears_mapping() { + let registry = ThreadRegistry::new(); + registry.register("1234.5678", "q-1"); + registry.remove("1234.5678"); + assert_eq!(registry.resolve("1234.5678"), None); + } + + #[test] + fn parse_thread_reply_valid() { + let payload = serde_json::json!({ + "event": { + "type": "message", + "text": "https://github.com/org/repo", + "thread_ts": "1234.5678", + "user": "U123" + } + }); + let result = parse_thread_reply(&payload).unwrap(); + assert_eq!(result.0, "1234.5678"); + assert_eq!(result.1, "https://github.com/org/repo"); + } + + #[test] + fn parse_thread_reply_ignores_bot_messages() { + let payload = serde_json::json!({ + "event": { + "type": "message", + "text": "bot reply", + "thread_ts": "1234.5678", + "bot_id": "B123" + } + }); + assert!(parse_thread_reply(&payload).is_none()); + } + + #[test] + fn parse_thread_reply_ignores_subtypes() { + let payload = serde_json::json!({ + "event": { + "type": "message", + "subtype": "message_changed", + "text": "edited", + "thread_ts": "1234.5678" + } + }); + assert!(parse_thread_reply(&payload).is_none()); + } + + #[test] + fn parse_thread_reply_ignores_non_thread_messages() { + let payload = serde_json::json!({ + "event": { + "type": "message", + "text": "hello", + "user": "U123" + } + }); + assert!(parse_thread_reply(&payload).is_none()); + } + + #[test] + fn parse_thread_reply_ignores_empty_text() { + let payload = serde_json::json!({ + "event": { + "type": "message", + "text": "", + "thread_ts": "1234.5678", + "user": "U123" + } + }); + assert!(parse_thread_reply(&payload).is_none()); + } + + #[test] + fn parse_thread_reply_ignores_non_message_events() { + let payload = serde_json::json!({ + "event": { + "type": "reaction_added", + "reaction": "thumbsup" + } + }); + assert!(parse_thread_reply(&payload).is_none()); + } + + #[test] + fn parse_thread_reply_app_mention_in_thread() { + let payload = serde_json::json!({ + "event": { + "type": "app_mention", + "text": "<@U0BOTID> https://github.com/org/repo", + "thread_ts": "1234.5678", + "user": "U123" + } + }); + let result = parse_thread_reply(&payload).unwrap(); + assert_eq!(result.0, "1234.5678"); + assert_eq!(result.1, "https://github.com/org/repo"); + } + + #[test] + fn parse_thread_reply_app_mention_strips_mention_prefix() { + let payload = serde_json::json!({ + "event": { + "type": "app_mention", + "text": "<@U0BOTID> my answer here", + "thread_ts": "1234.5678", + "user": "U123" + } + }); + let result = parse_thread_reply(&payload).unwrap(); + assert_eq!(result.1, "my answer here"); + } + + #[test] + fn parse_thread_reply_app_mention_ignores_non_thread() { + let payload = serde_json::json!({ + "event": { + "type": "app_mention", + "text": "<@U0BOTID> hello", + "user": "U123" + } + }); + assert!(parse_thread_reply(&payload).is_none()); + } + + #[test] + fn parse_thread_reply_app_mention_only_mention_is_empty() { + let payload = serde_json::json!({ + "event": { + "type": "app_mention", + "text": "<@U0BOTID>", + "thread_ts": "1234.5678", + "user": "U123" + } + }); + assert!(parse_thread_reply(&payload).is_none()); + } +}