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) <noreply@anthropic.com>
This commit is contained in:
Bryan Helmkamp 2026-03-05 09:16:22 -05:00
parent 8a1d1a3ff4
commit fb76c0ec3e
14 changed files with 2057 additions and 1 deletions

View file

@ -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=
export GITHUB_APP_PRIVATE_KEY=
export ARC_SLACK_APP_TOKEN=
export ARC_SLACK_BOT_TOKEN=

17
Cargo.lock generated
View file

@ -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"

View file

@ -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" }

View file

@ -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"] }

View file

@ -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<TestCase> {
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<WebInterviewer>,
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!");
}

View file

@ -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<Value> {
vec![text_block(&format!("~{question_text}~\n*Answer:* {answer_text}"))]
}
pub fn question_to_blocks(question_id: &str, question: &Question) -> Vec<Value> {
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<Value> = 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<Value> = 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"));
}
}

View file

@ -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<PostedMessage, SlackApiError> {
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<PostedMessage, SlackApiError> {
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<String, SlackApiError> {
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");
}
}

View file

@ -0,0 +1,106 @@
use serde::Deserialize;
#[derive(Debug, Clone, Default, Deserialize, PartialEq)]
pub struct SlackConfig {
pub default_channel: Option<String>,
}
#[derive(Debug, Clone)]
pub struct SlackCredentials {
pub bot_token: String,
pub app_token: String,
}
pub fn resolve_credentials() -> Option<SlackCredentials> {
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());
}
}

View file

@ -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<String>, 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<String, ConnectionError> {
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<WebInterviewer>,
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<WebInterviewer>,
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, &registry());
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, &registry());
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, &registry());
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, &registry());
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, &registry());
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, &reg);
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, &registry());
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);
}
}

View file

@ -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, &registry);
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, &registry);
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, &registry);
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, &registry);
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, &registry);
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, &registry);
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, &registry);
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, &registry);
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, &registry);
assert_eq!(action, DispatchAction::Ignored);
}
}

View file

@ -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<String> = 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());
}
}

View file

@ -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;

View file

@ -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<String>,
pub payload: Option<Value>,
}
#[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);
}
}

View file

@ -0,0 +1,225 @@
use std::collections::HashMap;
use std::sync::Mutex;
use serde_json::Value;
pub struct ThreadRegistry {
ts_to_question: Mutex<HashMap<String, String>>,
}
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<String> {
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());
}
}