agentmux_srv/messaging/telegram/
mod.rs

1// Copyright 2026, AgentMux Corp.
2// SPDX-License-Identifier: Apache-2.0
3
4//! Telegram messaging bridge — long-polling `getUpdates` + REST send.
5//!
6//! Lifecycle:
7//!   1. `TelegramBridge::init_global(config, http)` — call once at startup.
8//!   2. The poll loop runs in a background tokio task (retries forever).
9//!   3. `TelegramBridge::get()` returns the singleton for HTTP handler use.
10//!   4. `bridge.send(msg)` enqueues a message for the REST client.
11//!   5. `bridge.health()` returns the current connection status.
12
13mod poller;
14pub mod rest;
15pub mod types;
16
17use std::sync::{Arc, Mutex, OnceLock};
18
19use tokio::sync::mpsc;
20
21use crate::messaging::{BridgeHealth, OutboundMsg};
22
23// ── Config ─────────────────────────────────────────────────────────────────
24
25#[derive(Debug, Clone)]
26pub struct TelegramConfig {
27    /// Bot token from @BotFather.
28    pub token: String,
29    /// Allowlisted chat IDs. Inbound updates from chats not in this list are
30    /// silently dropped (no reply, no injection, no log-level above debug).
31    pub allowed_chat_ids: Vec<i64>,
32    /// Default chat ID for outbound sends when `OutboundMsg.channel_id` is empty.
33    pub default_chat_id: Option<i64>,
34    /// Agent ID to inject inbound Telegram messages into via the reactive bus.
35    /// If None, inbound messages are logged but not forwarded.
36    pub target_agent: Option<String>,
37}
38
39// ── Bridge ─────────────────────────────────────────────────────────────────
40
41pub struct TelegramBridge {
42    outbound_tx: mpsc::UnboundedSender<OutboundMsg>,
43    health: Arc<Mutex<BridgeHealth>>,
44}
45
46static GLOBAL_BRIDGE: OnceLock<TelegramBridge> = OnceLock::new();
47
48impl TelegramBridge {
49    /// Initialize the global Telegram bridge and start the poll background
50    /// task. No-op if already initialized.
51    pub fn init_global(config: TelegramConfig, http: reqwest::Client) {
52        let (outbound_tx, outbound_rx) = mpsc::unbounded_channel::<OutboundMsg>();
53        let health = Arc::new(Mutex::new(BridgeHealth::connecting("telegram")));
54
55        let bridge = TelegramBridge {
56            outbound_tx,
57            health: health.clone(),
58        };
59
60        if GLOBAL_BRIDGE.set(bridge).is_err() {
61            return; // already initialized
62        }
63
64        let allowed_chat_ids = config.allowed_chat_ids.clone();
65        let target_agent = config.target_agent.clone();
66
67        tokio::spawn(async move {
68            poller::run_poll_loop(config, http, outbound_rx, health).await;
69        });
70
71        tracing::info!(
72            "telegram_bridge: initialized (allowed_chats={:?}, target={:?})",
73            allowed_chat_ids,
74            target_agent
75        );
76    }
77
78    pub fn get() -> Option<&'static TelegramBridge> {
79        GLOBAL_BRIDGE.get()
80    }
81
82    /// Enqueue a message for delivery via Telegram REST.
83    /// Returns error if the bridge task has exited (should not happen in normal operation).
84    pub fn send(&self, msg: OutboundMsg) -> Result<(), String> {
85        self.outbound_tx
86            .send(msg)
87            .map_err(|_| "telegram_bridge: outbound channel closed".to_string())
88    }
89
90    pub fn health(&self) -> BridgeHealth {
91        self.health.lock().unwrap().clone()
92    }
93}
94
95impl crate::messaging::MessagingBridge for TelegramBridge {
96    fn health(&self) -> BridgeHealth {
97        TelegramBridge::health(self)
98    }
99}
agentmux_srv/messaging/telegram/mod.rs

agentmux_srv/messaging/telegram/
mod.rs
