use std::sync::Arc; use chrono::{DateTime, Local, Utc}; use rand::RngExt; use serde::{Deserialize, Serialize}; use teloxide::prelude::*; use teloxide::types::ParseMode; use tokio::sync::broadcast; use tokio_util::sync::CancellationToken; use tracing::{error, info, warn}; use core_api::config_api::ConfigApi; use core_api::system_bus::SystemEvent; use super::TgShared; /// Config-table key under which all Telegram bindings are stored as JSON. pub(crate) const CONFIG_KEY: &str = "telegram"; // ── Bindings schema (stored as JSON in the `config` table) ──────────────────── #[derive(Debug, Serialize, Deserialize, Default, Clone)] pub struct TelegramConfig { #[serde(default)] pub bindings: Vec, #[serde(default)] pub pending_pairings: Vec, } #[derive(Debug, Serialize, Deserialize, Clone)] pub struct Binding { pub chat_id: i64, pub user_id: String, #[serde(default, skip_serializing_if = "Option::is_none")] pub display: Option, } #[derive(Debug, Serialize, Deserialize, Clone)] pub struct PairingEntry { pub code: String, pub chat_id: i64, pub issued_at: String, } // ── Config-table read/write ──────────────────────────────────────────────────── /// Reads the Telegram config from the `config` table. Returns `Default` when /// the key is absent or unparseable (never fails the caller). pub(crate) async fn load_config(config: &dyn ConfigApi) -> anyhow::Result { match config.get(CONFIG_KEY).await? { Some(json) => Ok(serde_json::from_str(&json).unwrap_or_default()), None => Ok(TelegramConfig::default()), } } /// Writes the Telegram config to the `config` table. `ConfigApi::set` emits a /// `ConfigKeyUpdated` event when the value changes, so the in-memory cache and /// any forwarders are updated automatically. pub(crate) async fn save_config( config: &dyn ConfigApi, cfg: &TelegramConfig, ) -> anyhow::Result<()> { config.set(CONFIG_KEY, &serde_json::to_string_pretty(cfg)?).await } // ── Pairing ─────────────────────────────────────────────────────────────────── /// Pairing codes older than this are considered abandoned and pruned. const PAIRING_TTL_HOURS: i64 = 24; /// Called when an unbound `chat_id` sends a message. Generates (or reuses) a /// pairing code, persists it to the config table, and replies with instructions. pub(crate) async fn handle_pairing(bot: &Bot, chat_id: ChatId, shared: &Arc) { let mut cfg = shared.bindings.read().await.clone(); // Prune expired codes. let cutoff = Utc::now() - chrono::Duration::hours(PAIRING_TTL_HOURS); cfg.pending_pairings.retain(|e| match DateTime::parse_from_rfc3339(&e.issued_at) { Ok(ts) => ts.with_timezone(&Utc) > cutoff, Err(_) => true, }); // Reuse an existing code for this chat, or generate a new one. let (code, added) = if let Some(entry) = cfg.pending_pairings.iter().find(|e| e.chat_id == chat_id.0) { (entry.code.clone(), false) } else { let code = generate_code(); cfg.pending_pairings.push(PairingEntry { code: code.clone(), chat_id: chat_id.0, issued_at: Local::now().format("%Y-%m-%dT%H:%M:%S%:z").to_string(), }); (code, true) }; if added { if let Err(e) = save_config(&*shared.config, &cfg).await { error!(error = %e, "telegram: failed to write pairing to config table"); } else { // Update the in-memory cache immediately (the config_listener will // also fire, but this avoids a race if the user sends another // message before the event arrives). *shared.bindings.write().await = cfg.clone(); } info!(chat_id = chat_id.0, code = %code, "TELEGRAM PAIRING: code written to config table"); } bot.send_message( chat_id, format!( "🔐 Pairing required.\n\n\ Code: {code}\n\n\ Open the Plugins page in the Skald web app and paste this code \ to link your account (or ask the admin).", ), ) .parse_mode(ParseMode::Html) .await .ok(); } pub(crate) fn generate_code() -> String { const CHARS: &[u8] = b"ABCDEFGHJKLMNPQRSTUVWXYZ23456789"; let mut rng = rand::rng(); (0..6).map(|_| CHARS[rng.random_range(0..CHARS.len())] as char).collect() } /// Turns a pairing code into a binding for `user_id` (the web self-service /// flow — `Plugin::update_user_config`). Mirrors the `telegram_pairing` tool's /// bind semantics: the pending entry is consumed and any existing binding for /// that chat is replaced. Returns the bound `chat_id`. pub(crate) fn apply_pairing_code( cfg: &mut TelegramConfig, code: &str, user_id: &str, ) -> anyhow::Result { let code = code.trim(); let pos = cfg.pending_pairings.iter() .position(|e| e.code.eq_ignore_ascii_case(code)) .ok_or_else(|| anyhow::anyhow!("invalid or expired pairing code — send a message to the bot to get a new one"))?; let chat_id = cfg.pending_pairings.remove(pos).chat_id; cfg.bindings.retain(|b| b.chat_id != chat_id); cfg.bindings.push(Binding { chat_id, user_id: user_id.to_string(), display: None, }); Ok(chat_id) } // ── Config listener ──────────────────────────────────────────────────────────── /// Subscribes to the system bus and reloads the in-memory bindings whenever the /// `"telegram"` config key changes. Replaces the old file-polling watchdog. pub(crate) async fn config_listener( shared: Arc, mut rx: broadcast::Receiver, cancel: CancellationToken, ) { info!("telegram: config listener started"); loop { tokio::select! { _ = cancel.cancelled() => { info!("telegram: config listener stopped"); return; } result = rx.recv() => match result { Ok(SystemEvent::ConfigKeyUpdated { key, new_value, .. }) if key == CONFIG_KEY => { match serde_json::from_str::(&new_value) { Ok(cfg) => { let n = cfg.bindings.len(); *shared.bindings.write().await = cfg; info!(bindings = n, "telegram: bindings reloaded from config event"); } Err(e) => warn!(error = %e, "telegram: failed to parse config from event"), } } Ok(_) => {} Err(broadcast::error::RecvError::Lagged(n)) => { warn!(skipped = n, "telegram: config listener lagged"); } Err(broadcast::error::RecvError::Closed) => return, } } } } #[cfg(test)] mod tests { use super::*; fn cfg_with_pairing(code: &str, chat_id: i64) -> TelegramConfig { TelegramConfig { bindings: vec![], pending_pairings: vec![PairingEntry { code: code.to_string(), chat_id, issued_at: "2026-01-01T00:00:00+00:00".to_string(), }], } } #[test] fn pairing_code_creates_binding_and_consumes_entry() { let mut cfg = cfg_with_pairing("ABC123", 42); let chat_id = apply_pairing_code(&mut cfg, "ABC123", "u1").unwrap(); assert_eq!(chat_id, 42); assert!(cfg.pending_pairings.is_empty(), "the code must be consumed"); assert_eq!(cfg.bindings.len(), 1); assert_eq!(cfg.bindings[0].user_id, "u1"); assert_eq!(cfg.bindings[0].chat_id, 42); } #[test] fn pairing_code_is_case_and_whitespace_insensitive() { let mut cfg = cfg_with_pairing("ABC123", 42); apply_pairing_code(&mut cfg, " abc123 ", "u1").unwrap(); assert_eq!(cfg.bindings[0].user_id, "u1"); } #[test] fn pairing_replaces_an_existing_binding_for_the_same_chat() { let mut cfg = cfg_with_pairing("ABC123", 42); cfg.bindings.push(Binding { chat_id: 42, user_id: "old".into(), display: None }); cfg.bindings.push(Binding { chat_id: 99, user_id: "other".into(), display: None }); apply_pairing_code(&mut cfg, "ABC123", "u1").unwrap(); assert_eq!(cfg.bindings.len(), 2); assert!(cfg.bindings.iter().any(|b| b.chat_id == 42 && b.user_id == "u1")); assert!(cfg.bindings.iter().any(|b| b.chat_id == 99 && b.user_id == "other"), "bindings for other chats are untouched"); } #[test] fn unknown_code_fails_and_keeps_state() { let mut cfg = cfg_with_pairing("ABC123", 42); let err = apply_pairing_code(&mut cfg, "ZZZ999", "u1").unwrap_err(); assert!(err.to_string().contains("invalid or expired")); assert_eq!(cfg.pending_pairings.len(), 1, "the pending entry must survive a failed attempt"); assert!(cfg.bindings.is_empty()); } }