1. 1//! Telegram surface adapter (docs/design/22-gateway.md G1): long-polls the
2. 2//! Bot API and bridges messages through `POST /gateway/inbound`, then
3. 3//! delivers the final assistant text back via `sendMessage`; audio-originated
4. 4//! turns also receive a governed native `sendVoice` reply when configured.
5. 5//!
6. 6//! The gateway stays transport-agnostic; this client runs anywhere it can
7. 7//! reach both Telegram and a vak-server — laptop, VPS, sidecar. Launched via
8. 8//! `vak telegram --server URL --token GATEWAY_TOKEN` with
9. 9//! `TELEGRAM_BOT_TOKEN` in the environment (the credential store included).
10. 10
11. 11use serde_json::Value;
12. 12use std::path::PathBuf;
13. 13
14. 14use crate::gateway::{InboundChannel, InboundRequest};
15. 15
16. 16/// Convert the server's canonical WAV synthesis artifact to Telegram's
17. 17/// `sendVoice` contract (Ogg/Opus). ffmpeg is an optional host capability;
18. 18/// when absent the caller can retain the text response and report the
19. 19/// actionable error instead of sending malformed media.
20. 20async fn wav_to_telegram_opus(wav: Vec<u8>) -> Result<Vec<u8>, String> {
21. 21 use tokio::io::AsyncWriteExt;
22. 22 let mut child = tokio::process::Command::new("ffmpeg")
23. 23 .args([
24. 24 "-hide_banner",
25. 25 "-loglevel",
26. 26 "error",
27. 27 "-f",
28. 28 "wav",
29. 29 "-i",
30. 30 "pipe:0",
31. 31 "-c:a",
32. 32 "libopus",
33. 33 "-b:a",
34. 34 "32k",
35. 35 "-vbr",
36. 36 "on",
37. 37 "-f",
38. 38 "ogg",
39. 39 "pipe:1",
40. 40 ])
41. 41 .stdin(std::process::Stdio::piped())
42. 42 .stdout(std::process::Stdio::piped())
43. 43 .stderr(std::process::Stdio::piped())
44. 44 .spawn()
45. 45 .map_err(|e| format!("Telegram voice requires ffmpeg for Ogg/Opus conversion: {e}"))?;
46. 46 if let Some(mut stdin) = child.stdin.take() {
47. 47 stdin
48. 48 .write_all(&wav)
49. 49 .await
50. 50 .map_err(|e| format!("write ffmpeg input: {e}"))?;
51. 51 }
52. 52 let output = child
53. 53 .wait_with_output()
54. 54 .await
55. 55 .map_err(|e| format!("wait for ffmpeg: {e}"))?;
56. 56 if !output.status.success() || output.stdout.is_empty() {
57. 57 let detail = String::from_utf8_lossy(&output.stderr);
58. 58 return Err(format!("ffmpeg Ogg/Opus conversion failed: {detail}"));
59. 59 }
60. 60 Ok(output.stdout)
61. 61}
62. 62
63. 63fn telegram_http_error(operation: &str, error: &reqwest::Error) -> String {
64. 64 let kind = if error.is_timeout() {
65. 65 "request timed out"
66. 66 } else if error.is_connect() {
67. 67 "connection failed"
68. 68 } else if error.is_decode() {
69. 69 "response decode failed"
70. 70 } else if error.is_body() {
71. 71 "request or response body failed"
72. 72 } else if error.is_request() {
73. 73 "request failed"
74. 74 } else {
75. 75 "HTTP operation failed"
76. 76 };
77. 77 match error.status() {
78. 78 Some(status) => format!("{operation}: {kind} ({status})"),
79. 79 None => format!("{operation}: {kind}"),
80. 80 }
81. 81}
82. 82
83. 83pub struct TelegramBridge {
84. 84 /// Bot API base, e.g. `https://api.telegram.org`. Overridable for
85. 85 /// self-hosted relays and tests via `TELEGRAM_API_BASE`.
86. 86 pub api_base: String,
87. 87 pub bot_token: String,
88. 88 /// The env var this bridge's credential lives in.
89. 89 ///
90. 90 /// Held so the bridge can re-read it while running: a token resolved
91. 91 /// once at startup made revocation ineffective until a restart, which
92. 92 /// is why an API handler used to bounce this process
93. 93 /// (`surfaces::CredentialWatch`). Empty disables the watch, for tests
94. 94 /// that pass a literal token.
95. 95 pub token_env: String,
96. 96 pub gateway_url: String,
97. 97 pub gateway_token: String,
98. 98 /// Directory for the per-token single-instance lock
99. 99 /// (`$VAK_HOME/locks`). None skips locking (tests only).
100. 100 pub locks_dir: Option<PathBuf>,
101. 101 /// This bot's id in the admin console's Bots list (`--bot-id`), when
102. 102 /// running the multi-bot path. Forwarded on every inbound message so a
103. 103 /// chat's first-sight pending entry already knows which bot delivered
104. 104 /// it (docs/design/34) — `None` for the legacy single-bot flow.
105. 105 pub bot_id: Option<String>,
106. 106}
107. 107
108. 108impl InboundChannel for TelegramBridge {
109. 109 fn surface(&self) -> &'static str {
110. 110 "telegram"
111. 111 }
112. 112}
113. 113
114. 114/// Why a poll failed -- recovery differs by kind.
115. 115#[derive(Debug, Clone, PartialEq, Eq)]
116. 116pub enum PollBlock {
117. 117 /// Another consumer holds the getUpdates long-poll. Telegram allows
118. 118 /// exactly one per token; this is OWNERSHIP, not an outage.
119. 119 Conflict,
120. 120 /// Upstream blip / network / 5xx: retry shortly.
121. 121 Transient,
122. 122}
123. 123
124. 124/// Classify by message content (the bridge stores formatted errors).
125. 125pub fn classify_poll_error(err: &str) -> PollBlock {
126. 126 if err.contains("409") || err.to_lowercase().contains("conflict") {
127. 127 PollBlock::Conflict
128. 128 } else {
129. 129 PollBlock::Transient
130. 130 }
131. 131}
132. 132
133. 133/// Exponential standby backoff while a rival owns the bot: doubling,
134. 134/// capped at 30s. Pure so the schedule is testable.
135. 135pub fn standby_backoff_secs(attempt: u32) -> u64 {
136. 136 let exp = 1u64 << attempt.min(5);
137. 137 exp.min(30)
138. 138}
139. 139
140. 140fn hostname_fallback() -> String {
141. 141 std::process::Command::new("hostname")
142. 142 .output()
143. 143 .ok()
144. 144 .map(|o| String::from_utf8_lossy(&o.stdout).trim().to_string())
145. 145 .filter(|s| !s.is_empty())
146. 146 .unwrap_or_else(|| "unknown-host".into())
147. 147}
148. 148
149. 149fn identity() -> String {
150. 150 format!("{}[pid {}]", hostname_fallback(), std::process::id())
151. 151}
152. 152
153. 153/// Cross-process single-instance guard keyed by bot-token hash: two
154. 154/// bridges on one machine can never fight each other (flock releases
155. 155/// automatically when the holder dies, so stale locks are impossible).
156. 156/// Cross-process AND cross-description single-instance guard keyed by
157. 157/// bot-token hash: O_EXCL marker plus PID liveness check. A crashed
158. 158/// holder leaves a stale marker; the next contender detects the dead PID
159. 159/// and takes over. No unsafe, no extra dependencies.
160. 160#[derive(Debug)]
161. 161pub struct InstanceLock {
162. 162 path: PathBuf,
163. 163 owned: bool,
164. 164}
165. 165
166. 166impl Drop for InstanceLock {
167. 167 fn drop(&mut self) {
168. 168 if self.owned {
169. 169 let _ = std::fs::remove_file(&self.path);
170. 170 }
171. 171 }
172. 172}
173. 173
174. 174impl InstanceLock {
175. 175 pub fn acquire(locks_dir: &PathBuf, bot_token: &str) -> Result<InstanceLock, String> {
176. 176 std::fs::create_dir_all(locks_dir)
177. 177 .map_err(|e| format!("lock dir {}: {e}", locks_dir.display()))?;
178. 178 let mut h: u64 = 0xcbf2_9ce4_8422_2325;
179. 179 for b in bot_token.as_bytes() {
180. 180 h ^= u64::from(*b);
181. 181 h = h.wrapping_mul(0x0000_0100_0000_01b3);
182. 182 }
183. 183 let path = locks_dir.join(format!("telegram-{h:016x}.lock"));
184. 184
185. 185 if let Ok(lock) = Self::try_create(&path) {
186. 186 return Ok(lock);
187. 187 }
188. 188
189. 189 // Marker exists: is its holder still alive?
190. 190 let holder = std::fs::read_to_string(&path).unwrap_or_default();
191. 191 let pid: Option<u32> = holder
192. 192 .split_whitespace()
193. 193 .find_map(|t| t.parse::<u32>().ok());
194. 194 match pid {
195. 195 Some(p) if Self::pid_alive(p) => Err(format!(
196. 196 "another vak telegram bridge already owns this bot\n \
197. 197 lock: {}\n \
198. 198 holder pid: {p}\n \
199. 199 stop it first (launchctl kickstart -k gui/$(id -u)/com.vak.telegram,\n \
200. 200 or kill the stale process); rotating TELEGRAM_BOT_TOKEN also helps",
201. 201 path.display()
202. 202 )),
203. 203 _ => {
204. 204 // Stale marker (crashed holder): take over.
205. 205 let _ = std::fs::remove_file(&path);
206. 206 Self::try_create(&path)
207. 207 }
208. 208 }
209. 209 }
210. 210
211. 211 fn try_create(path: &PathBuf) -> Result<InstanceLock, String> {
212. 212 let file = std::fs::OpenOptions::new()
213. 213 .write(true)
214. 214 .create_new(true)
215. 215 .open(path)
216. 216 .map_err(|e| e.to_string())?;
217. 217 drop(file);
218. 218 let me = format!("{} pid {}\n", hostname_fallback(), std::process::id());
219. 219 std::fs::write(path, &me).map_err(|e| e.to_string())?;
220. 220 Ok(InstanceLock {
221. 221 path: path.clone(),
222. 222 owned: true,
223. 223 })
224. 224 }
225. 225
226. 226 /// Safe liveness probe without libc: `kill -0` via a subprocess.
227. 227 fn pid_alive(pid: u32) -> bool {
228. 228 std::process::Command::new("kill")
229. 229 .arg("-0")
230. 230 .arg(pid.to_string())
231. 231 .stdout(std::process::Stdio::null())
232. 232 .stderr(std::process::Stdio::null())
233. 233 .status()
234. 234 .map(|st| st.success())
235. 235 .unwrap_or(false)
236. 236 }
237. 237}
238. 238
239. 239struct TelegramUpdate {
240. 240 update_id: i64,
241. 241 chat_id: i64,
242. 242 /// Telegram's own numeric user id for whoever sent the message. Falls
243. 243 /// back to the chat id (never empty/placeholder) when Telegram omits
244. 244 /// `from` (rare, e.g. channel posts) so `InboundRequest::new` never
245. 245 /// sees a blank sender.
246. 246 sender_id: i64,
247. 247 text: String,
248. 248 /// Largest-photo file id when the message carries an image.
249. 249 photo_file_id: Option<String>,
250. 250 /// A non-photo file attachment (code, logs, CSVs, ...).
251. 251 document: Option<TelegramDocument>,
252. 252 /// Telegram voice-note file, kept distinct from documents so the
253. 253 /// gateway can select a transcription adapter rather than an image path.
254. 254 voice: Option<TelegramDocument>,
255. 255 /// Set instead of the fields above when this update is an
256. 256 /// inline-keyboard button tap rather than a message.
257. 257 callback: Option<TelegramCallback>,
258. 258}
259. 259
260. 260struct TelegramDocument {
261. 261 file_id: String,
262. 262 file_name: String,
263. 263 mime_type: String,
264. 264}
265. 265
266. 266/// One inline-keyboard tap: `data` is the `callback_data` set when the
267. 267/// button was built (`"approve:<request_id>"` / `"deny:<request_id>"`).
268. 268struct TelegramCallback {
269. 269 callback_id: String,
270. 270 data: String,
271. 271 chat_id: i64,
272. 272 sender_id: i64,
273. 273}
274. 274
275. 275struct GatewayReply {
276. 276 text: String,
277. 277 delivery: Option<vak_delivery::DeliveryPacket>,
278. 278 session_id: Option<String>,
279. 279 /// Office drafts the turn made, sent back as documents.
280. 280 files: Vec<ReturnedFile>,
281. 281}
282. 282
283. 283#[derive(serde::Deserialize)]
284. 284struct ReturnedFile {
285. 285 name: String,
286. 286 mime: String,
287. 287 /// Base64.
288. 288 data: String,
289. 289 caption: String,
290. 290}
291. 291
292. 292/// A tapped button's `callback_data` ("approve:<id>" / "deny:<id>"),
293. 293/// translated to the verdict text a typed chat reply would produce ("yes
294. 294/// <id>" / "no <id>") so `parse_verdict` in `gateway.rs` resolves it
295. 295/// through the one existing path. `None` for anything else — a stale or
296. 296/// tampered-with callback should be rejected, not guessed at.
297. 297fn callback_data_to_verdict_text(data: &str) -> (&str, Option<String>) {
298. 298 let mut parts = data.splitn(2, ':');
299. 299 let verb = parts.next().unwrap_or_default();
300. 300 let request_id = parts.next().unwrap_or_default();
301. 301 match verb {
302. 302 "approve" => (verb, Some(format!("yes {request_id}"))),
303. 303 "deny" => (verb, Some(format!("no {request_id}"))),
304. 304 _ => (verb, None),
305. 305 }
306. 306}
307. 307
308. 308fn http() -> reqwest::Client {
309. 309 static CLIENT: std::sync::OnceLock<reqwest::Client> = std::sync::OnceLock::new();
310. 310 CLIENT
311. 311 .get_or_init(|| {
312. 312 reqwest::Client::builder()
313. 313 .timeout(std::time::Duration::from_secs(300))
314. 314 .build()
315. 315 .unwrap_or_default()
316. 316 })
317. 317 .clone()
318. 318}
319. 319
320. 320impl TelegramBridge {
321. 321 /// The gateway's one inbound document cap; a larger upload is not
322. 322 /// downloaded, and the sender is told rather than the upload silently
323. 323 /// vanishing.
324. 324 const DOCUMENT_MAX_BYTES: usize = crate::gateway::INBOUND_DOCUMENT_MAX_BYTES;
325. 325
326. 326 /// Long-poll once, route every text through the gateway, deliver each
327. 327 /// reply. Returns the next offset even when nothing arrived.
328. 328 pub async fn tick(&self, offset: i64) -> Result<i64, String> {
329. 329 let updates = self.get_updates(offset).await?;
330. 330 let mut next = offset;
331. 331 for u in updates {
332. 332 if let Some(cb) = &u.callback {
333. 333 if let Err(e) = self.handle_callback(cb).await {
334. 334 eprintln!("[telegram] callback handling failed: {e}");
335. 335 }
336. 336 // Resolving a gate twice is a no-op on the server side, so
337. 337 // this is safe to advance unconditionally — unlike a text
338. 338 // turn, replaying a button tap can't re-run a tool.
339. 339 next = next.max(u.update_id + 1);
340. 340 continue;
341. 341 }
342. 342 if u.text.trim().is_empty()
343. 343 && u.photo_file_id.is_none()
344. 344 && u.document.is_none()
345. 345 && u.voice.is_none()
346. 346 {
347. 347 continue;
348. 348 }
349. 349 let mut attachments = Vec::new();
350. 350 let mut text = u.text.clone();
351. 351 if let Some(file_id) = &u.photo_file_id {
352. 352 match self.fetch_photo_base64(file_id).await {
353. 353 Ok((mime, data)) => attachments.push(serde_json::json!({
354. 354 "mime": mime, "data": data, "kind": "image"
355. 355 })),
356. 356 Err(e) => eprintln!("[telegram] photo download failed: {e}"),
357. 357 }
358. 358 }
359. 359 if let Some(doc) = &u.document {
360. 360 match self.fetch_document_base64(doc).await {
361. 361 Ok(Some(data)) => attachments.push(serde_json::json!({
362. 362 "mime": doc.mime_type,
363. 363 "data": data,
364. 364 "filename": doc.file_name,
365. 365 "kind": "document",
366. 366 })),
367. 367 Ok(None) => {
368. 368 text = format!(
369. 369 "{text}\n\n[attached file '{}' exceeds the {} KiB channel limit; \
370. 370 not received]",
371. 371 doc.file_name,
372. 372 Self::DOCUMENT_MAX_BYTES / 1024
373. 373 );
374. 374 }
375. 375 Err(e) => eprintln!("[telegram] document download failed: {e}"),
376. 376 }
377. 377 }
378. 378 if let Some(voice) = &u.voice {
379. 379 match self.download_file(&voice.file_id).await {
380. 380 Ok((mime, bytes)) => {
381. 381 use base64::Engine as _;
382. 382 attachments.push(serde_json::json!({
383. 383 "mime": if mime == "application/octet-stream" { "audio/ogg" } else { &mime },
384. 384 "data": base64::engine::general_purpose::STANDARD.encode(bytes),
385. 385 "kind": "audio",
386. 386 "filename": voice.file_name,
387. 387 }));
388. 388 }
389. 389 Err(e) => eprintln!("[telegram] voice download failed: {e}"),
390. 390 }
391. 391 }
392. 392 let reply = self
393. 393 .process(u.chat_id, u.sender_id, &text, &attachments)
394. 394 .await;
395. 395 // Deliver whatever we got — an error notice beats silence, but a
396. 396 // failed send must not lose our offset progress either way.
397. 397 self.send_message(u.chat_id, &reply)
398. 398 .await
399. 399 .map_err(|error| format!("send to {}: {error}", u.chat_id))?;
400. 400 for file in &reply.files {
401. 401 if let Err(error) = self.send_document(u.chat_id, file).await {
402. 402 eprintln!("[telegram] {} not sent: {error}", file.name);
403. 403 let notice = GatewayReply {
404. 404 text: format!("{} could not be sent here: {error}", file.name),
405. 405 delivery: None,
406. 406 session_id: None,
407. 407 files: Vec::new(),
408. 408 };
409. 409 let _ = self.send_message(u.chat_id, &notice).await;
410. 410 }
411. 411 }
412. 412 // Native voice delivery is best-effort: text remains the durable
413. 413 // fallback when voice is disabled, unconfigured, or unavailable.
414. 414 if attachments
415. 415 .iter()
416. 416 .any(|a| a.get("kind").and_then(Value::as_str) == Some("audio"))
417. 417 && !reply.text.trim().is_empty()
418. 418 && let Err(error) = self
419. 419 .send_voice(u.chat_id, &reply.text, reply.session_id.as_deref())
420. 420 .await
421. 421 {
422. 422 eprintln!("[telegram] voice reply unavailable: {error}");
423. 423 }
424. 424 next = next.max(u.update_id + 1);
425. 425 }
426. 426 Ok(next)
427. 427 }
428. 428
429. 429 /// A tapped approval button, translated into the same verdict text the
430. 430 /// yes/no chat flow already understands (`parse_verdict` in
431. 431 /// `gateway.rs`) — reuses all existing gate-resolution logic instead of
432. 432 /// adding a second path. `answerCallbackQuery` is mandatory: without it
433. 433 /// Telegram leaves the button showing a permanent loading spinner.
434. 434 async fn handle_callback(&self, cb: &TelegramCallback) -> Result<(), String> {
435. 435 let (verb, Some(text)) = callback_data_to_verdict_text(&cb.data) else {
436. 436 return self
437. 437 .answer_callback(&cb.callback_id, "Unknown action")
438. 438 .await;
439. 439 };
440. 440 let req =
441. 441 InboundRequest::new(self, cb.chat_id.to_string(), cb.sender_id.to_string(), text)?
442. 442 .with_bot_id(self.bot_id.clone());
443. 443 let res = http()
444. 444 .post(format!("{}/gateway/inbound", self.gateway_url))
445. 445 .bearer_auth(&self.gateway_token)
446. 446 .json(&req)
447. 447 .send()
448. 448 .await;
449. 449 let toast = match res {
450. 450 Ok(r) if r.status().is_success() => {
451. 451 if verb == "approve" {
452. 452 "Approved"
453. 453 } else {
454. 454 "Denied"
455. 455 }
456. 456 }
457. 457 Ok(r) => {
458. 458 eprintln!("[telegram] callback resolve returned {}", r.status());
459. 459 "Could not resolve (see server logs)"
460. 460 }
461. 461 Err(e) => {
462. 462 eprintln!("[telegram] callback resolve failed: {e}");
463. 463 "Could not reach gateway"
464. 464 }
465. 465 };
466. 466 self.answer_callback(&cb.callback_id, toast).await
467. 467 }
468. 468
469. 469 async fn answer_callback(&self, callback_id: &str, text: &str) -> Result<(), String> {
470. 470 let resp = http()
471. 471 .post(format!(
472. 472 "{}/bot{}/answerCallbackQuery",
473. 473 self.api_base, self.bot_token
474. 474 ))
475. 475 .json(&serde_json::json!({ "callback_query_id": callback_id, "text": text }))
476. 476 .send()
477. 477 .await
478. 478 .map_err(|e| telegram_http_error("answerCallbackQuery", &e))?;
479. 479 if !resp.status().is_success() {
480. 480 return Err(format!("answerCallbackQuery returned {}", resp.status()));
481. 481 }
482. 482 Ok(())
483. 483 }
484. 484
485. 485 /// One message through the gateway contract; wait for the final text.
486. 486 /// `attachments` are base64 image payloads posted alongside the text.
487. 487 async fn process(
488. 488 &self,
489. 489 chat_id: i64,
490. 490 sender_id: i64,
491. 491 text: &str,
492. 492 attachments: &[serde_json::Value],
493. 493 ) -> GatewayReply {
494. 494 // 0c-03: real per-user chat/sender, not a fixed placeholder — see
495. 495 // InboundRequest::new for why that distinction is enforced here.
496. 496 let req = match InboundRequest::new(
497. 497 self,
498. 498 chat_id.to_string(),
499. 499 sender_id.to_string(),
500. 500 text.to_string(),
501. 501 ) {
502. 502 Ok(req) => req
503. 503 .with_attachments(attachments.to_vec())
504. 504 .waiting()
505. 505 .accepting_files()
506. 506 .with_bot_id(self.bot_id.clone()),
507. 507 Err(e) => {
508. 508 return GatewayReply {
509. 509 text: format!("(bridge refused to send: {e})"),
510. 510 delivery: None,
511. 511 session_id: None,
512. 512 files: Vec::new(),
513. 513 };
514. 514 }
515. 515 };
516. 516 let res = http()
517. 517 .post(format!("{}/gateway/inbound", self.gateway_url))
518. 518 .bearer_auth(&self.gateway_token)
519. 519 .json(&req)
520. 520 .send()
521. 521 .await;
522. 522 match res {
523. 523 Ok(r) if r.status().as_u16() == 202 => GatewayReply {
524. 524 // Queued behind a running turn; poll the transcript later —
525. 525 // for v1 tell the user the work is acknowledged.
526. 526 text: "(queued: I'm still working on your previous message)".into(),
527. 527 delivery: None,
528. 528 session_id: None,
529. 529 files: Vec::new(),
530. 530 },
531. 531 Ok(r) if r.status().is_success() => match r.json::<Value>().await {
532. 532 Ok(v) => GatewayReply {
533. 533 text: v["text"].as_str().unwrap_or("(empty reply)").to_string(),
534. 534 delivery: serde_json::from_value(v["delivery"].clone()).ok(),
535. 535 session_id: v["session_id"].as_str().map(String::from),
536. 536 files: serde_json::from_value(v["files"].clone()).unwrap_or_default(),
537. 537 },
538. 538 Err(e) => GatewayReply {
539. 539 text: format!("(bad gateway reply: {e})"),
540. 540 delivery: None,
541. 541 session_id: None,
542. 542 files: Vec::new(),
543. 543 },
544. 544 },
545. 545 Ok(r) => GatewayReply {
546. 546 text: format!("(gateway error: {})", r.status()),
547. 547 delivery: None,
548. 548 session_id: None,
549. 549 files: Vec::new(),
550. 550 },
551. 551 Err(e) => GatewayReply {
552. 552 text: format!("(gateway unreachable: {e})"),
553. 553 delivery: None,
554. 554 session_id: None,
555. 555 files: Vec::new(),
556. 556 },
557. 557 }
558. 558 }
559. 559
560. 560 async fn get_updates(&self, offset: i64) -> Result<Vec<TelegramUpdate>, String> {
561. 561 let url = format!("{}/bot{}/getUpdates", self.api_base, self.bot_token);
562. 562 let resp = http()
563. 563 .get(&url)
564. 564 .query(&[("timeout", "25"), ("offset", &offset.to_string())])
565. 565 .send()
566. 566 .await
567. 567 .map_err(|e| telegram_http_error("getUpdates", &e))?;
568. 568 let status = resp.status();
569. 569 if !status.is_success() {
570. 570 return Err(format!("getUpdates returned {status}"));
571. 571 }
572. 572 let body: Value = resp
573. 573 .json()
574. 574 .await
575. 575 .map_err(|e| telegram_http_error("getUpdates body", &e))?;
576. 576 if body["ok"].as_bool() != Some(true) {
577. 577 return Err(format!(
578. 578 "getUpdates not ok: {}",
579. 579 body["description"].as_str().unwrap_or("?")
580. 580 ));
581. 581 }
582. 582 let mut out = Vec::new();
583. 583 if let Some(items) = body["result"].as_array() {
584. 584 for item in items {
585. 585 let Some(update_id) = item["update_id"].as_i64() else {
586. 586 continue;
587. 587 };
588. 588 // An inline-keyboard tap arrives as callback_query, not
589. 589 // message; route it separately and skip straight to the
590. 590 // next update.
591. 591 if let Some(cq) = item.get("callback_query").filter(|v| !v.is_null()) {
592. 592 let Some(callback_id) = cq["id"].as_str() else {
593. 593 continue;
594. 594 };
595. 595 let chat_id = cq["message"]["chat"]["id"].as_i64().unwrap_or(0);
596. 596 let sender_id = cq["from"]["id"].as_i64().unwrap_or(chat_id);
597. 597 out.push(TelegramUpdate {
598. 598 update_id,
599. 599 chat_id,
600. 600 sender_id,
601. 601 text: String::new(),
602. 602 photo_file_id: None,
603. 603 document: None,
604. 604 voice: None,
605. 605 callback: Some(TelegramCallback {
606. 606 callback_id: callback_id.to_string(),
607. 607 data: cq["data"].as_str().unwrap_or_default().to_string(),
608. 608 chat_id,
609. 609 sender_id,
610. 610 }),
611. 611 });
612. 612 continue;
613. 613 }
614. 614 // Text, photo, and document messages are routed; edits and
615. 615 // other media advance the offset so they are never replayed.
616. 616 let msg = &item["message"];
617. 617 let chat_id = msg["chat"]["id"].as_i64();
618. 618 let sender_id = msg["from"]["id"].as_i64();
619. 619 // Words sent with a photo or file arrive as its caption.
620. 620 let text = msg["text"]
621. 621 .as_str()
622. 622 .or_else(|| msg["caption"].as_str())
623. 623 .map(String::from);
624. 624 let photo_file_id = msg["photo"]
625. 625 .as_array()
626. 626 .and_then(|sizes| sizes.last())
627. 627 .and_then(|largest| largest["file_id"].as_str())
628. 628 .map(String::from);
629. 629 let document =
630. 630 msg["document"]["file_id"]
631. 631 .as_str()
632. 632 .map(|file_id| TelegramDocument {
633. 633 file_id: file_id.to_string(),
634. 634 file_name: msg["document"]["file_name"]
635. 635 .as_str()
636. 636 .unwrap_or("file")
637. 637 .to_string(),
638. 638 mime_type: msg["document"]["mime_type"]
639. 639 .as_str()
640. 640 .unwrap_or("application/octet-stream")
641. 641 .to_string(),
642. 642 });
643. 643 let voice = msg["voice"]["file_id"]
644. 644 .as_str()
645. 645 .map(|file_id| TelegramDocument {
646. 646 file_id: file_id.to_string(),
647. 647 file_name: "voice.ogg".into(),
648. 648 mime_type: "audio/ogg".into(),
649. 649 });
650. 650 let routable = chat_id.is_some()
651. 651 && (text.is_some()
652. 652 || photo_file_id.is_some()
653. 653 || document.is_some()
654. 654 || voice.is_some());
655. 655 match (chat_id, routable) {
656. 656 (Some(chat_id), true) => out.push(TelegramUpdate {
657. 657 update_id,
658. 658 chat_id,
659. 659 sender_id: sender_id.unwrap_or(chat_id),
660. 660 text: text.unwrap_or_default(),
661. 661 photo_file_id,
662. 662 document,
663. 663 voice,
664. 664 callback: None,
665. 665 }),
666. 666 _ => out.push(TelegramUpdate {
667. 667 update_id,
668. 668 chat_id: 0,
669. 669 sender_id: 0,
670. 670 text: String::new(),
671. 671 photo_file_id: None,
672. 672 document: None,
673. 673 voice: None,
674. 674 callback: None,
675. 675 }),
676. 676 }
677. 677 }
678. 678 }
679. 679 Ok(out)
680. 680 }
681. 681
682. 682 async fn send_message(&self, chat_id: i64, reply: &GatewayReply) -> Result<(), String> {
683. 683 // Validate the packet against the same schema/surface boundary used
684. 684 // by Slack and Discord. Telegram still keeps the plain reply text as
685. 685 // its lossless fallback when the packet is absent or incompatible.
686. 686 let rendered = reply.delivery.as_ref().and_then(|packet| {
687. 687 let body = serde_json::json!({"delivery": packet});
688. 688 super::prepared_packet(body, "telegram")
689. 689 .ok()
690. 690 .map(|packet| packet.chunks)
691. 691 });
692. 692 let (chunks, parse_html) = match rendered {
693. 693 Some(chunks) if !chunks.is_empty() => (chunks, true),
694. 694 _ => (vec![reply.text.clone()], false),
695. 695 };
696. 696 for chunk in chunks {
697. 697 let url = format!("{}/bot{}/sendMessage", self.api_base, self.bot_token);
698. 698 let mut body = serde_json::json!({
699. 699 "chat_id": chat_id,
700. 700 "text": chunk,
701. 701 "link_preview_options": { "is_disabled": true },
702. 702 });
703. 703 if parse_html {
704. 704 body["parse_mode"] = serde_json::Value::String("HTML".into());
705. 705 }
706. 706 let resp = http().post(&url).json(&body).send().await;
707. 707 match resp {
708. 708 Ok(r) if r.status().is_success() => {}
709. 709 Ok(r) => {
710. 710 let status = r.status();
711. 711 // Converter edge-case guard: resend that chunk as plain
712. 712 // text so a formatting bug degrades to ugly, not lost.
713. 713 if parse_html && status.as_u16() == 400 {
714. 714 eprintln!(
715. 715 "[telegram] HTML rejected ({status}); falling back to plain text — \
716. 716 chunk head: {}",
717. 717 chunk.chars().take(80).collect::<String>()
718. 718 );
719. 719 let fallback = serde_json::json!({
720. 720 "chat_id": chat_id,
721. 721 "text": crate::channels::strip_tags(&chunk),
722. 722 });
723. 723 let r2 = http().post(&url).json(&fallback).send().await;
724. 724 match r2 {
725. 725 Ok(r2) if r2.status().is_success() => continue,
726. 726 Ok(r2) => {
727. 727 return Err(format!(
728. 728 "sendMessage returned {} (plain retry: {})",
729. 729 status,
730. 730 r2.status()
731. 731 ));
732. 732 }
733. 733 Err(e) => {
734. 734 return Err(telegram_http_error("sendMessage retry", &e));
735. 735 }
736. 736 }
737. 737 }
738. 738 return Err(format!("sendMessage returned {status}"));
739. 739 }
740. 740 Err(e) => return Err(telegram_http_error("sendMessage", &e)),
741. 741 }
742. 742 }
743. 743 Ok(())
744. 744 }
745. 745
746. 746 /// Sends a file the turn produced, under this bot's own token, with the
747. 747 /// change summary as its caption (Telegram allows 1024 characters).
748. 748 async fn send_document(&self, chat_id: i64, file: &ReturnedFile) -> Result<(), String> {
749. 749 use base64::Engine as _;
750. 750 let bytes = base64::engine::general_purpose::STANDARD
751. 751 .decode(file.data.as_bytes())
752. 752 .map_err(|e| format!("bad file encoding: {e}"))?;
753. 753 let part = reqwest::multipart::Part::bytes(bytes)
754. 754 .file_name(file.name.clone())
755. 755 .mime_str(&file.mime)
756. 756 .map_err(|e| format!("document mime: {e}"))?;
757. 757 let caption: String = file.caption.chars().take(1024).collect();
758. 758 let body = reqwest::multipart::Form::new()
759. 759 .text("chat_id", chat_id.to_string())
760. 760 .text("caption", caption)
761. 761 .part("document", part);
762. 762 let sent = http()
763. 763 .post(format!(
764. 764 "{}/bot{}/sendDocument",
765. 765 self.api_base, self.bot_token
766. 766 ))
767. 767 .multipart(body)
768. 768 .send()
769. 769 .await
770. 770 .map_err(|e| telegram_http_error("sendDocument", &e))?;
771. 771 if !sent.status().is_success() {
772. 772 return Err(format!("sendDocument returned {}", sent.status()));
773. 773 }
774. 774 Ok(())
775. 775 }
776. 776
777. 777 async fn send_voice(
778. 778 &self,
779. 779 chat_id: i64,
780. 780 text: &str,
781. 781 session_id: Option<&str>,
782. 782 ) -> Result<(), String> {
783. 783 let response = http()
784. 784 .post(format!("{}/voice/speak", self.gateway_url))
785. 785 .bearer_auth(&self.gateway_token)
786. 786 .json(&serde_json::json!({"text": text, "format": "wav", "session_id": session_id}))
787. 787 .send()
788. 788 .await
789. 789 .map_err(|e| format!("voice speak request: {e}"))?;
790. 790 if !response.status().is_success() {
791. 791 return Err(format!("voice speak returned {}", response.status()));
792. 792 }
793. 793 let audio = response
794. 794 .bytes()
795. 795 .await
796. 796 .map_err(|e| format!("voice speak body: {e}"))?;
797. 797 if audio.is_empty() {
798. 798 return Err("voice speak returned empty audio".into());
799. 799 }
800. 800 // Telegram's native voice method only accepts an OGG container with
801. 801 // Opus audio. The provider-neutral synthesis endpoint deliberately
802. 802 // returns WAV, so negotiate the transport codec here instead of
803. 803 // silently uploading an invalid WAV as `sendVoice`.
804. 804 let audio = wav_to_telegram_opus(audio.to_vec()).await?;
805. 805 let part = reqwest::multipart::Part::bytes(audio)
806. 806 .file_name("reply.ogg")
807. 807 .mime_str("audio/ogg")
808. 808 .map_err(|e| format!("voice mime: {e}"))?;
809. 809 let body = reqwest::multipart::Form::new()
810. 810 .text("chat_id", chat_id.to_string())
811. 811 .part("voice", part);
812. 812 let sent = http()
813. 813 .post(format!("{}/bot{}/sendVoice", self.api_base, self.bot_token))
814. 814 .multipart(body)
815. 815 .send()
816. 816 .await
817. 817 .map_err(|e| format!("sendVoice: {e}"))?;
818. 818 if !sent.status().is_success() {
819. 819 return Err(format!("sendVoice returned {}", sent.status()));
820. 820 }
821. 821 Ok(())
822. 822 }
823. 823
824. 824 /// getFile → two-step download of a Telegram-hosted file, returned as
825. 825 /// raw bytes plus the `file_path` Telegram reported (its extension is
826. 826 /// how photo mime type gets inferred below). 1 MiB bot-API cap.
827. 827 async fn download_file(&self, file_id: &str) -> Result<(String, Vec<u8>), String> {
828. 828 #[derive(serde::Deserialize)]
829. 829 struct FileResp {
830. 830 ok: bool,
831. 831 result: FileMeta,
832. 832 }
833. 833 #[derive(serde::Deserialize)]
834. 834 struct FileMeta {
835. 835 file_path: Option<String>,
836. 836 }
837. 837 let meta: FileResp = http()
838. 838 .get(format!("{}/bot{}/getFile", self.api_base, self.bot_token))
839. 839 .query(&[("file_id", file_id)])
840. 840 .send()
841. 841 .await
842. 842 .map_err(|e| telegram_http_error("getFile", &e))?
843. 843 .json()
844. 844 .await
845. 845 .map_err(|e| telegram_http_error("getFile body", &e))?;
846. 846 if !meta.ok {
847. 847 return Err("getFile not ok".into());
848. 848 }
849. 849 let path = meta.result.file_path.ok_or("getFile missing file_path")?;
850. 850 let bytes = http()
851. 851 .get(format!(
852. 852 "{}/file/bot{}/{}",
853. 853 self.api_base, self.bot_token, path
854. 854 ))
855. 855 .send()
856. 856 .await
857. 857 .map_err(|e| telegram_http_error("download", &e))?
858. 858 .error_for_status()
859. 859 .map_err(|e| telegram_http_error("download status", &e))?
860. 860 .bytes()
861. 861 .await
862. 862 .map_err(|e| telegram_http_error("download body", &e))?;
863. 863 Ok((path, bytes.to_vec()))
864. 864 }
865. 865
866. 866 /// Largest-photo variant, downscaled by Telegram on the sender side —
867. 867 /// well inside vision budgets. Returned as (mime, base64).
868. 868 async fn fetch_photo_base64(&self, file_id: &str) -> Result<(String, String), String> {
869. 869 let (path, bytes) = self.download_file(file_id).await?;
870. 870 let mime = if path.ends_with(".jpg") || path.ends_with(".jpeg") {
871. 871 "image/jpeg"
872. 872 } else if path.ends_with(".webp") {
873. 873 "image/webp"
874. 874 } else {
875. 875 "image/png"
876. 876 };
877. 877 use base64::Engine as _;
878. 878 Ok((
879. 879 mime.to_string(),
880. 880 base64::engine::general_purpose::STANDARD.encode(bytes),
881. 881 ))
882. 882 }
883. 883
884. 884 /// A non-photo attachment (code, logs, CSVs, Office files, ...).
885. 885 /// `Ok(None)` means the file was over `DOCUMENT_MAX_BYTES` and was not
886. 886 /// received; the caller tells the sender rather than truncating it.
887. 887 async fn fetch_document_base64(
888. 888 &self,
889. 889 doc: &TelegramDocument,
890. 890 ) -> Result<Option<String>, String> {
891. 891 let (_, bytes) = self.download_file(&doc.file_id).await?;
892. 892 if bytes.len() > Self::DOCUMENT_MAX_BYTES {
893. 893 return Ok(None);
894. 894 }
895. 895 use base64::Engine as _;
896. 896 Ok(Some(
897. 897 base64::engine::general_purpose::STANDARD.encode(bytes),
898. 898 ))
899. 899 }
900. 900
901. 901 /// Run until the process is killed. Transient poll/send failures back
902. 902 /// off and retry; they never drop the update stream position.
903. 903 /// Non-acking ownership probe: `timeout=0, offset=-1` returns at most
904. 904 /// the LAST update and acknowledges nothing, so probing is safe before
905. 905 /// the real loop decides where to start.
906. 906 async fn probe_ownership(&self) -> Result<(), PollBlock> {
907. 907 let url = format!("{}/bot{}/getUpdates", self.api_base, self.bot_token);
908. 908 let resp = http()
909. 909 .get(&url)
910. 910 .query(&[("timeout", "0"), ("offset", "-1")])
911. 911 .send()
912. 912 .await
913. 913 .map_err(|_| PollBlock::Transient)?;
914. 914 match resp.status().as_u16() {
915. 915 200 => Ok(()),
916. 916 409 => Err(PollBlock::Conflict),
917. 917 _ => Err(PollBlock::Transient),
918. 918 }
919. 919 }
920. 920
921. 921 /// Hot-standby: while a rival owns the bot, wait quietly and take over
922. 922 /// the moment it disappears. Logs entry once, then every 10th attempt.
923. 923 async fn await_ownership(&self) {
924. 924 let id = identity();
925. 925 eprintln!(
926. 926 "[telegram] bot token is owned by ANOTHER getUpdates consumer;\n[telegram] {id} standing by as hot standby (auto-takeover on rival exit)"
927. 927 );
928. 928 let mut attempt: u32 = 0;
929. 929 loop {
930. 930 tokio::time::sleep(std::time::Duration::from_secs(standby_backoff_secs(
931. 931 attempt,
932. 932 )))
933. 933 .await;
934. 934 attempt += 1;
935. 935 match self.probe_ownership().await {
936. 936 Ok(()) => {
937. 937 eprintln!("[telegram] {id} took over polling (rival gone)");
938. 938 return;
939. 939 }
940. 940 Err(PollBlock::Conflict) => {
941. 941 if attempt.is_multiple_of(10) {
942. 942 eprintln!(
943. 943 "[telegram] still owned elsewhere ({attempt} probes); standing by"
944. 944 );
945. 945 }
946. 946 }
947. 947 Err(PollBlock::Transient) => {}
948. 948 }
949. 949 }
950. 950 }
951. 951
952. 952 pub async fn run(&self) -> Result<(), String> {
953. 953 // Local mutual exclusion first: two bridges on one host must fail
954. 954 // fast with the holder's identity instead of flapping 409s.
955. 955 let _instance_lock = match &self.locks_dir {
956. 956 Some(dir) => Some(InstanceLock::acquire(dir, &self.bot_token)?),
957. 957 None => None,
958. 958 };
959. 959
960. 960 // Ownership probe: if another machine/session holds the long-poll,
961. 961 // become hot standby instead of hammering 409s forever.
962. 962 match self.probe_ownership().await {
963. 963 Err(PollBlock::Conflict) => self.await_ownership().await,
964. 964 Err(PollBlock::Transient) | Ok(()) => {}
965. 965 }
966. 966
967. 967 let mut offset: i64 = 0;
968. 968 let mut failures: u32 = 0;
969. 969 let watch = (!self.token_env.is_empty())
970. 970 .then(|| super::CredentialWatch::new(&self.token_env, &self.bot_token));
971. 971 loop {
972. 972 // Revocation and rotation are facts about the credential
973. 973 // store, noticed here within one poll cycle, rather than
974. 974 // something an API handler orchestrates by restarting this
975. 975 // process.
976. 976 if let Some(watch) = &watch {
977. 977 match watch.check() {
978. 978 super::CredentialState::Unchanged => {}
979. 979 super::CredentialState::Rotated => {
980. 980 eprintln!(
981. 981 "[telegram] credential rotated; exiting so the service manager \
982. 982 restarts this bridge with the new one"
983. 983 );
984. 984 return Ok(());
985. 985 }
986. 986 super::CredentialState::Revoked => {
987. 987 eprintln!(
988. 988 "[telegram] credential revoked; this bridge is stopping and will \
989. 989 not poll again until a token is set"
990. 990 );
991. 991 return Ok(());
992. 992 }
993. 993 }
994. 994 }
995. 995 match self.tick(offset).await {
996. 996 Ok(next) => {
997. 997 offset = next;
998. 998 failures = 0;
999. 999 }
1000. 1000 Err(e) => {