AnyBot/src/channels.rs

// 訊息管道: a bot answers people in Slack, Discord or Telegram (the original: Slack / Microsoft
// Teams / Discord apps, plus Telegram-style messaging channels). The original's servers receive
// those messages; a desktop app has no public address, so here each channel is polled with the
// person's own bot token, and the bot's final answer of that turn is posted back (in a thread /
// as a reply to the message).

use serde::{Deserialize, Serialize};
use serde_json::{json, Value};
use std::collections::HashMap;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::mpsc::Sender;
use std::sync::{Arc, Mutex, OnceLock};
use std::time::Duration;

/// How often Slack and Discord are checked (estimated; within both APIs' rate limits).
pub const POLL_SECS: u64 = 3;
/// Telegram long-poll wait.
pub const TELEGRAM_WAIT_SECS: u64 = 25;
/// Longest single message each service takes.
pub const DISCORD_MAX: usize = 2000;
pub const TELEGRAM_MAX: usize = 4096;
pub const SLACK_MAX: usize = 39_000;

#[derive(Clone, Copy, Serialize, Deserialize, Debug, PartialEq, Eq, Hash)]
pub enum Kind {
    Slack,
    Discord,
    Telegram,
    Teams,
    /// Mac「訊息」(iMessage / SMS).
    IMessage,
    /// Mail on this Mac.
    Email,
    /// 專屬號碼: texts to the person's Twilio number.
    Sms,
    /// X: @mentions, answered as replies.
    X,
}

impl Kind {
    pub fn label(self) -> &'static str {
        match self {
            Kind::Slack => "Slack",
            Kind::Discord => "Discord",
            Kind::Telegram => "Telegram",
            Kind::Teams => "Teams",
            Kind::IMessage => "訊息",
            Kind::Email => "郵件",
            Kind::Sms => "簡訊（Twilio）",
            Kind::X => "X",
        }
    }
    pub fn token_hint(self) -> &'static str {
        match self {
            Kind::Slack => "Bot User OAuth Token（xoxb-…）。在 api.slack.com/apps 建立 App，加上 channels:history、groups:history、im:history、chat:write 權限並安裝到工作區，再把 App 邀請進頻道。",
            Kind::Discord => "Bot Token。在 discord.com/developers 建立 Application → Bot，打開 Message Content Intent，再用 OAuth2 把 Bot 加進你的伺服器。",
            Kind::Telegram => "跟 @BotFather 說 /newbot 拿到的 Token（123456:ABC…）。",
            Kind::IMessage => "不用 Token：用這台 Mac 的「訊息」。讀新訊息要給 AnyBot「完全取用磁碟」權限（系統設定 → 隱私權與安全性），傳訊息時 macOS 會問一次能否控制「訊息」。",
            Kind::Email => "不用 Token：用這台 Mac 的「郵件」收信和回信。macOS 會問一次能否控制「郵件」。",
            Kind::Sms => "「Account SID:Auth Token」（在 console.twilio.com 首頁）。先在 Twilio 買一個號碼；Bot 會回覆傳到這個號碼的簡訊，也能用它打電話和傳簡訊（每次都先問你）。",
            Kind::X => "X 的使用者存取權杖（OAuth 2.0，需要 tweet.read、tweet.write、users.read 權限；在 developer.x.com 建立 App 取得）。Bot 會回覆提到你的貼文，也能用你的帳號發文（每次都先問你）。讀取提及需要 X API 付費方案。",
            Kind::Teams => "不用 Token：填你的 Microsoft Entra 應用程式（用戶端）ID，按「登入 Microsoft」用裝置碼登入。應用程式要在「驗證」打開「允許公用用戶端流程」，並加上 Chat.ReadWrite、ChannelMessage.Read.All、ChannelMessage.Send 權限（頻道訊息可能要管理員同意）。",
        }
    }
    pub fn target_hint(self) -> &'static str {
        match self {
            Kind::Slack => "頻道 ID（例如 C0123ABCD，在頻道資訊最下面）",
            Kind::Discord => "頻道 ID（開發者模式下對頻道按右鍵 → 複製頻道 ID）",
            Kind::Telegram => "只回這個聊天 ID（留空 = 任何人私訊 Bot 都回）",
            Kind::IMessage => "只回這個電話或 Apple ID（例如 +886912345678；留空 = 任何人）",
            Kind::Email => "只看這個信箱名稱（留空 = 收件匣）",
            Kind::Sms => "你的 Twilio 號碼（例如 +15551234567）",
            Kind::X => "（留空）",
            Kind::Teams => "chat:<聊天 ID>，或 <團隊 ID>/<頻道 ID>（在 Teams 對聊天或頻道「取得連結」可以看到）",
        }
    }
}

