//! Telegram Bot API adapter — long-polling message receiver + REST sender. //! //! Uses the Telegram Bot API over HTTPS to receive updates via `getUpdates` //! and send messages via `sendMessage` / `sendDocument`. use async_trait::async_trait; use reqwest::multipart; use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::Arc; use tokio::sync::mpsc; use tracing::{debug, error, info, warn}; use crate::{IncomingMessage, OutgoingMessage, Platform}; use gyandoot_core::{PlatformError, PlatformResult}; // ─── Telegram Update types (subset of Bot API) ────────────────────────── /// Minimal deserialisation wrapper for Telegram API responses. #[derive(Debug, serde::Deserialize)] struct TgApiResponse { ok: bool, result: Option, description: Option, } /// A single update from `getUpdates`. #[derive(Debug, serde::Deserialize)] struct TgUpdate { #[serde(rename = "update_id")] update_id: i64, #[serde(default)] message: Option, #[serde(default)] edited_message: Option, #[serde(default)] channel_post: Option, #[serde(default)] edited_channel_post: Option, } /// A Telegram message object (subset of fields). #[derive(Debug, serde::Deserialize)] #[allow(dead_code)] struct TgMessage { #[serde(rename = "message_id")] message_id: i64, #[serde(default)] from: Option, #[serde(default)] chat: Option, #[serde(default)] text: Option, #[serde(default)] date: Option, } /// A Telegram user. #[derive(Debug, serde::Deserialize)] #[allow(dead_code)] struct TgUser { id: i64, #[serde(default)] first_name: String, #[serde(default)] last_name: Option, #[serde(default)] username: Option, } /// A Telegram chat. #[derive(Debug, serde::Deserialize)] #[allow(dead_code)] struct TgChat { id: i64, #[serde(rename = "type")] chat_type: String, #[serde(default)] title: Option, #[serde(default)] username: Option, #[serde(default)] first_name: Option, #[serde(default)] last_name: Option, } // ─── TelegramPlatform ────────────────────────────────────────────────────── /// Adapter for the Telegram Bot API (long-polling). /// /// # Example /// /// ```no_run /// use gyandoot_platform::telegram::TelegramPlatform; /// /// let platform = TelegramPlatform::new("YOUR_BOT_TOKEN"); /// ``` pub struct TelegramPlatform { /// The bot token (starts with `bot` prefix handling). bot_token: String, /// Base URL for the Bot API (e.g. `https://api.telegram.org/bot`). base_url: String, /// Shared HTTP client. client: reqwest::Client, /// The last processed `update_id` (used for offset-based long-polling). last_update_id: Arc>, /// Signal to stop the polling loop. stopped: Arc, } impl TelegramPlatform { /// Create a new Telegram platform adapter. /// /// `bot_token` is the secret token obtained from [@BotFather](https://t.me/BotFather). /// It can include or omit the `bot` prefix — the adapter normalises it. pub fn new(bot_token: impl Into) -> Self { let token: String = bot_token.into(); let prefix = if token.starts_with("bot") || token.starts_with("Bot") { String::new() } else { "bot".to_string() }; let base_url = format!("https://api.telegram.org/{prefix}{token}"); let client = reqwest::Client::builder() .timeout(std::time::Duration::from_secs(30)) .build() .expect("Failed to create reqwest Client"); Self { bot_token: token, base_url, client, last_update_id: Arc::new(std::sync::Mutex::new(0)), stopped: Arc::new(AtomicBool::new(false)), } } /// Build the full URL for a Bot API method. fn method_url(&self, method: &str) -> String { format!("{}/{}", self.base_url, method) } /// Internal: poll `getUpdates` and return parsed updates. async fn fetch_updates(&self, offset: i64, timeout: i64) -> PlatformResult> { let url = self.method_url("getUpdates"); let params = [ ("offset", offset.to_string()), ("timeout", timeout.to_string()), ("allowed_updates", r#"["message"]"#.to_string()), ]; let resp = self .client .post(&url) .form(¶ms) .send() .await .map_err(|e| { error!(error = %e, "Telegram getUpdates request failed"); PlatformError::Network(e) })?; let status = resp.status(); let body_text = resp.text().await.map_err(|e| { error!(error = %e, "Failed to read Telegram response body"); PlatformError::Network(e) })?; if !status.is_success() { warn!( status = %status, body = %body_text, "Telegram API returned non-success status" ); return Err(PlatformError::Send(format!( "HTTP {status}: {body_text}" ))); } let api_resp: TgApiResponse> = serde_json::from_str(&body_text).map_err(|e| { error!(error = %e, body = %body_text, "Failed to parse Telegram getUpdates response"); PlatformError::Send(format!("JSON parse error: {e}")) })?; if !api_resp.ok { let desc = api_resp.description.unwrap_or_default(); warn!(description = %desc, "Telegram API returned ok=false"); return Err(PlatformError::Send(format!("API error: {desc}"))); } Ok(api_resp.result.unwrap_or_default()) } /// Internal: send a text message via `sendMessage`. async fn send_message_internal(&self, chat_id: &str, text: &str, reply_to: Option) -> PlatformResult<()> { let url = self.method_url("sendMessage"); let mut params = vec![ ("chat_id", chat_id.to_string()), ("text", text.to_string()), ("parse_mode", "HTML".to_string()), ]; if let Some(reply) = reply_to { params.push(("reply_to_message_id", reply.to_string())); } let resp = self .client .post(&url) .form(¶ms) .send() .await .map_err(|e| { error!(error = %e, "Telegram sendMessage request failed"); PlatformError::Network(e) })?; let status = resp.status(); if !status.is_success() { let body = resp.text().await.unwrap_or_default(); warn!(status = %status, body = %body, "Telegram sendMessage failed"); return Err(PlatformError::Send(format!("HTTP {status}: {body}"))); } debug!(chat_id = %chat_id, text_len = %text.len(), "Telegram message sent"); Ok(()) } /// Internal: send a file via `sendDocument`. #[allow(dead_code)] async fn send_file_internal( &self, chat_id: &str, file_bytes: Vec, filename: &str, caption: Option<&str>, ) -> PlatformResult<()> { let url = self.method_url("sendDocument"); let file_part = multipart::Part::bytes(file_bytes) .file_name(filename.to_string()) .mime_str("application/octet-stream") .map_err(|e| PlatformError::Send(format!("MIME error: {e}")))?; let mut form = multipart::Form::new() .text("chat_id", chat_id.to_string()) .part("document", file_part); if let Some(cap) = caption { form = form.text("caption", cap.to_string()); } let resp = self .client .post(&url) .multipart(form) .send() .await .map_err(|e| { error!(error = %e, "Telegram sendDocument request failed"); PlatformError::Network(e) })?; let status = resp.status(); if !status.is_success() { let body = resp.text().await.unwrap_or_default(); warn!(status = %status, body = %body, "Telegram sendDocument failed"); return Err(PlatformError::Send(format!("HTTP {status}: {body}"))); } debug!(chat_id = %chat_id, filename = %filename, "Telegram document sent"); Ok(()) } /// Internal: send a chat action (typing indicator). async fn send_chat_action(&self, chat_id: &str, action: &str) -> PlatformResult<()> { let url = self.method_url("sendChatAction"); let params = [ ("chat_id", chat_id.to_string()), ("action", action.to_string()), ]; let resp = self .client .post(&url) .form(¶ms) .send() .await .map_err(|e| { debug!(error = %e, "Telegram sendChatAction failed (non-critical)"); return PlatformError::Network(e); })?; if !resp.status().is_success() { // sendChatAction failures are non-critical debug!( status = %resp.status(), "Telegram sendChatAction non-success (ignored)" ); } Ok(()) } /// Try to send a file upload. Returns true if the file was sent. /// Used for future extensibility. #[allow(dead_code)] async fn try_send_file( &self, _channel_id: &str, _text: &str, _reply_to: Option, ) -> PlatformResult { // Not yet automatically detecting files from message text. // This method provides the hook for future file content detection. Ok(false) } /// Normalise a Telegram chat ID to string. #[allow(dead_code)] fn chat_id_to_string(chat_id: i64) -> String { chat_id.to_string() } } #[async_trait] impl Platform for TelegramPlatform { fn name(&self) -> &str { "telegram" } async fn start(&self) -> PlatformResult> { info!("Starting Telegram long-polling adapter"); self.stopped.store(false, Ordering::SeqCst); let (tx, rx) = mpsc::channel::(256); let bot_token = self.bot_token.clone(); let base_url = self.base_url.clone(); let client = self.client.clone(); let last_update_id = Arc::clone(&self.last_update_id); let stopped = Arc::clone(&self.stopped); // Spawn the polling loop on a background task. tokio::spawn(async move { let platform = TelegramPlatform { bot_token, base_url, client, last_update_id, stopped, }; loop { if platform.stopped.load(Ordering::SeqCst) { info!("Telegram polling loop stopped"); break; } let offset = { let id = platform.last_update_id.lock().unwrap(); *id + 1 }; match platform.fetch_updates(offset, 30).await { Ok(updates) => { for update in updates { // Track the highest update_id { let mut id = platform.last_update_id.lock().unwrap(); if update.update_id >= *id { *id = update.update_id; } } // Extract the message from whichever field is populated let msg = update .message .or(update.edited_message) .or(update.channel_post) .or(update.edited_channel_post); if let Some(tg_msg) = msg { let text = tg_msg.text.unwrap_or_default(); let chat_id = tg_msg .chat .as_ref() .map(|c| c.id.to_string()) .unwrap_or_default(); let user_id = tg_msg .from .as_ref() .map(|u| u.id.to_string()) .unwrap_or_default(); let username = tg_msg .from .as_ref() .and_then(|u| u.username.clone()) .or_else(|| { tg_msg .from .as_ref() .map(|u| u.first_name.clone()) }) .unwrap_or_default(); let message_id = tg_msg.message_id.to_string(); // Skip empty messages if text.trim().is_empty() { continue; } let incoming = IncomingMessage { platform: "telegram".to_string(), channel_id: chat_id, user_id, username, text, message_id, }; if tx.send(incoming).await.is_err() { warn!("Telegram receiver channel closed"); return; } } } } Err(e) => { warn!(error = %e, "Telegram poll error, retrying in 5s"); tokio::time::sleep(std::time::Duration::from_secs(5)).await; } } } }); info!("Telegram polling loop spawned"); Ok(rx) } async fn send(&self, message: OutgoingMessage) -> PlatformResult<()> { let chat_id = &message.channel_id; let text = &message.text; // Try to extract reply_to as numeric message ID let reply_to_id: Option = message .reply_to .as_ref() .and_then(|r| r.parse::().ok()); self.send_message_internal(chat_id, text, reply_to_id).await } async fn send_typing(&self, channel_id: &str) -> PlatformResult<()> { self.send_chat_action(channel_id, "typing").await } async fn shutdown(&self) -> PlatformResult<()> { info!("Shutting down Telegram adapter"); self.stopped.store(true, Ordering::SeqCst); Ok(()) } } // ─── Tests ────────────────────────────────────────────────────────────────── #[cfg(test)] mod tests { use super::*; #[test] fn test_new_platform_with_prefix() { let p = TelegramPlatform::new("bot12345:ABC-DEF"); assert!(p.base_url.contains("bot12345:ABC-DEF")); } #[test] fn test_new_platform_without_prefix() { let p = TelegramPlatform::new("12345:ABC-DEF"); assert!(p.base_url.contains("bot12345:ABC-DEF")); } #[test] fn test_method_url() { let p = TelegramPlatform::new("test:token"); assert!(p.method_url("sendMessage").ends_with("/sendMessage")); } #[test] fn test_chat_id_to_string() { assert_eq!(TelegramPlatform::chat_id_to_string(-12345), "-12345"); } }