Vakyartha documentation

1//! Discord surface adapter (docs/design/34-channel-onboarding.md Phase 3),
2//! built to the same contract as `telegram.rs`: poll the remote API, route
3//! every message through `POST /gateway/inbound`, deliver the reply back.
4//!
5//! **Transport choice.** Discord's real-time surface is a gateway
6//! websocket, which would mean a new websocket dependency for the whole
7//! workspace (there is none today). This bridge instead polls
8//! `GET /channels/{id}/messages?after=<last>` for an explicitly configured
9//! set of channel ids and replies with `POST /channels/{id}/messages` —
10//! the same shape as Telegram's long poll, no new dependency, and correct
11//! for the "bot watches a few channels" case this phase is for. The
12//! trade-off is that channels must be named up front (`DISCORD_CHANNEL_IDS`)
13//! rather than DMs being auto-discovered; the gateway websocket is the
14//! follow-up that removes that limit.
15//!
16//! Launched via `vak discord --server URL --token GATEWAY_TOKEN` with
17//! `DISCORD_BOT_TOKEN` in the environment (the credential store included).
18
19use serde_json::Value;
20
21use crate::gateway::{InboundChannel, InboundRequest};
22
23const API_BASE_DEFAULT: &str = "https://discord.com/api/v10";
24
25pub struct DiscordBridge {
26 /// API base, e.g. `https://discord.com/api/v10`. Overridable for
27 /// tests and proxies via `DISCORD_API_BASE`.
28 pub api_base: String,
29 pub bot_token: String,
30 /// Channel ids this bot watches. Empty is a configuration error, not
31 /// a silent no-op: a bridge that polls nothing looks identical to a
32 /// broken one.
33 pub channel_ids: Vec<String>,
34 pub gateway_url: String,
35 pub gateway_token: String,
36 /// Seconds between polls. Discord's REST rate limits are generous at
37 /// this cadence for a handful of channels.
38 pub poll_secs: u64,
39 /// See `TelegramBridge::bot_id`.
40 pub bot_id: Option<String>,
41}
42
43impl InboundChannel for DiscordBridge {
44 fn surface(&self) -> &'static str {
45 "discord"
46 }
47}
48
49/// One routable message: Discord's own snowflake ids, never a placeholder
50/// (see `InboundRequest::new` for why that distinction is enforced).
51#[derive(Debug, Clone, PartialEq, Eq)]
52pub struct DiscordMessage {
53 pub id: String,
54 pub channel_id: String,
55 pub author_id: String,
56 pub text: String,
57 pub audio_url: Option<String>,
58 pub audio_mime: Option<String>,
59}
60
61#[derive(Debug, Default)]
62struct GatewayReply {
63 chunks: Vec<String>,
64 session_id: Option<String>,
65}
66
67/// Parse a `GET /channels/{id}/messages` page into routable messages,
68/// oldest first (Discord returns newest first). Bot-authored messages are
69/// dropped: echoing our own replies back into the gateway would loop.
70pub fn parse_messages(channel_id: &str, body: &Value) -> Vec<DiscordMessage> {
71 let mut out: Vec<DiscordMessage> = body
72 .as_array()
73 .map(|items| {
74 items
75 .iter()
76 .filter(|m| m["author"]["bot"].as_bool() != Some(true))
77 .filter_map(|m| {
78 let text = m["content"].as_str().unwrap_or_default().to_string();
79 let audio = m["attachments"].as_array().and_then(|items| {
80 items.iter().find(|a| {
81 a["content_type"]
82 .as_str()
83 .is_some_and(|mime| mime.starts_with("audio/"))
84 })
85 });
86 if text.trim().is_empty() && audio.is_none() {
87 return None;
88 }
89 Some(DiscordMessage {
90 id: m["id"].as_str()?.to_string(),
91 channel_id: channel_id.to_string(),
92 author_id: m["author"]["id"].as_str()?.to_string(),
93 text,
94 audio_url: audio.and_then(|a| a["url"].as_str()).map(str::to_string),
95 audio_mime: audio
96 .and_then(|a| a["content_type"].as_str())
97 .map(str::to_string),
98 })
99 })
100 .collect()
101 })
102 .unwrap_or_default();
103 // Snowflakes are monotonic and fixed-width enough that lexical order
104 // matches time order for any ids of the same length; sort by (len, id)
105 // so a rollover to a longer snowflake still orders correctly.
106 out.sort_by(|a, b| (a.id.len(), &a.id).cmp(&(b.id.len(), &b.id)));
107 out
108}
109
110fn http() -> reqwest::Client {
111 static CLIENT: std::sync::OnceLock<reqwest::Client> = std::sync::OnceLock::new();
112 CLIENT
113 .get_or_init(|| {
114 reqwest::Client::builder()
115 .timeout(std::time::Duration::from_secs(300))
116 .build()
117 .unwrap_or_default()
118 })
119 .clone()
120}
121
122/// Backoff for consecutive poll failures: doubling, capped at 30s — the
123/// same ceiling the Telegram bridge uses.
124pub fn backoff_secs(attempt: u32) -> u64 {
125 (1u64 << attempt.min(5)).min(30)
126}
127
128impl DiscordBridge {
129 pub fn from_env(
130 gateway_url: String,
131 gateway_token: String,
132 bot_token: String,
133 bot_id: Option<String>,
134 ) -> Self {
135 let channel_ids = vak_config::get_var("DISCORD_CHANNEL_IDS")
136 .unwrap_or_default()
137 .split(',')
138 .map(|s| s.trim().to_string())
139 .filter(|s| !s.is_empty())
140 .collect();
141 DiscordBridge {
142 api_base: vak_config::get_var("DISCORD_API_BASE")
143 .unwrap_or_else(|| API_BASE_DEFAULT.to_string()),
144 bot_token,
145 channel_ids,
146 gateway_url: gateway_url.trim_end_matches('/').to_string(),
147 gateway_token,
148 poll_secs: 3,
149 bot_id,
150 }
151 }
152
153 /// One poll pass over every watched channel. `cursors` maps channel id
154 /// to the last message id already routed, so a restart never replays
155 /// and a transient failure never skips.
156 pub async fn tick(
157 &self,
158 cursors: &mut std::collections::HashMap<String, String>,
159 ) -> Result<(), String> {
160 for channel_id in &self.channel_ids {
161 let messages = self.fetch(channel_id, cursors.get(channel_id)).await?;
162 for message in messages {
163 // First sight of a channel: adopt the cursor without
164 // replaying its backlog into the agent.
165 let cold_start = !cursors.contains_key(channel_id);
166 cursors.insert(channel_id.clone(), message.id.clone());
167 if cold_start {
168 continue;
169 }
170 let reply = self.process(&message).await;
171 if let Err(e) = self.send_message(channel_id, &reply.chunks).await {
172 eprintln!("[discord] send to {channel_id} failed: {e}");
173 }
174 if message.audio_url.is_some()
175 && let Err(e) = self
176 .send_voice(
177 channel_id,
178 &reply.chunks.join("\n"),
179 reply.session_id.as_deref(),
180 )
181 .await
182 {
183 eprintln!("[discord] voice reply unavailable: {e}");
184 }
185 }
186 }
187 Ok(())
188 }
189
190 async fn send_voice(
191 &self,
192 channel_id: &str,
193 text: &str,
194 session_id: Option<&str>,
195 ) -> Result<(), String> {
196 let response = http()
197 .post(format!("{}/voice/speak", self.gateway_url))
198 .bearer_auth(&self.gateway_token)
199 .json(&serde_json::json!({"text": text, "format": "wav", "session_id": session_id}))
200 .send()
201 .await
202 .map_err(|e| e.to_string())?;
203 if !response.status().is_success() {
204 return Err(format!("voice speak returned {}", response.status()));
205 }
206 let audio = response.bytes().await.map_err(|e| e.to_string())?;
207 let part = reqwest::multipart::Part::bytes(audio.to_vec())
208 .file_name("reply.wav")
209 .mime_str("audio/wav")
210 .map_err(|e| e.to_string())?;
211 let sent = http()
212 .post(format!("{}/channels/{channel_id}/messages", self.api_base))
213 .header("Authorization", format!("Bot {}", self.bot_token))
214 .multipart(reqwest::multipart::Form::new().part("files[0]", part))
215 .send()
216 .await
217 .map_err(|e| e.to_string())?;
218 if !sent.status().is_success() {
219 return Err(format!("discord voice upload returned {}", sent.status()));
220 }
221 Ok(())
222 }
223
224 async fn fetch(
225 &self,
226 channel_id: &str,
227 after: Option<&String>,
228 ) -> Result<Vec<DiscordMessage>, String> {
229 let mut query: Vec<(&str, String)> = vec![("limit", "25".to_string())];
230 match after {
231 Some(id) => query.push(("after", id.clone())),
232 // Cold start: one message is enough to seed the cursor.
233 None => query[0].1 = "1".to_string(),
234 }
235 let resp = http()
236 .get(format!("{}/channels/{channel_id}/messages", self.api_base))
237 .header("Authorization", format!("Bot {}", self.bot_token))
238 .query(&query)
239 .send()
240 .await
241 .map_err(|e| format!("discord getMessages: {e}"))?;
242 if !resp.status().is_success() {
243 return Err(format!("discord getMessages returned {}", resp.status()));
244 }
245 let body: Value = resp
246 .json()
247 .await
248 .map_err(|e| format!("discord getMessages body: {e}"))?;
249 Ok(parse_messages(channel_id, &body))
250 }
251
252 /// One message through the gateway contract; wait for the final text.
253 async fn process(&self, message: &DiscordMessage) -> GatewayReply {
254 let text = message.text.clone();
255 let mut attachments = Vec::new();
256 if let Some(url) = &message.audio_url {
257 match http()
258 .get(url)
259 .header("Authorization", format!("Bot {}", self.bot_token))
260 .send()
261 .await
262 {
263 Ok(response) if response.status().is_success() => {
264 if let Ok(bytes) = response.bytes().await
265 && bytes.len() <= 16 * 1024 * 1024
266 {
267 use base64::Engine as _;
268 let encoded = base64::engine::general_purpose::STANDARD.encode(bytes);
269 attachments.push(serde_json::json!({
270 "data": encoded,
271 "mime": message.audio_mime.as_deref().unwrap_or("audio/ogg"),
272 "kind": "audio",
273 "filename": "voice",
274 }));
275 }
276 }
277 _ => {}
278 }
279 }
280 // 0c-03: real per-user chat/sender, never a fixed placeholder.
281 let req = match InboundRequest::new(
282 self,
283 message.channel_id.clone(),
284 message.author_id.clone(),
285 text,
286 ) {
287 Ok(req) => req
288 .with_attachments(attachments)
289 .waiting()
290 .with_bot_id(self.bot_id.clone()),
291 Err(e) => {
292 return GatewayReply {
293 chunks: vec![format!("(bridge refused to send: {e})")],
294 ..Default::default()
295 };
296 }
297 };
298 let res = http()
299 .post(format!("{}/gateway/inbound", self.gateway_url))
300 .bearer_auth(&self.gateway_token)
301 .json(&req)
302 .send()
303 .await;
304 match res {
305 Ok(r) if r.status().as_u16() == 202 => GatewayReply {
306 chunks: vec!["(queued: I'm still working on your previous message)".into()],
307 ..Default::default()
308 },
309 Ok(r) if r.status().is_success() => match r.json::<Value>().await {
310 Ok(v) => {
311 let session_id = v["session_id"].as_str().map(String::from);
312 let chunks = super::prepared_chunks(v, "discord")
313 .unwrap_or_else(|e| vec![format!("(delivery failed: {e})")]);
314 GatewayReply { chunks, session_id }
315 }
316 Err(e) => GatewayReply {
317 chunks: vec![format!("(bad gateway reply: {e})")],
318 ..Default::default()
319 },
320 },
321 Ok(r) => GatewayReply {
322 chunks: vec![format!("(gateway error: {})", r.status())],
323 ..Default::default()
324 },
325 Err(e) => GatewayReply {
326 chunks: vec![format!("(gateway unreachable: {e})")],
327 ..Default::default()
328 },
329 }
330 }
331
332 async fn send_message(&self, channel_id: &str, chunks: &[String]) -> Result<(), String> {
333 for chunk in chunks {
334 let resp = http()
335 .post(format!("{}/channels/{channel_id}/messages", self.api_base))
336 .header("Authorization", format!("Bot {}", self.bot_token))
337 .json(&serde_json::json!({ "content": chunk }))
338 .send()
339 .await
340 .map_err(|e| format!("discord createMessage: {e}"))?;
341 if !resp.status().is_success() {
342 return Err(format!("discord createMessage returned {}", resp.status()));
343 }
344 }
345 Ok(())
346 }
347
348 /// Run until the process is killed. Transient failures back off and
349 /// retry; the per-channel cursor makes every recovery gap-free.
350 pub async fn run(&self) -> Result<(), String> {
351 if self.channel_ids.is_empty() {
352 return Err(
353 "no channels to watch — set DISCORD_CHANNEL_IDS to a comma-separated \
354 list of Discord channel ids the bot can read"
355 .into(),
356 );
357 }
358 let mut cursors = std::collections::HashMap::new();
359 let mut failures: u32 = 0;
360 loop {
361 match self.tick(&mut cursors).await {
362 Ok(()) => {
363 failures = 0;
364 tokio::time::sleep(std::time::Duration::from_secs(self.poll_secs)).await;
365 }
366 Err(e) => {
367 failures += 1;
368 if failures == 1 || failures.is_multiple_of(10) {
369 eprintln!("[discord] poll failed ({failures} consecutive): {e}");
370 }
371 tokio::time::sleep(std::time::Duration::from_secs(backoff_secs(failures)))
372 .await;
373 }
374 }
375 }
376 }
377}
378
379/// Split on message boundaries the remote surface enforces, preferring a
380/// line break so a code block or list is not cut mid-token.
381pub fn chunk_text(text: &str, max: usize) -> Vec<String> {
382 if text.chars().count() <= max {
383 return vec![text.to_string()];
384 }
385 let mut out = Vec::new();
386 let mut current = String::new();
387 for line in text.split_inclusive('\n') {
388 if current.chars().count() + line.chars().count() > max && !current.is_empty() {
389 out.push(std::mem::take(&mut current));
390 }
391 // A single line longer than the cap is hard-split; nothing else can
392 // be done without dropping content.
393 if line.chars().count() > max {
394 let mut buf = String::new();
395 for ch in line.chars() {
396 buf.push(ch);
397 if buf.chars().count() == max {
398 out.push(std::mem::take(&mut buf));
399 }
400 }
401 current = buf;
402 continue;
403 }
404 current.push_str(line);
405 }
406 if !current.is_empty() {
407 out.push(current);
408 }
409 out
410}
411
412#[cfg(test)]
413#[allow(clippy::unwrap_used, clippy::expect_used)]
414mod tests {
415 use super::*;
416
417 fn bridge() -> DiscordBridge {
418 DiscordBridge {
419 api_base: "http://localhost".into(),
420 bot_token: "t".into(),
421 channel_ids: vec!["555".into()],
422 gateway_url: "http://localhost".into(),
423 gateway_token: "g".into(),
424 poll_secs: 1,
425 bot_id: None,
426 }
427 }
428
429 #[test]
430 fn surface_is_the_allowlist_key_prefix() {
431 assert_eq!(bridge().surface(), "discord");
432 }
433
434 #[test]
435 fn identity_maps_channel_to_chat_and_author_to_sender() {
436 let req = InboundRequest::new(&bridge(), "555", "42", "hi").unwrap();
437 assert_eq!(req.surface, "discord");
438 assert_eq!(req.chat, "555");
439 assert_eq!(req.sender, "42");
440 }
441
442 #[test]
443 fn empty_or_placeholder_identity_is_refused() {
444 // 0c-03: a bridge that has not wired up real identity must fail
445 // loudly, not merge every stranger into one session.
446 assert!(InboundRequest::new(&bridge(), "", "42", "hi").is_err());
447 assert!(InboundRequest::new(&bridge(), "555", " ", "hi").is_err());
448 assert!(InboundRequest::new(&bridge(), "discord", "42", "hi").is_err());
449 assert!(InboundRequest::new(&bridge(), "555", "discord", "hi").is_err());
450 }
451
452 #[test]
453 fn parses_oldest_first_and_drops_bot_and_empty_messages() {
454 let body = serde_json::json!([
455 { "id": "30", "content": "third", "author": { "id": "7" } },
456 { "id": "20", "content": "from me", "author": { "id": "1", "bot": true } },
457 { "id": "10", "content": "first", "author": { "id": "7" } },
458 { "id": "40", "content": " ", "author": { "id": "7" } },
459 ]);
460 let parsed = parse_messages("555", &body);
461 assert_eq!(parsed.len(), 2);
462 assert_eq!(parsed[0].id, "10");
463 assert_eq!(parsed[0].text, "first");
464 assert_eq!(parsed[0].channel_id, "555");
465 assert_eq!(parsed[0].author_id, "7");
466 assert_eq!(parsed[1].id, "30");
467 }
468
469 #[test]
470 fn parses_audio_attachment_without_text() {
471 let body = serde_json::json!([{"id":"50","content":"","author":{"id":"7"},"attachments":[{"content_type":"audio/ogg","url":"https://cdn.example/voice"}]}]);
472 let parsed = parse_messages("555", &body);
473 assert_eq!(parsed.len(), 1);
474 assert_eq!(parsed[0].audio_mime.as_deref(), Some("audio/ogg"));
475 assert_eq!(
476 parsed[0].audio_url.as_deref(),
477 Some("https://cdn.example/voice")
478 );
479 }
480
481 #[test]
482 fn parses_multiple_supported_audio_mimes_for_transcription() {
483 let body = serde_json::json!([
484 {"id":"51","content":"","author":{"id":"7"},"attachments":[{"content_type":"audio/webm","url":"https://cdn.example/a"}]},
485 {"id":"52","content":"","author":{"id":"7"},"attachments":[{"content_type":"audio/mpeg","url":"https://cdn.example/b"}]}
486 ]);
487 let parsed = parse_messages("555", &body);
488 assert_eq!(parsed.len(), 2);
489 assert_eq!(parsed[0].audio_mime.as_deref(), Some("audio/webm"));
490 assert_eq!(parsed[1].audio_mime.as_deref(), Some("audio/mpeg"));
491 }
492
493 #[test]
494 fn gateway_reply_session_id_is_preserved_for_playback() {
495 let value = serde_json::json!({"chunks":["ok"],"session_id":"discord-session"});
496 let reply = GatewayReply {
497 chunks: value["chunks"]
498 .as_array()
499 .unwrap()
500 .iter()
501 .filter_map(|v| v.as_str().map(String::from))
502 .collect(),
503 session_id: value["session_id"].as_str().map(String::from),
504 };
505 assert_eq!(reply.session_id.as_deref(), Some("discord-session"));
506 }
507
508 #[test]
509 fn longer_snowflakes_sort_after_shorter_ones() {
510 let body = serde_json::json!([
511 { "id": "1000000000000000000", "content": "new", "author": { "id": "7" } },
512 { "id": "999999999999999999", "content": "old", "author": { "id": "7" } },
513 ]);
514 let parsed = parse_messages("555", &body);
515 assert_eq!(parsed[0].text, "old");
516 assert_eq!(parsed[1].text, "new");
517 }
518
519 #[test]
520 fn chunking_respects_the_cap_and_keeps_every_character() {
521 let text = "abcd\n".repeat(100);
522 let chunks = chunk_text(&text, 40);
523 assert!(chunks.iter().all(|c| c.chars().count() <= 40));
524 assert_eq!(chunks.concat(), text);
525 }
526
527 #[test]
528 fn short_text_is_one_chunk() {
529 assert_eq!(chunk_text("hello", 100), vec!["hello".to_string()]);
530 }
531
532 #[test]
533 fn backoff_doubles_then_caps() {
534 assert_eq!(backoff_secs(0), 1);
535 assert_eq!(backoff_secs(3), 8);
536 assert_eq!(backoff_secs(99), 30);
537 }
538}
539