#[derive(Clone, Serialize, Deserialize, Debug, PartialEq)]
pub struct Channel {
    pub id: u64,
    pub kind: Kind,
    pub name: String,
    /// The AnyBot bot that answers.
    pub bot_id: u64,
    pub token: String,
    /// Slack / Discord channel id; Telegram chat id (empty = any).
    #[serde(default)]
    pub target: String,
    #[serde(default = "yes")]
    pub enabled: bool,
    /// Teams: the Entra app (client) id and tenant (empty = organizations); `token` is the refresh token.
    #[serde(default)]
    pub client_id: String,
    #[serde(default)]
    pub tenant: String,
}

fn yes() -> bool {
    true
}

/// Where an answer goes back.
#[derive(Clone, Debug, PartialEq)]
pub struct ReplyTo {
    pub kind: Kind,
    pub token: String,
    pub target: String,
    /// Slack thread ts / Discord message id / Telegram message id.
    pub thread: String,
    pub channel_name: String,
}

#[derive(Clone, Debug)]
pub struct Incoming {
    pub bot_id: u64,
    pub from: String,
    pub text: String,
    pub reply: ReplyTo,
}

#[derive(Clone, Default, Debug)]
pub struct Status {
    pub ok: bool,
    pub note: String,
    pub received: u64,
}

pub fn statuses() -> &'static Mutex<HashMap<u64, Status>> {
    static S: OnceLock<Mutex<HashMap<u64, Status>>> = OnceLock::new();
    S.get_or_init(|| Mutex::new(HashMap::new()))
}

fn set_status(id: u64, ok: bool, note: &str, got: u64) {
    let mut m = statuses().lock().unwrap();
    let s = m.entry(id).or_default();
    s.ok = ok;
    s.note = note.to_string();
    s.received += got;
}

fn agent(timeout: u64) -> ureq::Agent {
    ureq::Agent::config_builder().timeout_global(Some(Duration::from_secs(timeout))).http_status_as_error(false).build().into()
}

fn get_json(url: &str, auth: Option<&str>, timeout: u64) -> Result<Value, String> {
    let mut r = agent(timeout).get(url);
    if let Some(a) = auth {
        r = r.header("Authorization", a);
    }
    let mut resp = r.call().map_err(|e| e.to_string())?;
    let status = resp.status().as_u16();
    let text = resp.body_mut().read_to_string().unwrap_or_default();
    if status == 429 {
        return Err("rate".into());
    }
    if !(200..300).contains(&status) {
        return Err(format!("HTTP {status}：{}", text.chars().take(160).collect::<String>()));
    }
    serde_json::from_str(&text).map_err(|e| e.to_string())
}

fn post_json(url: &str, auth: Option<&str>, body: &Value) -> Result<Value, String> {
    let mut r = agent(30).post(url).header("Content-Type", "application/json; charset=utf-8");
    if let Some(a) = auth {
        r = r.header("Authorization", a);
    }
    let mut resp = r.send(body.to_string().as_bytes()).map_err(|e| e.to_string())?;
    let status = resp.status().as_u16();
    let text = resp.body_mut().read_to_string().unwrap_or_default();
    if !(200..300).contains(&status) {
        return Err(format!("HTTP {status}：{}", text.chars().take(160).collect::<String>()));
    }
    Ok(serde_json::from_str(&text).unwrap_or(Value::Null))
}

/// Split long answers on line breaks so each piece fits.
pub fn split(text: &str, max: usize) -> Vec<String> {
    let mut out = vec![];
    let mut cur = String::new();
    for line in text.split_inclusive('\n') {
        if cur.chars().count() + line.chars().count() > max && !cur.is_empty() {
            out.push(std::mem::take(&mut cur));
        }
        if line.chars().count() > max {
            let chars: Vec<char> = line.chars().collect();
            for c in chars.chunks(max) {
                out.push(c.iter().collect());
            }
        } else {
            cur.push_str(line);
        }
    }
    if !cur.trim().is_empty() {
        out.push(cur);
    }
    out
}

/// Telegram's address (a test server can stand in for it).
fn tg() -> String {
    std::env::var("ANYBOT_TELEGRAM_API").unwrap_or_else(|_| "https://api.telegram.org".into())
}

/// Check a token and say who it is.
pub fn test(kind: Kind, token: &str) -> Result<String, String> {
    let t = token.trim();
    match kind {
        Kind::Slack => {
            let v = get_json("https://slack.com/api/auth.test", Some(&format!("Bearer {t}")), 15)?;
            if v["ok"].as_bool() == Some(true) {
                Ok(format!("已連上 Slack：{}（{}）", v["user"].as_str().unwrap_or(""), v["team"].as_str().unwrap_or("")))
            } else {
                Err(format!("Slack 說：{}", v["error"].as_str().unwrap_or("不明錯誤")))
            }
        }
        Kind::Discord => {
            let v = get_json("https://discord.com/api/v10/users/@me", Some(&format!("Bot {t}")), 15)?;
            Ok(format!("已連上 Discord：{}", v["username"].as_str().unwrap_or("")))
        }
        Kind::Teams => Err("Teams 請用「登入 Microsoft」".into()),
        Kind::Sms => crate::phone::twilio_test(t),
        Kind::X => crate::phone::x_me(t).map(|(_, u)| format!("已連上 X：@{u}")),
        Kind::IMessage => imessage_rows(0).map(|_| "可以讀「訊息」了。".to_string()),
        Kind::Email => {
            let (o, ok) = crate::macapps::run("mac_mail_accounts", "{}");
            if ok { Ok("可以用「郵件」了。".into()) } else { Err(o) }
        }
        Kind::Telegram => {
            let v = get_json(&format!("{}/bot{t}/getMe", tg()), None, 15)?;
            if v["ok"].as_bool() == Some(true) {
                Ok(format!("已連上 Telegram：@{}", v["result"]["username"].as_str().unwrap_or("")))
            } else {
                Err(format!("Telegram 說：{}", v["description"].as_str().unwrap_or("不明錯誤")))
            }
        }
    }
}

/// Post the bot's answer back where the message came from.
pub fn reply(to: &ReplyTo, text: &str) -> Result<(), String> {
    let t = to.token.trim();
    match to.kind {
        // `thread` carries our own number (texts) or the post being answered (X).
        Kind::Sms => crate::phone::send_sms(t, &to.thread, &to.target, text)?,
        Kind::X => {
            let mut reply_to = to.thread.clone();
            for part in split(text, crate::phone::X_MAX - 20) {
                let url = crate::phone::x_post(t, &part, &reply_to)?;
                reply_to = url.rsplit('/').next().unwrap_or("").to_string();
            }
        }
        Kind::IMessage => {
            let (o, ok) = crate::macapps::run(crate::macapps::SEND_MESSAGE, &json!({"to": to.target, "text": text}).to_string());
            if !ok {
                return Err(o);
            }
        }
        Kind::Email => {
            // `thread` carries the subject.
            let subject = if to.thread.to_lowercase().starts_with("re:") { to.thread.clone() } else { format!("Re: {}", to.thread) };
            let (o, ok) = crate::macapps::run("mac_send_mail", &json!({"to": [to.target], "subject": subject, "body": text}).to_string());
            if !ok {
                return Err(o);
            }
        }
        Kind::Teams => {
            // `token` carries the access token for Teams replies.
            for part in split(text, 25_000) {
                crate::teams::send(t, &to.target, &to.thread, &part)?;
            }
        }
        Kind::Slack => {
            for part in split(text, SLACK_MAX) {
                let v = post_json("https://slack.com/api/chat.postMessage", Some(&format!("Bearer {t}")), &json!({"channel": to.target, "text": part, "thread_ts": to.thread}))?;
                if v["ok"].as_bool() != Some(true) {
                    return Err(format!("Slack 說：{}", v["error"].as_str().unwrap_or("不明錯誤")));
                }
            }
        }
        Kind::Discord => {
            for (i, part) in split(text, DISCORD_MAX).into_iter().enumerate() {
                let mut body = json!({"content": part, "allowed_mentions": {"parse": []}});
                if i == 0 && !to.thread.is_empty() {
                    body["message_reference"] = json!({"message_id": to.thread, "fail_if_not_exists": false});
                }
                post_json(&format!("https://discord.com/api/v10/channels/{}/messages", to.target), Some(&format!("Bot {t}")), &body)?;
            }
        }
        Kind::Telegram => {
            for (i, part) in split(text, TELEGRAM_MAX).into_iter().enumerate() {
                let mut body = json!({"chat_id": to.target, "text": part});
                if i == 0 && !to.thread.is_empty() {
                    body["reply_to_message_id"] = json!(to.thread.parse::<i64>().unwrap_or(0));
                }
                let v = post_json(&format!("{}/bot{t}/sendMessage", tg()), None, &body)?;
                if v["ok"].as_bool() == Some(false) {
                    return Err(format!("Telegram 說：{}", v["description"].as_str().unwrap_or("")));
                }
            }
        }
    }
    Ok(())
}

// ---------- reading new messages ----------