1//! Discord surface adapter (docs/design/34-channel-onboarding.md Phase 3),
2//! built to the same contract as `telegram.rs`: poll the remote API, route
3//! every message through `POST /gateway/inbound`, deliver the reply back.
4//!
5//! **Transport choice.** Discord's real-time surface is a gateway
6//! websocket, which would mean a new websocket dependency for the whole
7//! workspace (there is none today). This bridge instead polls
8//! `GET /channels/{id}/messages?after=<last>` for an explicitly configured
9//! set of channel ids and replies with `POST /channels/{id}/messages` —
10//! the same shape as Telegram's long poll, no new dependency, and correct
11//! for the "bot watches a few channels" case this phase is for. The
12//! trade-off is that channels must be named up front (`DISCORD_CHANNEL_IDS`)
13//! rather than DMs being auto-discovered; the gateway websocket is the
14//! follow-up that removes that limit.
15//!
16//! Launched via `vak discord --server URL --token GATEWAY_TOKEN` with
17//! `DISCORD_BOT_TOKEN` in the environment (the credential store included).
18
19use serde_json::Value;
20
21use crate::gateway::{InboundChannel, InboundRequest};
22
23const API_BASE_DEFAULT: &str = "https://discord.com/api/v10";
24
25pub struct DiscordBridge {
26 /// API base, e.g. `https://discord.com/api/v10`. Overridable for
27 /// tests and proxies via `DISCORD_API_BASE`.
28 pub api_base: String,
29 pub bot_token: String,
30 /// Channel ids this bot watches. Empty is a configuration error, not
31 /// a silent no-op: a bridge that polls nothing looks identical to a
32 /// broken one.
33 pub channel_ids: Vec<String>,
34 pub gateway_url: String,
35 pub gateway_token: String,
36 /// Seconds between polls. Discord's REST rate limits are generous at
37 /// this cadence for a handful of channels.
38 pub poll_secs: u64,
39 /// See `TelegramBridge::bot_id`.
40 pub bot_id: Option<String>,
41}
42
43impl InboundChannel for DiscordBridge {
44 fn surface(&self) -> &'static str {
45 "discord"
46 }
47}
48
49/// One routable message: Discord's own snowflake ids, never a placeholder
50/// (see `InboundRequest::new` for why that distinction is enforced).
51#[derive(Debug, Clone, PartialEq, Eq)]
52pub struct DiscordMessage {
53 pub id: String,
54 pub channel_id: String,
55 pub author_id: String,
56 pub text: String,
57 pub audio_url: Option<String>,
58 pub audio_mime: Option<String>,
59}
60
61#[derive(Debug, Default)]
62struct GatewayReply {
63 chunks: Vec<String>,
64 session_id: Option<String>,
65}
66
67/// Parse a `GET /channels/{id}/messages` page into routable messages,
68/// oldest first (Discord returns newest first). Bot-authored messages are
69/// dropped: echoing our own replies back into the gateway would loop.
70pub fn parse_messages(channel_id: &str, body: &Value) -> Vec<DiscordMessage> {
71 let mut out: Vec<DiscordMessage> = body
72 .as_array()
73 .map(|items| {
74 items
75 .iter()
76 .filter(|m| m["author"]["bot"].as_bool() != Some(true))
77 .filter_map(|m| {
78 let text = m["content"].as_str().unwrap_or_default().to_string();
79 let audio = m["attachments"].as_array().and_then(|items| {
80 items.iter().find(|a| {
81 a["content_type"]
82 .as_str()
83 .is_some_and(|mime| mime.starts_with("audio/"))
84 })
85 });
86 if text.trim().is_empty() && audio.is_none() {
87 return None;
88 }
89 Some(DiscordMessage {
90 id: m["id"].as_str()?.to_string(),
91 channel_id: channel_id.to_string(),
92 author_id: m["author"]["id"].as_str()?.to_string(),
93 text,
94 audio_url: audio.and_then(|a| a["url"].as_str()).map(str::to_string),
95 audio_mime: audio
96 .and_then(|a| a["content_type"].as_str())
97 .map(str::to_string),
98 })
99 })
100 .collect()
101 })
102 .unwrap_or_default();
103 // Snowflakes are monotonic and fixed-width enough that lexical order
104 // matches time order for any ids of the same length; sort by (len, id)
105 // so a rollover to a longer snowflake still orders correctly.
106 out.sort_by(|a, b| (a.id.len(), &a.id).cmp(&(b.id.len(), &b.id)));
107 out
108}
109
110fn http() -> reqwest::Client {
111 static CLIENT: std::sync::OnceLock<reqwest::Client> = std::sync::OnceLock::new();
112 CLIENT
113 .get_or_init(|| {
114 reqwest::Client::builder()
115 .timeout(std::time::Duration::from_secs(300))
116 .build()
117 .unwrap_or_default()
118 })
119 .clone()
120}
121
122/// Backoff for consecutive poll failures: doubling, capped at 30s — the
123/// same ceiling the Telegram bridge uses.
124pub fn backoff_secs(attempt: u32) -> u64 {
125 (1u64 << attempt.min(5)).min(30)
126}
127
128impl DiscordBridge {
129 pub fn from_env(
130 gateway_url: String,
131 gateway_token: String,
132 bot_token: String,
133 bot_id: Option<String>,
134 ) -> Self {
135 let channel_ids = vak_config::get_var("DISCORD_CHANNEL_IDS")
136 .unwrap_or_default()
137 .split(',')
138 .map(|s| s.trim().to_string())
139 .filter(|s| !s.is_empty())
140 .collect();
141 DiscordBridge {
142 api_base: vak_config::get_var("DISCORD_API_BASE")
143 .unwrap_or_else(|| API_BASE_DEFAULT.to_string()),
144 bot_token,
145 channel_ids,
146 gateway_url: gateway_url.trim_end_matches('/').to_string(),
147 gateway_token,
148 poll_secs: 3,
149 bot_id,
150 }
151 }
152
153 /// One poll pass over every watched channel. `cursors` maps channel id
154 /// to the last message id already routed, so a restart never replays
155 /// and a transient failure never skips.
156 pub async fn tick(
157 &self,
158 cursors: &mut std::collections::HashMap<String, String>,
159 ) -> Result<(), String> {
160 for channel_id in &self.channel_ids {
161 let messages = self.fetch(channel_id, cursors.get(channel_id)).await?;
162 for message in messages {
163 // First sight of a channel: adopt the cursor without
164 // replaying its backlog into the agent.
165 let cold_start = !cursors.contains_key(channel_id);
166 cursors.insert(channel_id.clone(), message.id.clone());
167 if cold_start {
168 continue;
169 }
170 let reply = self.process(&message).await;
171 if let Err(e) = self.send_message(channel_id, &reply.chunks).await {
172 eprintln!("[discord] send to {channel_id} failed: {e}");
173 }
174 if message.audio_url.is_some()
175 && let Err(e) = self
176 .send_voice(
177 channel_id,
178 &reply.chunks.join("\n"),
179 reply.session_id.as_deref(),
180 )
181 .await
182 {
183 eprintln!("[discord] voice reply unavailable: {e}");
184 }
185 }
186 }
187 Ok(())
188 }
189
190 async fn send_voice(
191 &self,
192 channel_id: &str,
193 text: &str,
194 session_id: Option<&str>,
195 ) -> Result<(), String> {
196 let response = http()
197 .post(format!("{}/voice/speak", self.gateway_url))
198 .bearer_auth(&self.gateway_token)
199 .json(&serde_json::json!({"text": text, "format": "wav", "session_id": session_id}))
200 .send()
201 .await
202 .map_err(|e| e.to_string())?;
203 if !response.status().is_success() {
204 return Err(format!("voice speak returned {}", response.status()));
205 }
206 let audio = response.bytes().await.map_err(|e| e.to_string())?;
207 let part = reqwest::multipart::Part::bytes(audio.to_vec())
208 .file_name("reply.wav")
209 .mime_str("audio/wav")
210 .map_err(|e| e.to_string())?;
211 let sent = http()
212 .post(format!("{}/channels/{channel_id}/messages", self.api_base))
213 .header("Authorization", format!("Bot {}", self.bot_token))
214 .multipart(reqwest::multipart::Form::new().part("files[0]", part))
215 .send()
216 .await
217 .map_err(|e| e.to_string())?;
218 if !sent.status().is_success() {
219 return Err(format!("discord voice upload returned {}", sent.status()));
220 }
221 Ok(())
222 }
223
224 async fn fetch(
225 &self,
226 channel_id: &str,
227 after: Option<&String>,
228 ) -> Result<Vec<DiscordMessage>, String> {
229 let mut query: Vec<(&str, String)> = vec![("limit", "25".to_string())];
230 match after {
231 Some(id) => query.push(("after", id.clone())),
232 // Cold start: one message is enough to seed the cursor.
233 None => query[0].1 = "1".to_string(),
234 }
235 let resp = http()
236 .get(format!("{}/channels/{channel_id}/messages", self.api_base))
237 .header("Authorization", format!("Bot {}", self.bot_token))
238 .query(&query)
239 .send()
240 .await
241 .map_err(|e| format!("discord getMessages: {e}"))?;
242 if !resp.status().is_success() {
243 return Err(format!("discord getMessages returned {}", resp.status()));
244 }
245 let body: Value = resp
246 .json()
247 .await
248 .map_err(|e| format!("discord getMessages body: {e}"))?;
249 Ok(parse_messages(channel_id, &body))
250 }
251
252 /// One message through the gateway contract; wait for the final text.
253 async fn process(&self, message: &DiscordMessage) -> GatewayReply {
254 let text = message.text.clone();
255 let mut attachments = Vec::new();
256 if let Some(url) = &message.audio_url {
257 match http()
258 .get(url)
259 .header("Authorization", format!("Bot {}", self.bot_token))
260 .send()
261 .await
262 {
263 Ok(response) if response.status().is_success() => {
264 if let Ok(bytes) = response.bytes().await
265 && bytes.len() <= 16 * 1024 * 1024
266 {
267 use base64::Engine as _;
268 let encoded = base64::engine::general_purpose::STANDARD.encode(bytes);
269 attachments.push(serde_json::json!({
270 "data": encoded,
271 "mime": message.audio_mime.as_deref().unwrap_or("audio/ogg"),
272 "kind": "audio",
273 "filename": "voice",
274 }));
275 }
276 }
277 _ => {}
278 }
279 }
280 // 0c-03: real per-user chat/sender, never a fixed placeholder.
281 let req = match InboundRequest::new(
282 self,
283 message.channel_id.clone(),
284 message.author_id.clone(),
285 text,
286 ) {
287 Ok(req) => req
288 .with_attachments(attachments)
289 .waiting()
290 .with_bot_id(self.bot_id.clone()),
291 Err(e) => {
292 return GatewayReply {
293 chunks: vec![format!("(bridge refused to send: {e})")],
294 ..Default::default()
295 };
296 }
297 };
298 let res = http()
299 .post(format!("{}/gateway/inbound", self.gateway_url))
300 .bearer_auth(&self.gateway_token)
301 .json(&req)
302 .send()
303 .await;
304 match res {
305 Ok(r) if r.status().as_u16() == 202 => GatewayReply {
306 chunks: vec!["(queued: I'm still working on your previous message)".into()],
307 ..Default::default()
308 },
309 Ok(r) if r.status().is_success() => match r.json::<Value>().await {
310 Ok(v) => {
311 let session_id = v["session_id"].as_str().map(String::from);
312 let chunks = super::prepared_chunks(v, "discord")
313 .unwrap_or_else(|e| vec![format!("(delivery failed: {e})")]);
314 GatewayReply { chunks, session_id }
315 }
316 Err(e) => GatewayReply {
317 chunks: vec![format!("(bad gateway reply: {e})")],
318 ..Default::default()
319 },
320 },
321 Ok(r) => GatewayReply {
322 chunks: vec![format!("(gateway error: {})", r.status())],
323 ..Default::default()
324 },
325 Err(e) => GatewayReply {
326 chunks: vec![format!("(gateway unreachable: {e})")],
327 ..Default::default()
328 },
329 }
330 }
331
332 async fn send_message(&self, channel_id: &str, chunks: &[String]) -> Result<(), String> {
333 for chunk in chunks {
334 let resp = http()
335 .post(format!("{}/channels/{channel_id}/messages", self.api_base))
336 .header("Authorization", format!("Bot {}", self.bot_token))
337 .json(&serde_json::json!({ "content": chunk }))
338 .send()
339 .await
340 .map_err(|e| format!("discord createMessage: {e}"))?;
341 if !resp.status().is_success() {
342 return Err(format!("discord createMessage returned {}", resp.status()));
343 }
344 }
345 Ok(())
346 }
347
348 /// Run until the process is killed. Transient failures back off and
349 /// retry; the per-channel cursor makes every recovery gap-free.
350 pub async fn run(&self) -> Result<(), String> {
351 if self.channel_ids.is_empty() {
352 return Err(
353 "no channels to watch — set DISCORD_CHANNEL_IDS to a comma-separated \
354 list of Discord channel ids the bot can read"
355 .into(),
356 );
357 }
358 let mut cursors = std::collections::HashMap::new();
359 let mut failures: u32 = 0;
360 loop {
361 match self.tick(&mut cursors).await {
362 Ok(()) => {
363 failures = 0;
364 tokio::time::sleep(std::time::Duration::from_secs(self.poll_secs)).await;
365 }
366 Err(e) => {
367 failures += 1;
368 if failures == 1 || failures.is_multiple_of(10) {
369 eprintln!("[discord] poll failed ({failures} consecutive): {e}");
370 }
371 tokio::time::sleep(std::time::Duration::from_secs(backoff_secs(failures)))
372 .await;
373 }
374 }
375 }
376 }
377}
378
379/// Split on message boundaries the remote surface enforces, preferring a
380/// line break so a code block or list is not cut mid-token.
381pub fn chunk_text(text: &str, max: usize) -> Vec<String> {
382 if text.chars().count() <= max {
383 return vec![text.to_string()];
384 }
385 let mut out = Vec::new();
386 let mut current = String::new();
387 for line in text.split_inclusive('\n') {
388 if current.chars().count() + line.chars().count() > max && !current.is_empty() {
389 out.push(std::mem::take(&mut current));
390 }
391 // A single line longer than the cap is hard-split; nothing else can
392 // be done without dropping content.
393 if line.chars().count() > max {
394 let mut buf = String::new();
395 for ch in line.chars() {
396 buf.push(ch);
397 if buf.chars().count() == max {
398 out.push(std::mem::take(&mut buf));
399 }
400 }
401 current = buf;
402 continue;
403 }
404 current.push_str(line);
405 }
406 if !current.is_empty() {
407 out.push(current);
408 }
409 out
410}
411
412#[cfg(test)]
413#[allow(clippy::unwrap_used, clippy::expect_used)]
414mod tests {
415 use super::*;
416
417 fn bridge() -> DiscordBridge {
418 DiscordBridge {
419 api_base: "http://localhost".into(),
420 bot_token: "t".into(),
421 channel_ids: vec!["555".into()],
422 gateway_url: "http://localhost".into(),
423 gateway_token: "g".into(),
424 poll_secs: 1,
425 bot_id: None,
426 }
427 }
428
429 #[test]
430 fn surface_is_the_allowlist_key_prefix() {
431 assert_eq!(bridge().surface(), "discord");
432 }
433
434 #[test]
435 fn identity_maps_channel_to_chat_and_author_to_sender() {
436 let req = InboundRequest::new(&bridge(), "555", "42", "hi").unwrap();
437 assert_eq!(req.surface, "discord");
438 assert_eq!(req.chat, "555");
439 assert_eq!(req.sender, "42");
440 }
441
442 #[test]
443 fn empty_or_placeholder_identity_is_refused() {
444 // 0c-03: a bridge that has not wired up real identity must fail
445 // loudly, not merge every stranger into one session.
446 assert!(InboundRequest::new(&bridge(), "", "42", "hi").is_err());
447 assert!(InboundRequest::new(&bridge(), "555", " ", "hi").is_err());
448 assert!(InboundRequest::new(&bridge(), "discord", "42", "hi").is_err());
449 assert!(InboundRequest::new(&bridge(), "555", "discord", "hi").is_err());
450 }
451
452 #[test]
453 fn parses_oldest_first_and_drops_bot_and_empty_messages() {
454 let body = serde_json::json!([
455 { "id": "30", "content": "third", "author": { "id": "7" } },
456 { "id": "20", "content": "from me", "author": { "id": "1", "bot": true } },
457 { "id": "10", "content": "first", "author": { "id": "7" } },
458 { "id": "40", "content": " ", "author": { "id": "7" } },
459 ]);
460 let parsed = parse_messages("555", &body);
461 assert_eq!(parsed.len(), 2);
462 assert_eq!(parsed[0].id, "10");
463 assert_eq!(parsed[0].text, "first");
464 assert_eq!(parsed[0].channel_id, "555");
465 assert_eq!(parsed[0].author_id, "7");
466 assert_eq!(parsed[1].id, "30");
467 }
468
469 #[test]
470 fn parses_audio_attachment_without_text() {
471 let body = serde_json::json!([{"id":"50","content":"","author":{"id":"7"},"attachments":[{"content_type":"audio/ogg","url":"https://cdn.example/voice"}]}]);
472 let parsed = parse_messages("555", &body);
473 assert_eq!(parsed.len(), 1);
474 assert_eq!(parsed[0].audio_mime.as_deref(), Some("audio/ogg"));
475 assert_eq!(
476 parsed[0].audio_url.as_deref(),
477 Some("https://cdn.example/voice")
478 );
479 }
480
481 #[test]
482 fn parses_multiple_supported_audio_mimes_for_transcription() {
483 let body = serde_json::json!([
484 {"id":"51","content":"","author":{"id":"7"},"attachments":[{"content_type":"audio/webm","url":"https://cdn.example/a"}]},
485 {"id":"52","content":"","author":{"id":"7"},"attachments":[{"content_type":"audio/mpeg","url":"https://cdn.example/b"}]}
486 ]);
487 let parsed = parse_messages("555", &body);
488 assert_eq!(parsed.len(), 2);
489 assert_eq!(parsed[0].audio_mime.as_deref(), Some("audio/webm"));
490 assert_eq!(parsed[1].audio_mime.as_deref(), Some("audio/mpeg"));
491 }
492
493 #[test]
494 fn gateway_reply_session_id_is_preserved_for_playback() {
495 let value = serde_json::json!({"chunks":["ok"],"session_id":"discord-session"});
496 let reply = GatewayReply {
497 chunks: value["chunks"]
498 .as_array()
499 .unwrap()
500 .iter()
501 .filter_map(|v| v.as_str().map(String::from))
502 .collect(),
503 session_id: value["session_id"].as_str().map(String::from),
504 };
505 assert_eq!(reply.session_id.as_deref(), Some("discord-session"));
506 }
507
508 #[test]
509 fn longer_snowflakes_sort_after_shorter_ones() {
510 let body = serde_json::json!([
511 { "id": "1000000000000000000", "content": "new", "author": { "id": "7" } },
512 { "id": "999999999999999999", "content": "old", "author": { "id": "7" } },
513 ]);
514 let parsed = parse_messages("555", &body);
515 assert_eq!(parsed[0].text, "old");
516 assert_eq!(parsed[1].text, "new");
517 }
518
519 #[test]
520 fn chunking_respects_the_cap_and_keeps_every_character() {
521 let text = "abcd\n".repeat(100);
522 let chunks = chunk_text(&text, 40);
523 assert!(chunks.iter().all(|c| c.chars().count() <= 40));
524 assert_eq!(chunks.concat(), text);
525 }
526
527 #[test]
528 fn short_text_is_one_chunk() {
529 assert_eq!(chunk_text("hello", 100), vec!["hello".to_string()]);
530 }
531
532 #[test]
533 fn backoff_doubles_then_caps() {
534 assert_eq!(backoff_secs(0), 1);
535 assert_eq!(backoff_secs(3), 8);
536 assert_eq!(backoff_secs(99), 30);
537 }
538}
539

Indexing the workspace…