/// Slack: messages newer than `cursor` (a ts), oldest first, people only. Returns (new cursor, items).
pub fn slack_new(history: &Value, cursor: &str, own_user: &str) -> (String, Vec<(String, String, String)>) {
    let mut items: Vec<(String, String, String)> = vec![];
    let mut newest = cursor.to_string();
    if let Some(msgs) = history["messages"].as_array() {
        for m in msgs.iter().rev() {
            let ts = m["ts"].as_str().unwrap_or("").to_string();
            if ts.parse::<f64>().unwrap_or(0.0) <= cursor.parse::<f64>().unwrap_or(0.0) {
                continue;
            }
            if ts.parse::<f64>().unwrap_or(0.0) > newest.parse::<f64>().unwrap_or(0.0) {
                newest = ts.clone();
            }
            let is_bot = m.get("bot_id").is_some() || m["subtype"].is_string() || m["user"].as_str() == Some(own_user);
            if is_bot {
                continue;
            }
            let text = m["text"].as_str().unwrap_or("").replace(&format!("<@{own_user}>"), "").trim().to_string();
            if !text.is_empty() {
                items.push((m["user"].as_str().unwrap_or("someone").to_string(), text, m["thread_ts"].as_str().unwrap_or(&ts).to_string()));
            }
        }
    }
    (newest, items)
}

/// Discord: messages after `cursor` (a snowflake), oldest first, people only.
pub fn discord_new(list: &Value, cursor: &str) -> (String, Vec<(String, String, String)>) {
    let mut items = vec![];
    let mut newest = cursor.to_string();
    let n = |s: &str| s.parse::<u128>().unwrap_or(0);
    if let Some(msgs) = list.as_array() {
        let mut msgs: Vec<&Value> = msgs.iter().collect();
        msgs.sort_by_key(|m| n(m["id"].as_str().unwrap_or("0")));
        for m in msgs {
            let id = m["id"].as_str().unwrap_or("0").to_string();
            if n(&id) <= n(cursor) {
                continue;
            }
            if n(&id) > n(&newest) {
                newest = id.clone();
            }
            if m["author"]["bot"].as_bool() == Some(true) {
                continue;
            }
            let text = m["content"].as_str().unwrap_or("").trim().to_string();
            if !text.is_empty() {
                let who = m["author"]["global_name"].as_str().or(m["author"]["username"].as_str()).unwrap_or("someone").to_string();
                items.push((who, text, id));
            }
        }
    }
    (newest, items)
}

/// Telegram: (next offset, [(from, text, chat id, message id)]).
pub fn telegram_new(updates: &Value, offset: i64, only_chat: &str) -> (i64, Vec<(String, String, String, String)>) {
    let mut next = offset;
    let mut items = vec![];
    if let Some(us) = updates["result"].as_array() {
        for u in us {
            next = next.max(u["update_id"].as_i64().unwrap_or(0) + 1);
            let m = &u["message"];
            let chat = m["chat"]["id"].as_i64().map(|c| c.to_string()).unwrap_or_default();
            if chat.is_empty() || (!only_chat.trim().is_empty() && only_chat.trim() != chat) {
                continue;
            }
            if let Some(text) = m["text"].as_str() {
                let who = m["from"]["first_name"].as_str().or(m["from"]["username"].as_str()).unwrap_or("someone").to_string();
                items.push((who, text.to_string(), chat, m["message_id"].as_i64().unwrap_or(0).to_string()));
            }
        }
    }
    (next, items)
}

/// "王小明 <ming@example.com>" → "ming@example.com"
pub fn address_of(sender: &str) -> String {
    match (sender.rfind('<'), sender.rfind('>')) {
        (Some(a), Some(b)) if b > a => sender[a + 1..b].trim().to_string(),
        _ => sender.trim().to_string(),
    }
}

fn chat_db() -> String {
    dirs::home_dir().unwrap_or_default().join("Library/Messages/chat.db").display().to_string()
}

/// Messages after `after` (ROWID): (id, text, handle, from me).
pub fn imessage_rows(after: i64) -> Result<Vec<(i64, String, String, bool)>, String> {
    let q = format!("SELECT m.ROWID AS id, m.text AS text, h.id AS handle, m.is_from_me AS me FROM message m LEFT JOIN handle h ON m.handle_id = h.ROWID WHERE m.ROWID > {after} ORDER BY m.ROWID DESC LIMIT 50");
    let out = std::process::Command::new("sqlite3").args(["-readonly", "-json", &chat_db(), &q]).output().map_err(|e| e.to_string())?;
    if !out.status.success() {
        let e = String::from_utf8_lossy(&out.stderr).to_string();
        return Err(if e.contains("authorization denied") || e.contains("unable to open") {
            "讀不到「訊息」：請給 AnyBot（或「終端機」）「完全取用磁碟」權限。".into()
        } else {
            e
        });
    }
    let v: Value = serde_json::from_slice(&out.stdout).unwrap_or(json!([]));
    let mut rows: Vec<(i64, String, String, bool)> = v
        .as_array()
        .into_iter()
        .flatten()
        .map(|r| (r["id"].as_i64().unwrap_or(0), r["text"].as_str().unwrap_or("").to_string(), r["handle"].as_str().unwrap_or("").to_string(), r["me"].as_i64() == Some(1)))
        .collect();
    rows.reverse();
    Ok(rows)
}

// ---------- the pollers ----------

static GENERATION: AtomicU64 = AtomicU64::new(0);

/// (Re)start one poller per enabled channel; older pollers stop by themselves.
pub fn restart(channels: Vec<Channel>, tx: Sender<Incoming>, repaint: Arc<dyn Fn() + Send + Sync>) {
    let gen = GENERATION.fetch_add(1, Ordering::SeqCst) + 1;
    // The bot's phone and X tools use the same accounts.
    {
        let mut a = crate::phone::accounts().lock().unwrap();
        a.twilio = channels.iter().find(|c| c.enabled && c.kind == Kind::Sms && !c.token.trim().is_empty()).map(|c| (c.token.trim().to_string(), c.target.trim().to_string()));
        a.x = channels.iter().find(|c| c.enabled && c.kind == Kind::X && !c.token.trim().is_empty()).map(|c| c.token.trim().to_string());
    }
    for c in channels.into_iter().filter(|c| c.enabled && !c.token.trim().is_empty()) {
        let tx = tx.clone();
        let repaint = repaint.clone();
        std::thread::spawn(move || poll(c, gen, tx, repaint));
    }
}

fn alive(gen: u64) -> bool {
    GENERATION.load(Ordering::SeqCst) == gen
}

fn poll(c: Channel, gen: u64, tx: Sender<Incoming>, repaint: Arc<dyn Fn() + Send + Sync>) {
    let token = c.token.trim().to_string();
    let deliver = |from: String, text: String, target: String, thread: String| {
        let _ = tx.send(Incoming { bot_id: c.bot_id, from, text, reply: ReplyTo { kind: c.kind, token: token.clone(), target, thread, channel_name: c.name.clone() } });
        set_status(c.id, true, "收到新訊息", 1);
        repaint();
    };
    match c.kind {
        Kind::Slack => {
            let auth = format!("Bearer {token}");
            let own = get_json("https://slack.com/api/auth.test", Some(&auth), 15).ok().and_then(|v| v["user_id"].as_str().map(String::from)).unwrap_or_default();
            let mut cursor = String::new();
            while alive(gen) {
                let url = format!("https://slack.com/api/conversations.history?channel={}&limit=20{}", c.target.trim(), if cursor.is_empty() { String::new() } else { format!("&oldest={cursor}") });
                match get_json(&url, Some(&auth), 20) {
                    Ok(v) if v["ok"].as_bool() == Some(true) => {
                        if cursor.is_empty() {
                            // Start from now: older messages aren't answered.
                            cursor = v["messages"][0]["ts"].as_str().unwrap_or("0").to_string();
                            set_status(c.id, true, "已連上，等待新訊息", 0);
                        } else {
                            let (next, items) = slack_new(&v, &cursor, &own);
                            cursor = next;
                            for (from, text, thread) in items {
                                deliver(from, text, c.target.trim().to_string(), thread);
                            }
                        }
                    }
                    Ok(v) => set_status(c.id, false, &format!("Slack 說：{}", v["error"].as_str().unwrap_or("不明錯誤")), 0),
                    Err(e) if e == "rate" => std::thread::sleep(Duration::from_secs(20)),
                    Err(e) => set_status(c.id, false, &e, 0),
                }
                std::thread::sleep(Duration::from_secs(POLL_SECS));
            }
        }
        Kind::Discord => {
            let auth = format!("Bot {token}");
            let mut cursor = String::new();
            while alive(gen) {
                let url = if cursor.is_empty() {
                    format!("https://discord.com/api/v10/channels/{}/messages?limit=1", c.target.trim())
                } else {
                    format!("https://discord.com/api/v10/channels/{}/messages?after={cursor}&limit=50", c.target.trim())
                };
                match get_json(&url, Some(&auth), 20) {
                    Ok(v) => {
                        if cursor.is_empty() {
                            cursor = v[0]["id"].as_str().unwrap_or("0").to_string();
                            set_status(c.id, true, "已連上，等待新訊息", 0);
                        } else {
                            let (next, items) = discord_new(&v, &cursor);
                            cursor = next;
                            for (from, text, id) in items {
                                deliver(from, text, c.target.trim().to_string(), id);
                            }
                        }
                    }
                    Err(e) if e == "rate" => std::thread::sleep(Duration::from_secs(10)),
                    Err(e) => set_status(c.id, false, &e, 0),
                }
                std::thread::sleep(Duration::from_secs(POLL_SECS));
            }
        }
        Kind::Teams => {
            let mut cursor = String::new();
            let mut me = String::new();
            while alive(gen) {
                match crate::teams::access(c.id, &c.client_id, &c.tenant, &token) {
                    Ok(access) => {
                        if me.is_empty() {
                            me = crate::teams::graph_get(&access, "/me").ok().and_then(|v| v["id"].as_str().map(String::from)).unwrap_or_default();
                        }
                        match crate::teams::graph_get(&access, &crate::teams::messages_path(&c.target)) {
                            Ok(v) => {
                                if cursor.is_empty() {
                                    cursor = crate::teams::new_messages(&v, "", "").0;
                                    if cursor.is_empty() {
                                        cursor = chrono::Utc::now().format("%Y-%m-%dT%H:%M:%S").to_string();
                                    }
                                    set_status(c.id, true, "已連上，等待新訊息", 0);
                                } else {
                                    let (next, items) = crate::teams::new_messages(&v, &cursor, &me);
                                    cursor = next;
                                    for (from, text, id) in items {
                                        let _ = tx.send(Incoming { bot_id: c.bot_id, from, text, reply: ReplyTo { kind: Kind::Teams, token: access.clone(), target: c.target.trim().to_string(), thread: if c.target.starts_with("chat:") { String::new() } else { id }, channel_name: c.name.clone() } });
                                        set_status(c.id, true, "收到新訊息", 1);
                                        repaint();
                                    }
                                }
                            }
                            Err(e) if e == "rate" => std::thread::sleep(Duration::from_secs(20)),
                            Err(e) => set_status(c.id, false, &e, 0),
                        }
                    }
                    Err(e) => set_status(c.id, false, &e, 0),
                }
                std::thread::sleep(Duration::from_secs(POLL_SECS + 2));
            }
        }
        Kind::IMessage => {
            let mut cursor: Option<i64> = None;
            while alive(gen) {
                match imessage_rows(cursor.unwrap_or(0)) {
                    Ok(rows) => {
                        let newest = rows.iter().map(|r| r.0).max();
                        match cursor {
                            None => {
                                cursor = Some(newest.unwrap_or(0));
                                set_status(c.id, true, "已連上，等待新訊息", 0);
                            }
                            Some(_) => {
                                for (id, text, handle, me) in rows {
                                    cursor = Some(id.max(cursor.unwrap_or(0)));
                                    if me || text.trim().is_empty() || (!c.target.trim().is_empty() && handle != c.target.trim()) {
                                        continue;
                                    }
                                    deliver(handle.clone(), text, handle, String::new());
                                }
                            }
                        }
                    }
                    Err(e) => set_status(c.id, false, &e, 0),
                }
                std::thread::sleep(Duration::from_secs(POLL_SECS));
            }
        }
        Kind::Email => {
            let mut seen: std::collections::HashSet<String> = std::collections::HashSet::new();
            let mut first = true;
            while alive(gen) {
                let (o, ok) = crate::macapps::run("mac_read_mail", &json!({"mailbox": c.target.trim(), "unreadOnly": true, "days": 3, "limit": 20}).to_string());
                if ok {
                    let list: Value = serde_json::from_str(o.trim_start_matches("郵件：")).unwrap_or(json!([]));
                    for m in list.as_array().into_iter().flatten() {
                        let id = m["id"].as_str().unwrap_or("").to_string();
                        if !seen.insert(id.clone()) || first {
                            continue;
                        }
                        // Read the whole message to answer it.
                        let (full, ok) = crate::macapps::run("mac_read_mail", &json!({"id": id}).to_string());
                        let body: Value = if ok { serde_json::from_str(&full).unwrap_or(Value::Null) } else { Value::Null };
                        let from = m["from"].as_str().unwrap_or("").to_string();
                        let subject = m["subject"].as_str().unwrap_or("").to_string();
                        let text = format!("主旨：{subject}\n{}", body["body"].as_str().unwrap_or(""));
                        deliver(from.clone(), text, address_of(&from), subject);
                    }
                    if first {
                        set_status(c.id, true, "已連上，等待新郵件", 0);
                    }
                    first = false;
                } else {
                    set_status(c.id, false, &o, 0);
                }
                std::thread::sleep(Duration::from_secs(30));
            }
        }
        Kind::Sms => {
            let number = c.target.trim().to_string();
            let mut seen = std::collections::HashSet::new();
            let mut first = true;
            while alive(gen) {
                let path = format!("Messages.json?To={}&PageSize=20", number.replace('+', "%2B"));
                match crate::phone::twilio_get(&token, &path) {
                    Ok(v) => {
                        let items = crate::phone::sms_new(&v, &mut seen);
                        if first {
                            // Older texts aren't answered.
                            first = false;
                            set_status(c.id, true, "已連上，等待簡訊", 0);
                        } else {
                            for (_, from, body) in items {
                                deliver(from.clone(), body, from, number.clone());
                            }
                        }
                    }
                    Err(e) => set_status(c.id, false, &e, 0),
                }
                std::thread::sleep(Duration::from_secs(POLL_SECS));
            }
        }
        Kind::X => {
            let me = match crate::phone::x_me(&token) {
                Ok(m) => m,
                Err(e) => {
                    set_status(c.id, false, &e, 0);
                    return;
                }
            };
            let mut since = String::new();
            let mut first = true;
            while alive(gen) {
                match crate::phone::x_mentions(&token, &me.0, &since) {
                    Ok((newest, items)) => {
                        since = newest;
                        if first {
                            first = false;
                            set_status(c.id, true, &format!("已連上 @{}，等待提及", me.1), 0);
                        } else {
                            for (id, who, text) in items {
                                deliver(format!("@{who}"), text, who, id);
                            }
                        }
                    }
                    Err(e) => set_status(c.id, false, &e, 0),
                }
                // X's free and basic plans allow few reads: once a minute.
                for _ in 0..60 {
                    if !alive(gen) {
                        break;
                    }
                    std::thread::sleep(Duration::from_secs(1));
                }
            }
        }
        Kind::Telegram => {
            // Skip what's waiting from before: start after the latest update.
            let mut offset = get_json(&format!("{}/bot{token}/getUpdates?offset=-1", tg()), None, 15)
                .ok()
                .and_then(|v| v["result"][0]["update_id"].as_i64())
                .map(|u| u + 1)
                .unwrap_or(0);
            set_status(c.id, true, "已連上，等待新訊息", 0);
            while alive(gen) {
                let url = format!("{}/bot{token}/getUpdates?offset={offset}&timeout={TELEGRAM_WAIT_SECS}", tg());
                match get_json(&url, None, TELEGRAM_WAIT_SECS + 10) {
                    Ok(v) => {
                        let (next, items) = telegram_new(&v, offset, &c.target);
                        offset = next;
                        if !alive(gen) {
                            break;
                        }
                        for (from, text, chat, mid) in items {
                            deliver(from, text, chat, mid);
                        }
                    }
                    Err(e) => {
                        set_status(c.id, false, &e, 0);
                        std::thread::sleep(Duration::from_secs(5));
                    }
                }
            }
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn slack_skips_bots_and_old_messages() {
        let v = json!({"ok": true, "messages": [
            {"ts": "1700000003.000", "user": "UBOT", "text": "my own"},
            {"ts": "1700000002.500", "bot_id": "B1", "text": "other bot"},
            {"ts": "1700000002.000", "user": "U1", "text": "<@UBOT> 幫我查天氣", "thread_ts": "1700000001.000"},
            {"ts": "1700000001.500", "user": "U2", "text": "嗨"},
            {"ts": "1700000000.000", "user": "U3", "text": "old"}
        ]});
        let (cur, items) = slack_new(&v, "1700000001.000", "UBOT");
        assert_eq!(cur, "1700000003.000");
        assert_eq!(items, vec![("U2".into(), "嗨".into(), "1700000001.500".into()), ("U1".into(), "幫我查天氣".into(), "1700000001.000".into())]);
    }

    #[test]
    fn discord_orders_and_skips_bots() {
        let v = json!([
            {"id": "30", "content": "second", "author": {"username": "amy"}},
            {"id": "25", "content": "bot", "author": {"username": "b", "bot": true}},
            {"id": "20", "content": "first", "author": {"username": "bob", "global_name": "Bob"}}
        ]);
        let (cur, items) = discord_new(&v, "10");
        assert_eq!(cur, "30");
        assert_eq!(items.iter().map(|i| i.1.as_str()).collect::<Vec<_>>(), vec!["first", "second"]);
        assert_eq!(items[0].0, "Bob");
    }

    #[test]
    fn telegram_filters_chat() {
        let v = json!({"ok": true, "result": [
            {"update_id": 7, "message": {"message_id": 1, "chat": {"id": 42}, "from": {"first_name": "小明"}, "text": "hi"}},
            {"update_id": 8, "message": {"message_id": 2, "chat": {"id": 99}, "from": {"first_name": "x"}, "text": "no"}}
        ]});
        let (next, items) = telegram_new(&v, 0, "42");
        assert_eq!(next, 9);
        assert_eq!(items, vec![("小明".into(), "hi".into(), "42".into(), "1".into())]);
        assert_eq!(telegram_new(&v, 0, "").1.len(), 2);
    }

    /// A whole round trip against a stand-in Telegram server: a message arrives, the answer goes back.
    #[test]
    fn telegram_round_trip() {
        use std::io::{Read, Write};
        let server = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
        let port = server.local_addr().unwrap().port();
        std::env::set_var("ANYBOT_TELEGRAM_API", format!("http://127.0.0.1:{port}"));
        let sent = Arc::new(Mutex::new(Vec::<String>::new()));
        let s2 = sent.clone();
        std::thread::spawn(move || {
            let mut served = 0;
            for conn in server.incoming().flatten() {
                let mut c = conn;
                // Read the whole request: head, then as much body as Content-Length says.
                c.set_read_timeout(Some(Duration::from_secs(3))).ok();
                let mut got = vec![];
                let mut buf = vec![0u8; 16384];
                loop {
                    let n = c.read(&mut buf).unwrap_or(0);
                    got.extend_from_slice(&buf[..n]);
                    let text = String::from_utf8_lossy(&got).to_string();
                    if let Some(h) = text.find("\r\n\r\n") {
                        let len = text[..h].lines().find_map(|l| l.to_lowercase().strip_prefix("content-length:").map(|v| v.trim().parse::<usize>().unwrap_or(0))).unwrap_or(0);
                        if got.len() >= h + 4 + len {
                            break;
                        }
                    }
                    if n == 0 {
                        break;
                    }
                }
                let req = String::from_utf8_lossy(&got).to_string();
                let body = if req.contains("offset=-1") {
                    r#"{"ok":true,"result":[{"update_id":100,"message":{"message_id":1,"chat":{"id":42},"text":"old"}}]}"#.to_string()
                } else if req.contains("getUpdates") {
                    served += 1;
                    if served == 1 {
                        r#"{"ok":true,"result":[{"update_id":101,"message":{"message_id":5,"chat":{"id":42},"from":{"first_name":"小明"},"text":"明天天氣？"}}]}"#.to_string()
                    } else {
                        std::thread::sleep(Duration::from_millis(300));
                        r#"{"ok":true,"result":[]}"#.to_string()
                    }
                } else {
                    s2.lock().unwrap().push(req.clone());
                    r#"{"ok":true,"result":{}}"#.to_string()
                };
                let _ = c.write_all(format!("HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}", body.len()).as_bytes());
            }
        });
        let (tx, rx) = std::sync::mpsc::channel();
        let ch = Channel { id: 1, kind: Kind::Telegram, name: "測試".into(), bot_id: 7, token: "T0K".into(), target: String::new(), enabled: true, client_id: String::new(), tenant: String::new() };
        restart(vec![ch], tx, Arc::new(|| {}));
        let got = rx.recv_timeout(Duration::from_secs(10)).expect("message arrives");
        GENERATION.fetch_add(1, Ordering::SeqCst);
        assert_eq!((got.bot_id, got.from.as_str(), got.text.as_str(), got.reply.target.as_str(), got.reply.thread.as_str()), (7, "小明", "明天天氣？", "42", "5"));
        reply(&got.reply, "明天晴天").unwrap();
        let reqs = sent.lock().unwrap().clone();
        assert!(reqs.iter().any(|r| r.contains("/botT0K/sendMessage") && r.contains("明天晴天") && r.contains("\"reply_to_message_id\":5")), "{reqs:?}");
        std::env::remove_var("ANYBOT_TELEGRAM_API");
    }

    #[test]
    fn email_addresses_are_found() {
        assert_eq!(address_of("王小明 <ming@example.com>"), "ming@example.com");
        assert_eq!(address_of("ming@example.com"), "ming@example.com");
    }

    #[test]
    fn long_answers_are_split() {
        let text = "a\n".repeat(3000);
        let parts = split(&text, DISCORD_MAX);
        assert!(parts.len() >= 3 && parts.iter().all(|p| p.chars().count() <= DISCORD_MAX));
        assert_eq!(parts.concat(), text);
    }
}