chore: bump version to 1.3.0

AGENTS.md

@@ -4,7 +4,7 @@
 
 Telegram bot (teloxide) that turns post links from X/Twitter, Pixiv, and Bluesky into media messages (images, video, GIF) with the post's title, author, and tags. It supports batch media splitting, retry with persistence, inline queries, forward-channel rebinding with caption templates, and Pixiv ugoira→MP4 transcoding. README and user-facing strings are in Chinese. The project is a Rust port of a Python predecessor (see `queue.rs` comments referencing `utils/task_queue.py`).
 
-Two-crate Cargo workspace (both v1.2.2, edition 2024, resolver 3):
+Two-crate Cargo workspace (both v1.3.0, edition 2024, resolver 3):
 
 - **`crates/x-media`** — library that fetches and normalizes media from the three sites. Pure, no Telegram knowledge.
 - **`crates/xmedia-bot`** — the bot binary: teloxide dispatcher, SQLite-backed chat state, persistent task queue.
@@ -20,6 +20,8 @@ Telegram update → Dispatcher (polling or axum webhook) → dptree branches
 
 Message flow: `message_handler` extracts URLs (from `url`/`text_link` entities, text + caption, deduped) → `x_media::site::fetch(url)` → `Fetched` → builds a `Task` → `send::send_media_sequence` (media groups ≤ 9, caption on first item) or `send::send_animation`. On Telegram URL-fetch failure or size error (`send_batch_via_upload`): download via `x_media::site::download_media` to a temp file (≤ 10 MiB), sniff magic bytes (`sniff_ext`), upload via multipart; oversized items fall back to `fallback_url`. On failure: `enqueue_retry` persists resume-state `Task` into the SQLite queue → workers lease (120 s lock TTL) → retry with exponential backoff (≤ 30 s, `MAX_RETRIES = 2`) → dead-letter → `notify_failure`. Success → `post_send_actions`: edit-before-forward prompt with inline buttons, or `copy_messages` to the bound forward channel.
 
+Debug command: `/test <url>` runs the same `x_media::site::fetch` and replies with `test_parse_report` (`handlers/commands.rs`) — site id, normalized cache key, source URL, title/author/tags, sensitive flag, caption and the media list — nothing is sent, cached or forwarded; the report is capped at 4000 chars (Telegram's 4096 plain-text limit). It uses a custom `parse_test_arg` parser (whole remainder, trimmed) because teloxide's built-in `split` parser takes exactly one space-separated token.
+
 The `x-media` library: `site::fetch(url)` dispatches through the `SITES` registry (per-site `impl Site`, in order twitter → bsky → pixiv) and returns `Ok(None)` for unmatched URLs. `Fetched { source_url, caption, title, media: Vec<Media>, sensitive, site_id, … }`; `caption_with(format)` substitutes `{url} {author} {author_url} {title} {tags}`.
 
 ## Key Directories
@@ -30,7 +32,7 @@ The `x-media` library: `site::fetch(url)` dispatches through the `SITES` registr
 | `crates/x-media/src/site/<twitter\|pixiv\|bsky>/` | One directory per site: `mod.rs` (re-exports), `interface.rs` (PATTERN, `enabled()`, `fetch_from_url()`, `cache_key`/`is_retryable`/`media_headers`, unit struct `<Name>Site` implementing `site::Site`, `From<SiteStruct> for Fetched`), `model.rs` (serde DTOs). Pixiv adds `api.rs` (auth + transport); twitter adds `auth.rs` (logged-in GraphQL `TweetDetail` fallback for NSFW tweets, gated on `TWITTER_AUTH_TOKEN`) |
 | `crates/xmedia-bot/src/main.rs` | Entry point: env/log init, command registration (`register_commands`), shared `send::BOT` force-init, queue worker start, site login validation (`site::validate_all`), 300 s edit-expiry sweep, dptree handler tree, webhook vs polling dispatch |
 | `crates/xmedia-bot/src/config.rs` | Manual env parsing into `Config` |
-| `crates/xmedia-bot/src/handlers.rs` | `Command` enum (teloxide `BotCommands`), message/inline/callback handlers, URL extraction, global statics; per-URL work flows through a bounded job channel (256) drained by `URL_WORKERS = 8` workers (`start_url_workers`) — backpressure instead of unbounded spawns (teloxide's per-chat workers are sequential — batch-forwards need concurrency) |
+| `crates/xmedia-bot/src/handlers/` | Handler modules: `mod.rs` (message entry point, `reply`, `log_key`), `commands.rs` (teloxide `BotCommands` enum + command executor, incl. the `/test <url>` parse-only debug command), `urls.rs` (URL extraction + bounded job channel (256) drained by `URL_WORKERS = 8` workers (`start_url_workers`) — backpressure instead of unbounded spawns; teloxide's per-chat workers are sequential — batch-forwards need concurrency), `inline.rs`/`callback.rs` (inline queries / edit-before-forward buttons), `statics.rs` (global statics) |
 | `crates/xmedia-bot/src/state.rs` | `ChatStore`: parking_lot `Mutex<HashMap>` cache + SQLite write-through (`chat_state` table) |
 | `crates/xmedia-bot/src/link_cache.rs` | `LinkCache`: SQLite-backed cache (`link_cache` table) of successfully sent posts — raw caption fields + Telegram `file_id`s; repeat links re-send locally (no fetch/upload), TTL + prune, invalidated on permanent send failure |
 | `crates/xmedia-bot/src/queue.rs` | `PersistentTaskQueue`: SQLite-backed queue (`tasks` table), `QUEUE_WORKERS = 4` concurrent workers (lease via `BEGIN IMMEDIATE` + `locked_until` TTL), retry→dead-letter, `Notify::notify_waiters` wakeup, `busy_timeout` on all connections |
@@ -67,7 +69,7 @@ Docker: `docker build -t tgxmb .` then `docker run --rm -d --name tgxmb --env-fi
 | File | Why it matters |
 |---|---|
 | `crates/xmedia-bot/src/main.rs` | Startup sequence, webhook vs polling, graceful shutdown (SIGINT via teloxide ctrlc / SIGTERM via `stop_token` for docker, → sweep stop → admin msg → queue stop) |
-| `crates/xmedia-bot/src/handlers.rs` | `CHAT_STORE`/`TASK_QUEUE`/`CONFIG` singletons (open `data/task_queue.db` **relative to CWD**); command dispatch; URL extraction; retry enqueue |
+| `crates/xmedia-bot/src/handlers/` | `statics.rs` = `CHAT_STORE`/`TASK_QUEUE`/`CONFIG` singletons (open `data/task_queue.db` **relative to CWD**); `commands.rs` = command dispatch (incl. the `/test <url>` parse-only debug command); `urls.rs` = URL extraction + retry enqueue; `inline.rs`/`callback.rs` = inline queries / edit-before-forward buttons |
 | `crates/xmedia-bot/src/send.rs` | Constants `MAX_MEDIA_GROUP = 9`; fallback chain; `classify_request_error`; download-and-reupload fallback triggered only by Telegram API errors (`is_media_fetch_failure` / `is_size_error`) |
 | `crates/xmedia-bot/src/photo.rs` | Pure-Rust photo processing (no ffmpeg): `png` (image-png) decode/encode + `zune-jpeg` decode + `fast_image_resize` Lanczos3 downscale + `jpeg-encoder`. Photos over Telegram's limits (width + height > 10000 px → `PHOTO_INVALID_DIMENSIONS`; bytes > 10 MiB) are decoded, downscaled keeping the format, PNG bit depth > 24 (RGBA 32-bit / 16-bit per channel) reduced to 24-bit RGB with alpha flattened white (≤24-bit untouched, never upconverted), and transcoded to JPEG only if still over the cap; memory budget guarded, otherwise the item's smaller fallback URL |
 | `crates/x-media/src/site/mod.rs` | Dispatcher, `Fetched`/`FetchError`, shared `CLIENT`, `download_media` (adds `Referer: https://www.pixiv.net/` for `pximg.net` hotlink protection) |

Cargo.lock

@@ -2925,7 +2925,7 @@ checksum = "1ffae5123b2d3fc086436f8834ae3ab053a283cfac8fe0a0b8eaae044768a4c4"
 
 [[package]]
 name = "x-media"
-version = "1.2.2"
+version = "1.3.0"
 dependencies = [
  "bytes",
  "dotenv",
@@ -2945,7 +2945,7 @@ dependencies = [
 
 [[package]]
 name = "xmedia-bot"
-version = "1.2.2"
+version = "1.3.0"
 dependencies = [
  "bytes",
  "dotenv",

README.en.md

@@ -113,6 +113,7 @@ Telegram only accepts ports 443/80/88/8443.
 | `/set_format <site> <format>` | Customize the caption format for one site. Sites: `twitter` / `bsky` / `pixiv`. Placeholders: `{url}` `{author}` `{author_url}` `{title}` `{tags}` |
 | `/clear_cache [link]` | Clear the link cache (admin only); with a link only that entry, otherwise everything |
 | `/bot_dict` | Show the current chat state (debugging) |
+| `/test <link>` | Debug: parse a link and report the parse result only (site, title, author, tags, media list) — no media is sent |
 
 Link processing works only in private chats; commands work in any chat.
 

README.md

@@ -113,6 +113,7 @@ Telegram 只接受 443/80/88/8443 端口。
 | `/set_format <站点> <格式>` | 自定义某站点的 caption 格式。站点：`twitter` / `bsky` / `pixiv`。占位符：`{url}` `{author}` `{author_url}` `{title}` `{tags}` |
 | `/clear_cache [链接]` | 清空链接缓存（仅管理员）；带链接只清该条，否则清空全部 |
 | `/bot_dict` | 查看当前聊天状态（调试用） |
+| `/test <链接>` | 调试：只解析链接并返回解析结果（站点、标题、作者、标签、媒体列表），不发送任何媒体 |
 
 链接处理仅限私聊；命令在任意聊天可用。
 

crates/x-media/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "x-media"
-version = "1.2.2"
+version = "1.3.0"
 edition = "2024"
 
 [dependencies]

crates/xmedia-bot/Cargo.toml

@@ -1,11 +1,11 @@
 [package]
 name = "xmedia-bot"
-version = "1.2.2"
+version = "1.3.0"
 edition = "2024"
 
 [dependencies]
 teloxide = { version = "0.17", default-features = false, features = ["webhooks-axum", "macros", "rustls", "ctrlc_handler"] }
-tokio = { version = "1.40", features = ["rt-multi-thread", "macros"] }
+tokio = { version = "1.40", features = ["rt-multi-thread", "macros", "time"] }
 serde = { version = "1", features = ["derive"] }
 serde_json = "1"
 log = "0.4"
@@ -23,3 +23,6 @@ zune-jpeg = "0.5"
 fast_image_resize = "6"
 jpeg-encoder = "0.7"
 x-media = { path = "../x-media" }
+
+[dev-dependencies]
+tokio = { version = "1.40", features = ["test-util"] }

crates/xmedia-bot/src/db.rs

@@ -134,6 +134,12 @@ fn rusqlite_error(e: std::io::Error) -> rusqlite::Error {
 /// Creates the `tasks`, `chat_state` and `link_cache` tables (idempotent).
 /// The three stores used to own their own schema; keeping it in one place
 /// means one initialization for the whole database file.
+///
+/// ⚠️ Schema-change reminder (deferred, see `docs/architecture-refactor.md`
+/// §5): this is a plain `CREATE TABLE IF NOT EXISTS` with no versioning.
+/// Before any column/table change that must migrate existing databases, land
+/// the `PRAGMA user_version` migration chain first (`MIGRATIONS: &[&str]` +
+/// `migrate(conn)`), then restructure this function.
 pub fn schema_init(conn: &Connection) -> rusqlite::Result<()> {
     conn.execute_batch(
         "CREATE TABLE IF NOT EXISTS tasks (id TEXT PRIMARY KEY, payload TEXT NOT NULL, \

crates/xmedia-bot/src/handlers/callback.rs

@@ -0,0 +1,130 @@
+//! Callback query handling: the edit-before-forward prompt's "forward" and
+//! "template|<name>" buttons.
+
+use super::urls::enqueue_retry;
+use super::{CHAT_STORE, CONFIG, TASK_QUEUE};
+use crate::send::{self, Task};
+use crate::state::unix_now;
+use teloxide::RequestError;
+use teloxide::prelude::*;
+use teloxide::types::{CallbackQuery, ChatId, MessageId, ParseMode};
+
+pub async fn callback_query_handler(bot: Bot, query: CallbackQuery) -> Result<(), RequestError> {
+    let callback_query_id = query.id;
+    let data = query.data.clone();
+    let Some(message) = &query.message else {
+        return respond(());
+    };
+    let chat_id = message.chat().id.0;
+    let prompt_message_id = message.id().0 as i64;
+    let ttl_secs = CONFIG.edit_message_ttl.as_secs() as i64;
+    let chat_data = CHAT_STORE.get(chat_id).await;
+    let edit = chat_data.edit_message.get(&prompt_message_id).cloned();
+    let Some(edit) = edit else {
+        log::debug!(
+            "callback from {}: no edit record for prompt {prompt_message_id}",
+            chat_id
+        );
+        bot.answer_callback_query(callback_query_id)
+            .text("Expired")
+            .await?;
+        return respond(());
+    };
+    // Lazy expiry: a stale record (past the TTL, not yet swept) is dropped.
+    if edit.created_at + ttl_secs <= unix_now() {
+        CHAT_STORE
+            .update(chat_id, |data| {
+                data.edit_message.remove(&prompt_message_id);
+            })
+            .await;
+        bot.answer_callback_query(callback_query_id)
+            .text("Expired")
+            .await?;
+        return respond(());
+    }
+
+    let Some(data) = data else {
+        return respond(());
+    };
+    log::info!(
+        "callback from {} on prompt {prompt_message_id}: {data}",
+        chat_id
+    );
+    if data == "forward" {
+        match chat_data.forward_channel_id {
+            Some(channel_id) => {
+                let forward_task = Task::ForwardMessages {
+                    from_chat_id: edit.chat_id,
+                    to_chat_id: channel_id,
+                    message_ids: edit.forward_message_ids.clone(),
+                    notify_chat_id: Some(chat_id),
+                    notify_message_id: Some(prompt_message_id),
+                };
+                match send::forward_messages(&bot, &forward_task).await {
+                    Ok(()) => {
+                        log::info!(
+                            "forwarded {} message(s) to channel {channel_id}",
+                            edit.forward_message_ids.len()
+                        );
+                        bot.answer_callback_query(callback_query_id)
+                            .text("✅ Forwarded")
+                            .await?;
+                        let _ = bot
+                            .delete_message(ChatId(chat_id), MessageId(prompt_message_id as i32))
+                            .await;
+                        CHAT_STORE
+                            .update(chat_id, |data| {
+                                data.edit_message.remove(&prompt_message_id);
+                            })
+                            .await;
+                    }
+                    Err(send::SendError::Retryable {
+                        delay_seconds,
+                        task,
+                    }) => {
+                        log::info!("forward queued for retry in {delay_seconds:.1}s");
+                        enqueue_retry(&TASK_QUEUE, task, delay_seconds).await;
+                        bot.answer_callback_query(callback_query_id)
+                            .text("Forward queued for retry.")
+                            .await?;
+                    }
+                    Err(send::SendError::Permanent { message, .. }) => {
+                        log::error!("forward failed permanently: {message}");
+                        bot.answer_callback_query(callback_query_id)
+                            .text(format!("Forward failed: {message}"))
+                            .await?;
+                    }
+                }
+            }
+            None => {
+                log::debug!("forward callback without a forward channel set");
+                bot.answer_callback_query(callback_query_id)
+                    .text("No forward channel set.")
+                    .await?;
+            }
+        }
+        return respond(());
+    }
+    if let Some(name) = data.strip_prefix("template|") {
+        if let Some(template_html) = chat_data.template.get(name).cloned()
+            && let Some(first_forward_id) = edit.forward_message_ids.first().copied()
+        {
+            // Raw template including the [] placeholder (Python parity).
+            let _ = bot
+                .edit_message_caption(ChatId(chat_id), MessageId(first_forward_id as i32))
+                .caption(template_html)
+                .parse_mode(ParseMode::Html)
+                .await;
+            CHAT_STORE
+                .update(chat_id, |data| {
+                    if let Some(entry) = data.edit_message.get_mut(&prompt_message_id) {
+                        entry.template = name.to_string();
+                    }
+                })
+                .await;
+            log::info!("template '{name}' applied to prompt {prompt_message_id}");
+        }
+        bot.answer_callback_query(callback_query_id).await?;
+    }
+    respond(())
+}

crates/xmedia-bot/src/handlers/commands.rs

@@ -0,0 +1,518 @@
+//! Bot command parsing, the `/`-command executor and `setMyCommands`
+//! registration. URL/inline/callback flows live in their own modules.
+
+use super::{CHAT_STORE, CONFIG, LINK_CACHE, log_key, reply};
+use teloxide::RequestError;
+use teloxide::prelude::*;
+use teloxide::types::{ChatId, Message, Recipient};
+use teloxide::utils::command::{BotCommands, ParseError};
+
+#[derive(BotCommands, Clone)]
+#[command(
+    rename_rule = "snake_case",
+    description = "Turn X/Pixiv/Bluesky links into media messages"
+)]
+pub(crate) enum Command {
+    #[command(description = "Get started")]
+    Start,
+    #[command(description = "Show command help")]
+    Help,
+    #[command(
+        description = "Set forward channel (@channel or ID)",
+        parse_with = "split"
+    )]
+    SetForwardChannel(String),
+    #[command(description = "Remove forward channel")]
+    RemoveForwardChannel,
+    #[command(description = "Toggle edit-before-forward")]
+    EditBeforeForward,
+    #[command(
+        description = "Reply with [] to save as template",
+        parse_with = "split"
+    )]
+    SetTemplate(String),
+    #[command(description = "Show chat state (debug)")]
+    BotDict,
+    #[command(description = "Set site caption format", parse_with = "split")]
+    SetFormat(String),
+    #[command(
+        description = "Clear link cache (admin; optional URL, else all)",
+        parse_with = "split"
+    )]
+    ClearCache(String),
+    #[command(
+        description = "Test link parsing (debug; no media sent)",
+        parse_with = parse_test_arg
+    )]
+    Test(String),
+}
+
+/// `/test` argument parser: the whole remainder after the command name,
+/// trimmed. The built-in `split` parser takes exactly one space-separated
+/// token and rejects the rest, so a URL followed by a trailing space (or
+/// pasted text) would silently fall through to the URL flow instead.
+fn parse_test_arg(s: String) -> Result<(String,), ParseError> {
+    Ok((s.trim().to_string(),))
+}
+
+enum SetForwardChannelError {
+    EmptyParameter,
+    NotChannel,
+    NotAdmin,
+    NotBotAdmin(RequestError),
+    NotBotCanPost,
+}
+
+async fn set_forward_channel_handler(
+    bot: &Bot,
+    message: &Message,
+    channel: String,
+) -> Result<i64, SetForwardChannelError> {
+    if channel.is_empty() {
+        return Err(SetForwardChannelError::EmptyParameter);
+    }
+    let channel = match channel.parse::<i64>() {
+        Ok(id) => Recipient::Id(ChatId(id)),
+        Err(_) => Recipient::ChannelUsername(channel),
+    };
+    if let Some(from) = &message.from {
+        log::info!(
+            "Set forward channel for {} ({}) to {}",
+            from.full_name(),
+            message.chat.id,
+            channel
+        );
+    }
+    let chat = match bot.get_chat(channel.clone()).await {
+        Err(e) => {
+            log::error!("Failed to get channel {}: {}", channel, e);
+            return Err(SetForwardChannelError::NotBotAdmin(e));
+        }
+        Ok(chat) => chat,
+    };
+    if !chat.is_channel() {
+        return Err(SetForwardChannelError::NotChannel);
+    }
+    let channel_id = chat.id.0;
+    // The sender must be a channel administrator. Compare against the
+    // sender's user id, NOT the chat id (they only coincide in private
+    // chats, so the old check broke group usage).
+    let Some(sender) = message.from.as_ref() else {
+        return Err(SetForwardChannelError::NotAdmin);
+    };
+    match bot.get_chat_administrators(channel.clone()).await {
+        Err(e) => {
+            log::error!("Failed to get channel administrators {}: {}", channel, e);
+            return Err(SetForwardChannelError::NotBotAdmin(e));
+        }
+        Ok(admins) => {
+            if !admins.iter().any(|admin| admin.user.id == sender.id) {
+                return Err(SetForwardChannelError::NotAdmin);
+            }
+            // The bot itself must be an admin that can post; a missing
+            // bot entry must not pass silently (copy would fail later).
+            let bot_id = match bot.get_me().await {
+                Ok(me) => me.user.id,
+                Err(e) => return Err(SetForwardChannelError::NotBotAdmin(e)),
+            };
+            let bot_ok = admins
+                .iter()
+                .any(|admin| admin.user.id == bot_id && admin.can_post_messages());
+            if !bot_ok {
+                return Err(SetForwardChannelError::NotBotCanPost);
+            }
+        }
+    }
+    Ok(channel_id)
+}
+
+pub(crate) async fn execute_command(
+    bot: &Bot,
+    message: &Message,
+    command: Command,
+) -> Result<(), RequestError> {
+    match command {
+        Command::Start => {
+            bot.send_message(message.chat.id, "Hello!").await?;
+        }
+        Command::Help => {
+            bot.send_message(message.chat.id, Command::descriptions().to_string())
+                .await?;
+        }
+        Command::SetForwardChannel(channel) => {
+            let result = match set_forward_channel_handler(bot, message, channel).await {
+                Ok(channel_id) => {
+                    CHAT_STORE
+                        .update(message.chat.id.0, |data| {
+                            data.forward_channel_id = Some(channel_id);
+                        })
+                        .await;
+                    "Add successfully.".to_string()
+                }
+                Err(SetForwardChannelError::EmptyParameter) => {
+                    "Receive empty parameter.\nYou should enter a channel id or username"
+                        .to_string()
+                }
+                Err(SetForwardChannelError::NotChannel) => {
+                    "Given id / username is not a channel".to_string()
+                }
+                Err(SetForwardChannelError::NotAdmin) => {
+                    "You are not an administrator of the channel".to_string()
+                }
+                Err(SetForwardChannelError::NotBotAdmin(e)) => {
+                    e.to_string() + "\nPlease add the bot as an admin to the channel"
+                }
+                Err(SetForwardChannelError::NotBotCanPost) => {
+                    "Bot can't post messages to the channel".to_string()
+                }
+            };
+            reply(bot, message.chat.id.0, message.id, result).await?;
+        }
+        Command::RemoveForwardChannel => {
+            let chat_id = message.chat.id.0;
+            let text = CHAT_STORE
+                .update(chat_id, |data| {
+                    if data.forward_channel_id.is_some() {
+                        data.forward_channel_id = None;
+                        "Remove successfully.".to_string()
+                    } else {
+                        "No channel to remove.".to_string()
+                    }
+                })
+                .await;
+            reply(bot, message.chat.id.0, message.id, text).await?;
+        }
+        Command::EditBeforeForward => {
+            let chat_id = message.chat.id.0;
+            let text = CHAT_STORE
+                .update(chat_id, |data| {
+                    if data.forward_channel_id.is_none() {
+                        "Please enable forward channel first.".to_string()
+                    } else if data.edit_before_forward {
+                        data.edit_before_forward = false;
+                        data.edit_message.clear();
+                        "Disable edit before forward.".to_string()
+                    } else {
+                        data.edit_before_forward = true;
+                        "Enable edit before forward.".to_string()
+                    }
+                })
+                .await;
+            reply(bot, message.chat.id.0, message.id, text).await?;
+        }
+        Command::SetTemplate(name) => {
+            let chat_id = message.chat.id.0;
+            let text = match message.reply_to_message() {
+                None => "Please reply to a message to set as template.".to_string(),
+                Some(reply) => {
+                    let reply_text = reply.text().unwrap_or_default();
+                    if !reply_text.contains("[]") {
+                        "Please reply to a message with [] to set as template.".to_string()
+                    } else if name.is_empty() {
+                        "Please provide a name for the template.".to_string()
+                    } else {
+                        CHAT_STORE
+                            .update(chat_id, |data| {
+                                data.template.insert(
+                                    name,
+                                    html_escape::encode_text(reply_text).into_owned(),
+                                );
+                            })
+                            .await;
+                        "Template set.".to_string()
+                    }
+                }
+            };
+            reply(bot, message.chat.id.0, message.id, text).await?;
+        }
+        Command::BotDict => {
+            let chat_data = CHAT_STORE.get(message.chat.id.0).await;
+            let debug = format!("{chat_data:?}");
+            let text = html_escape::encode_text(&debug).into_owned();
+            reply(bot, message.chat.id.0, message.id, text).await?;
+        }
+        Command::SetFormat(arg) => {
+            let chat_id = message.chat.id.0;
+            let (site, format) = match arg.split_once(char::is_whitespace) {
+                Some((site, format)) if !format.trim().is_empty() => {
+                    (site.trim(), format.trim().to_string())
+                }
+                _ => {
+                    reply(
+                        bot,
+                        message.chat.id.0,
+                        message.id,
+                        "Usage: /set_format <site> <format>",
+                    )
+                    .await?;
+                    return Ok(());
+                }
+            };
+            if !x_media::site::site_ids().contains(&site) {
+                reply(
+                    bot,
+                    message.chat.id.0,
+                    message.id,
+                    "Unknown site. Use twitter, bsky or pixiv.",
+                )
+                .await?;
+                return Ok(());
+            }
+            CHAT_STORE
+                .update(chat_id, |data| {
+                    data.message_format.insert(site.to_string(), format);
+                })
+                .await;
+            reply(bot, message.chat.id.0, message.id, "Format set.").await?;
+        }
+        Command::ClearCache(arg) => {
+            let sender_id = message
+                .from
+                .as_ref()
+                .map(|user| user.id.0 as i64)
+                .unwrap_or(-1);
+            if !CONFIG.admin_ids.contains(&sender_id) {
+                reply(bot, message.chat.id.0, message.id, "Admin only.").await?;
+                return Ok(());
+            }
+            let arg = arg.trim();
+            if arg.is_empty() {
+                let removed = LINK_CACHE.clear(None).await;
+                log::info!("cache cleared by {sender_id}: {removed} entries");
+                reply(
+                    bot,
+                    message.chat.id.0,
+                    message.id,
+                    format!("Cleared {removed} cached entr{}.", plural(removed)),
+                )
+                .await?;
+            } else {
+                let key = match x_media::site::cache_key(arg) {
+                    Some(key) => key,
+                    None => {
+                        reply(
+                            bot,
+                            message.chat.id.0,
+                            message.id,
+                            "Unrecognized link. Use a twitter/x, pixiv or bsky post URL.",
+                        )
+                        .await?;
+                        return Ok(());
+                    }
+                };
+                let removed = LINK_CACHE.clear(Some(&key)).await;
+                log::info!("cache entry cleared by {sender_id}: {key} ({removed} rows)");
+                reply(
+                    bot,
+                    message.chat.id.0,
+                    message.id,
+                    format!(
+                        "Cleared cache for {arg} ({} entr{}).",
+                        removed,
+                        plural(removed)
+                    ),
+                )
+                .await?;
+            }
+        }
+        Command::Test(arg) => {
+            let url = arg.trim();
+            if url.is_empty() {
+                reply(
+                    bot,
+                    message.chat.id.0,
+                    message.id,
+                    "Usage: /test <post url>",
+                )
+                .await?;
+                return Ok(());
+            }
+            // Debug tool: report the parse result only — nothing is sent,
+            // cached or forwarded. Info level echoes the normalized key
+            // (never the raw URL) per the logging convention.
+            log::info!("test: parsing [key={}]", log_key(url));
+            match x_media::site::fetch(url).await {
+                Ok(None) => {
+                    reply(
+                        bot,
+                        message.chat.id.0,
+                        message.id,
+                        "No enabled site matches this link (twitter/x, pixiv or bsky).",
+                    )
+                    .await?;
+                }
+                Err(e) => {
+                    reply(
+                        bot,
+                        message.chat.id.0,
+                        message.id,
+                        format!("Fetch failed: {e}"),
+                    )
+                    .await?;
+                }
+                Ok(Some(fetched)) => {
+                    let report = test_parse_report(
+                        url,
+                        fetched.site_name(),
+                        &fetched.source_url,
+                        &fetched.title,
+                        fetched.render_fields(),
+                        fetched.sensitive,
+                        &fetched.caption,
+                        &fetched.media,
+                    );
+                    reply(bot, message.chat.id.0, message.id, report).await?;
+                }
+            }
+        }
+    }
+    Ok(())
+}
+
+/// `"y"` for one, `"ies"` for anything else — "1 entry" / "2 entries".
+fn plural(n: usize) -> &'static str {
+    if n == 1 { "y" } else { "ies" }
+}
+
+/// Registers the bot's command list with Telegram so clients show it in the
+/// `/` menu (Bot API `setMyCommands`).
+pub async fn register_commands(bot: &Bot) -> Result<(), RequestError> {
+    let commands = Command::bot_commands();
+    bot.set_my_commands(commands.clone()).await?;
+    log::info!("registered {} commands", commands.len());
+    Ok(())
+}
+
+/// Telegram's plain-text message limit is 4096 chars; the report stays under
+/// it even for very large threads (many media lines + a long caption).
+const MAX_TEST_REPORT_CHARS: usize = 4000;
+
+/// Builds the plain-text report for the `/test` command: what the parser
+/// produced for a link (site, canonical URL, title/author/tags, caption and
+/// the media list) — no media is sent and nothing is cached or forwarded.
+/// Fields are passed individually so the formatter stays a pure function
+/// testable without constructing a `Fetched` (its render fields are
+/// `pub(crate)` to the x-media crate).
+#[allow(clippy::too_many_arguments)]
+fn test_parse_report(
+    url: &str,
+    site_id: &str,
+    source_url: &str,
+    title: &str,
+    render: Option<(&str, &str, &str, &str)>,
+    sensitive: bool,
+    caption: &str,
+    media: &[x_media::media::Media],
+) -> String {
+    let mut lines = vec![
+        format!("Parse result for {url}"),
+        format!("site: {site_id}"),
+        format!(
+            "key: {}",
+            x_media::site::cache_key(url).unwrap_or_else(|| "<unsupported>".to_string())
+        ),
+    ];
+    lines.push(format!("source_url: {source_url}"));
+    lines.push(format!("title: {title}"));
+    if let Some((author, author_url, _title, tags)) = render {
+        lines.push(format!("author: {author}"));
+        lines.push(format!("author_url: {author_url}"));
+        lines.push(format!("tags: {tags}"));
+    }
+    lines.push(format!("sensitive: {sensitive}"));
+    lines.push(format!(
+        "caption: {}",
+        x_media::site::truncate_caption(caption)
+    ));
+    lines.push(format!("media ({}):", media.len()));
+    for (i, item) in media.iter().enumerate() {
+        let kind = match item {
+            x_media::media::Media::Illustration { .. } => "image",
+            x_media::media::Media::Video { .. } => "video",
+            x_media::media::Media::Animated { .. } => "gif",
+        };
+        lines.push(format!("  {}. {kind}: {}", i + 1, item.url()));
+    }
+    let mut out = lines.join(
+        "
+",
+    );
+    if out.chars().count() > MAX_TEST_REPORT_CHARS {
+        let end = out.floor_char_boundary(MAX_TEST_REPORT_CHARS - 1);
+        out = format!("{}…", &out[..end]);
+    }
+    out
+}
+
+#[cfg(test)]
+mod tests {
+    use super::{MAX_TEST_REPORT_CHARS, test_parse_report};
+    use x_media::media::Media;
+
+    #[test]
+    fn test_parse_report_renders_fields_and_media() {
+        let media = vec![
+            Media::Illustration {
+                title: None,
+                url: "https://cdn.example/1.jpg".into(),
+                thumbnail_url: None,
+                fallback_url: None,
+            },
+            Media::Video {
+                title: None,
+                url: "https://cdn.example/2.mp4".into(),
+                thumbnail_url: "https://cdn.example/2.jpg".into(),
+            },
+        ];
+        let report = test_parse_report(
+            "https://x.com/u/status/1",
+            "twitter",
+            "https://x.com/u/status/1",
+            "My title",
+            Some(("Author", "https://x.com/u", "My title", "tag1 tag2")),
+            false,
+            "<a href=\"https://x.com/u\">Author</a> · My title",
+            &media,
+        );
+        assert!(report.contains("site: twitter"), "{report}");
+        assert!(report.contains("key: twitter:1"), "{report}");
+        assert!(report.contains("title: My title"), "{report}");
+        assert!(report.contains("author: Author"), "{report}");
+        assert!(report.contains("author_url: https://x.com/u"), "{report}");
+        assert!(report.contains("tags: tag1 tag2"), "{report}");
+        assert!(report.contains("sensitive: false"), "{report}");
+        assert!(report.contains("media (2):"), "{report}");
+        assert!(
+            report.contains("1. image: https://cdn.example/1.jpg"),
+            "{report}"
+        );
+        assert!(
+            report.contains("2. video: https://cdn.example/2.mp4"),
+            "{report}"
+        );
+    }
+
+    #[test]
+    fn test_parse_report_without_render_data_and_no_media() {
+        let report = test_parse_report("u", "pixiv", "s", "t", None, true, "c", &[]);
+        assert!(!report.contains("author:"), "{report}");
+        assert!(report.contains("sensitive: true"), "{report}");
+        assert!(report.contains("media (0):"), "{report}");
+    }
+
+    #[test]
+    fn test_parse_report_is_capped() {
+        // 200 media lines ≈ 8 KB, comfortably over the cap.
+        let media: Vec<Media> = (0..200)
+            .map(|i| Media::Illustration {
+                title: None,
+                url: format!("https://cdn.example/{i}.jpg"),
+                thumbnail_url: None,
+                fallback_url: None,
+            })
+            .collect();
+        let report = test_parse_report("u", "twitter", "s", "t", None, false, "c", &media);
+        assert!(report.chars().count() <= MAX_TEST_REPORT_CHARS, "{report}");
+        assert!(report.ends_with('…'), "{report}");
+    }
+}

crates/xmedia-bot/src/handlers/inline.rs

@@ -0,0 +1,161 @@
+//! Inline query handling with a keystroke debounce: only a query stable for
+//! [`INLINE_DEBOUNCE`] triggers a fetch, and repeats are served by Telegram's
+//! inline cache instead of re-fetching.
+
+use super::log_key;
+use std::sync::LazyLock;
+use teloxide::RequestError;
+use teloxide::prelude::*;
+use teloxide::types::{
+    InlineQuery, InlineQueryResult, InlineQueryResultMpeg4Gif, InlineQueryResultPhoto,
+    InlineQueryResultVideo, ParseMode,
+};
+use x_media::media::Media;
+
+/// Debounce window for inline queries: Telegram fires an inline query on
+/// every keystroke, and each prefix of a pasted URL (e.g. `.../status/12`,
+/// `.../status/123`, ...) already matches the site patterns. Without a
+/// debounce every keystroke triggers a fetch (3 attempts!) of a half-typed
+/// post id. Only answer once the query has been stable for this long.
+const INLINE_DEBOUNCE: std::time::Duration = std::time::Duration::from_millis(800);
+
+/// Last seen inline query and whether it was already answered. Guards the
+/// debounce timer: a repeat of an answered query is served by Telegram's
+/// inline cache (see `cache_time`), not by another fetch.
+struct InlineDebounceState {
+    query: String,
+    answered: bool,
+}
+
+static INLINE_DEBOUNCE_STATE: LazyLock<parking_lot::Mutex<Option<InlineDebounceState>>> =
+    LazyLock::new(|| parking_lot::Mutex::new(None));
+
+pub async fn inline_query_handler(bot: Bot, query: InlineQuery) -> Result<(), RequestError> {
+    if query.query.is_empty() {
+        return respond(());
+    }
+    // Only run a fetch for something that is actually a supported post URL.
+    if x_media::site::cache_key(&query.query).is_none() {
+        return respond(());
+    }
+    // Debounce: record the query and answer only after it has been stable for
+    // INLINE_DEBOUNCE (the timer below). An already-answered repeat of the
+    // same query is left to Telegram's inline cache instead of re-fetching.
+    {
+        let mut state = INLINE_DEBOUNCE_STATE.lock();
+        if let Some(prev) = state.as_ref()
+            && prev.query == query.query
+            && prev.answered
+        {
+            return respond(());
+        }
+        *state = Some(InlineDebounceState {
+            query: query.query.clone(),
+            answered: false,
+        });
+    }
+    let query_text = query.query.clone();
+    tokio::spawn(async move {
+        tokio::time::sleep(INLINE_DEBOUNCE).await;
+        // Only the last query of a typing burst survives: earlier timers see
+        // the query changed and give up without answering.
+        {
+            let mut state = INLINE_DEBOUNCE_STATE.lock();
+            let Some(state) = state.as_mut() else {
+                return;
+            };
+            if state.query != query_text || state.answered {
+                return;
+            }
+            // Claim the answer so a repeat of the same query cannot start a
+            // second fetch; reset below when no answer was produced.
+            state.answered = true;
+        }
+        match answer_inline_query(bot, query).await {
+            Ok(true) => {}
+            // No results produced (or nothing to answer): let a repeat of the
+            // same query retry the fetch.
+            Ok(false) | Err(_) => {
+                let mut state = INLINE_DEBOUNCE_STATE.lock();
+                if let Some(state) = state.as_mut()
+                    && state.query == query_text
+                {
+                    state.answered = false;
+                }
+            }
+        }
+    });
+    respond(())
+}
+
+/// Fetches the post behind an inline query and answers it. The caller has
+/// already applied the debounce. Returns `true` when an answer was sent.
+async fn answer_inline_query(bot: Bot, query: InlineQuery) -> Result<bool, RequestError> {
+    log::debug!(
+        "inline query: {} [key={}]",
+        query.query,
+        log_key(&query.query)
+    );
+    match x_media::site::fetch(&query.query).await {
+        Ok(Some(fetched)) => {
+            let mut results: Vec<InlineQueryResult> = Vec::new();
+            // Inline results have the same 1024-char caption limit as regular
+            // messages; truncate once here for all items.
+            let caption = x_media::site::truncate_caption(&fetched.caption);
+            for (i, media) in fetched.media.iter().enumerate() {
+                let id = format!("{i}");
+                let Some(url) = url::Url::parse(media.url()).ok() else {
+                    continue;
+                };
+                let thumbnail = media
+                    .thumbnail_url()
+                    .and_then(|t| url::Url::parse(t).ok())
+                    .unwrap_or_else(|| url.clone());
+                let caption = caption.clone();
+                let result = match media {
+                    Media::Illustration { .. } => {
+                        // Inline photo results have their own (smaller) size
+                        // cap; use the reduced variant when one exists.
+                        let photo_url = media
+                            .smaller_url()
+                            .and_then(|u| url::Url::parse(u).ok())
+                            .unwrap_or_else(|| url.clone());
+                        InlineQueryResult::Photo(
+                            InlineQueryResultPhoto::new(id, photo_url, thumbnail)
+                                .caption(caption)
+                                .parse_mode(ParseMode::Html),
+                        )
+                    }
+                    Media::Video { .. } => InlineQueryResult::Video(
+                        InlineQueryResultVideo::new(
+                            id,
+                            url,
+                            "video/mp4".parse().expect("valid mime"),
+                            thumbnail,
+                            fetched.title.clone(),
+                        )
+                        .caption(caption)
+                        .parse_mode(ParseMode::Html),
+                    ),
+                    Media::Animated { .. } => InlineQueryResult::Mpeg4Gif(
+                        InlineQueryResultMpeg4Gif::new(id, url, thumbnail)
+                            .caption(caption)
+                            .parse_mode(ParseMode::Html),
+                    ),
+                };
+                results.push(result);
+            }
+            if !results.is_empty() {
+                // Explicit cache window: repeats of the same query within 5
+                // minutes are served by Telegram without hitting the bot.
+                bot.answer_inline_query(query.id, results)
+                    .cache_time(300)
+                    .await?;
+                return Ok(true);
+            }
+        }
+        Ok(None) => {}
+        Err(e) => log::error!("inline fetch {}: {e}", query.query),
+    }
+    Ok(false)
+}

crates/xmedia-bot/src/handlers/mod.rs

@@ -0,0 +1,147 @@
+//! Update handlers and the per-URL media pipeline.
+//!
+//! Split into per-concern modules: [`commands`] (the `/`-command executor),
+//! [`urls`] (URL extraction + the bounded worker pool + send dispatch),
+//! [`inline`] (debounced inline queries), [`callback`] (edit-before-forward
+//! buttons) and [`statics`] (the shared process-wide stores). This module
+//! holds the message entry point and the helpers the others share.
+
+mod callback;
+mod commands;
+mod inline;
+mod statics;
+mod urls;
+
+pub use callback::callback_query_handler;
+pub use commands::register_commands;
+pub use inline::inline_query_handler;
+pub use statics::{CHAT_STORE, CONFIG, LINK_CACHE, TASK_QUEUE};
+pub use urls::{start_url_workers, stop_url_workers};
+
+use crate::media_sender::MediaSender;
+use commands::{Command, execute_command};
+use teloxide::RequestError;
+use teloxide::prelude::*;
+use teloxide::types::{ChatId, ChatKind, Message, MessageId, ParseMode};
+use teloxide::utils::command::BotCommands;
+use urls::{URL_JOBS, extract_urls};
+
+/// Reply to a message by id, keeping the reply decoration even if the
+/// original was already deleted.
+pub(crate) async fn reply<T>(
+    sender: &dyn MediaSender,
+    chat_id: i64,
+    reply_to: MessageId,
+    text: T,
+) -> Result<Message, RequestError>
+where
+    T: Into<String>,
+{
+    sender
+        .send_message(ChatId(chat_id), text.into(), Some(reply_to), None)
+        .await
+}
+
+/// Log prefix tying the whole lifecycle of one link (fetch → send → cache →
+/// forward) together: the normalized cache key (`twitter:123…`, `pixiv:123`,
+/// `bsky:handle/rkey`) instead of the raw URL, so logs stay short and do not
+/// echo full user-submitted URLs at info level.
+pub fn log_key(url: &str) -> String {
+    x_media::site::cache_key(url).unwrap_or_else(|| "<unsupported>".to_string())
+}
+
+/// Edit-before-forward: a reply to the prompt swaps the caption of the first
+/// forwarded message. Returns true when the message was consumed as an edit.
+async fn edit_message_handler(bot: &Bot, message: &Message) -> bool {
+    let Some(reply) = message.reply_to_message() else {
+        return false;
+    };
+    let chat_id = message.chat.id.0;
+    let Some(text) = message.text() else {
+        return false;
+    };
+    let chat_data = CHAT_STORE.get(chat_id).await;
+    let Some(edit) = chat_data.edit_message.get(&(reply.id.0 as i64)) else {
+        return false;
+    };
+    let Some(first_forward_id) = edit.forward_message_ids.first() else {
+        return false;
+    };
+    let link = format!(
+        "<a href=\"{0}\">{1}</a>",
+        html_escape::encode_double_quoted_attribute(&edit.url),
+        html_escape::encode_text(text)
+    );
+    let new_text = if edit.template.is_empty() {
+        link
+    } else {
+        chat_data
+            .template
+            .get(&edit.template)
+            .map(|template| template.replace("[]", &link))
+            .unwrap_or(link)
+    };
+    let result = bot
+        .edit_message_caption(ChatId(chat_id), MessageId(*first_forward_id as i32))
+        .caption(new_text)
+        .parse_mode(ParseMode::Html)
+        .await;
+    match result {
+        Ok(_) => log::info!(
+            "edit-before-forward: caption swapped on message {first_forward_id} for prompt {}",
+            reply.id.0
+        ),
+        Err(e) => log::error!("edit_message_caption failed: {e}"),
+    }
+    true
+}
+
+pub async fn message_handler(bot: Bot, message: Message) -> Result<(), RequestError> {
+    let is_private = matches!(message.chat.kind, ChatKind::Private(_));
+    let sender = message
+        .from
+        .as_ref()
+        .map(|from| from.full_name())
+        .unwrap_or_else(|| "unknown".to_string());
+    let text_preview = message
+        .text()
+        .map(|t| {
+            let end = t.floor_char_boundary(120.min(t.len()));
+            &t[..end]
+        })
+        .unwrap_or("<no text>");
+    // Per-request detail: debug only (message text is user data).
+    log::debug!(
+        "message from {sender} in {} (private={is_private}): {text_preview}",
+        message.chat.id
+    );
+    // URL/edit flows only run in private chats; commands run in any chat.
+    if is_private && edit_message_handler(&bot, &message).await {
+        return respond(());
+    }
+    if let Some(text) = message.text()
+        && let Ok(command) = Command::parse(text, "")
+    {
+        log::debug!("command from {}: {text_preview}", message.chat.id);
+        execute_command(&bot, &message, command).await?;
+        return respond(());
+    }
+    if is_private {
+        let urls = extract_urls(&message);
+        if !urls.is_empty() {
+            // Debug only, and echo the normalized keys instead of the raw URLs.
+            let keys: Vec<String> = urls.iter().map(|u| log_key(u)).collect();
+            log::debug!("extracted {} URL(s): {keys:?}", urls.len());
+        }
+        for url in urls {
+            // Clone out of the lock: the parking_lot guard is !Send and must
+            // not be held across the await below.
+            let Some(tx) = URL_JOBS.lock().clone() else {
+                log::warn!("url workers not started; dropping link");
+                break;
+            };
+            let _ = tx.send((message.clone(), url)).await;
+        }
+    }
+    respond(())
+}

crates/xmedia-bot/src/handlers/statics.rs

@@ -0,0 +1,22 @@
+//! Process-wide singletons shared by the handler modules: the one SQLite
+//! pool (and the three stores built on it) plus the configuration.
+
+use crate::config::Config;
+use crate::db::{self};
+use crate::link_cache::LinkCache;
+use crate::queue::PersistentTaskQueue;
+use crate::state::ChatStore;
+use std::sync::{Arc, LazyLock};
+
+/// One shared SQLite pool for the three stores (chat state, task queue, link
+/// cache): a single pool bounds concurrent DB work on `data/task_queue.db`
+/// instead of three independent pools competing for the same file. The schema
+/// for all three tables is initialized once, here.
+static DB: LazyLock<Arc<db::DbPool>> =
+    LazyLock::new(|| db::open_store("data/task_queue.db").expect("failed to open database"));
+
+pub static CHAT_STORE: LazyLock<ChatStore> = LazyLock::new(|| ChatStore::new(Arc::clone(&DB)));
+pub static TASK_QUEUE: LazyLock<PersistentTaskQueue> =
+    LazyLock::new(|| PersistentTaskQueue::new(Arc::clone(&DB)));
+pub static LINK_CACHE: LazyLock<LinkCache> = LazyLock::new(|| LinkCache::new(Arc::clone(&DB)));
+pub static CONFIG: LazyLock<Config> = LazyLock::new(Config::load);

crates/xmedia-bot/src/handlers/urls.rs

@@ -0,0 +1,560 @@
+//! URL extraction and the per-URL media pipeline: bounded job channel +
+//! worker pool, link-cache fast path, fetch, task build and send dispatch.
+
+use super::{CHAT_STORE, CONFIG, LINK_CACHE, TASK_QUEUE, log_key, reply};
+use crate::config::Config;
+use crate::db::now_f64;
+use crate::link_cache::{CachedMediaKind, CachedPost, LinkCache};
+use crate::media_sender::MediaSender;
+use crate::queue::PersistentTaskQueue;
+use crate::send::{self, MediaItemPayload, Task};
+use crate::state::{ChatData, ChatStore};
+use std::collections::HashSet;
+use std::sync::LazyLock;
+use teloxide::types::{ChatAction, ChatId, Message, MessageEntityKind, MessageId};
+use x_media::media::Media;
+
+/// One URL job: the message + the extracted URL (the sender and stores come
+/// from the shared [`AppContext`], assembled from statics inside the worker).
+type UrlJob = (Message, String);
+/// Bounded channel of URL jobs drained by [`start_url_workers`]. The bound
+/// caps both queued memory and shutdown backlog; a full channel applies
+/// backpressure to the per-chat handler instead of spawning unbounded tasks.
+pub(crate) static URL_JOBS: LazyLock<
+    parking_lot::Mutex<Option<tokio::sync::mpsc::Sender<UrlJob>>>,
+> = LazyLock::new(|| parking_lot::Mutex::new(None));
+/// Set by main's shutdown sequence; workers stop pulling new jobs.
+pub(crate) static URL_STOP: std::sync::atomic::AtomicBool =
+    std::sync::atomic::AtomicBool::new(false);
+
+/// JoinHandles of the URL workers, awaited by [`stop_url_workers`].
+static URL_WORKER_HANDLES: LazyLock<parking_lot::Mutex<Option<Vec<tokio::task::JoinHandle<()>>>>> =
+    LazyLock::new(|| parking_lot::Mutex::new(None));
+
+/// Worker count draining URL jobs; keeps the old 8-permit concurrency cap
+/// while bounding how many jobs can be queued at all.
+const URL_WORKERS: usize = 8;
+
+/// Dependencies of the per-URL pipeline, injected so tests can substitute a
+/// mock sender and tempdir-backed stores.
+pub(crate) struct AppContext<'a> {
+    pub sender: &'a dyn MediaSender,
+    pub chat_store: &'a ChatStore,
+    pub task_queue: &'a PersistentTaskQueue,
+    pub link_cache: &'a LinkCache,
+    pub config: &'a Config,
+}
+
+/// Assembles the production context from the process-wide statics.
+fn app_context() -> AppContext<'static> {
+    AppContext {
+        sender: &*crate::send::BOT,
+        chat_store: &CHAT_STORE,
+        task_queue: &TASK_QUEUE,
+        link_cache: &LINK_CACHE,
+        config: &CONFIG,
+    }
+}
+
+/// Starts the URL job workers (called once from main after the queue starts).
+/// teloxide dispatches updates to a per-chat worker that handles them
+/// sequentially, so a batch-forward of many messages would otherwise be
+/// processed one at a time (fetch + send each, roughly a second per
+/// message); the workers add throughput, and FIFO order preserves per-message
+/// URL order.
+pub async fn start_url_workers() {
+    let (tx, rx) = tokio::sync::mpsc::channel::<UrlJob>(256);
+    *URL_JOBS.lock() = Some(tx);
+    let rx = std::sync::Arc::new(tokio::sync::Mutex::new(rx));
+    let mut handles = Vec::with_capacity(URL_WORKERS);
+    for _ in 0..URL_WORKERS {
+        let rx = std::sync::Arc::clone(&rx);
+        handles.push(tokio::spawn(async move {
+            let ctx = app_context();
+            while !URL_STOP.load(std::sync::atomic::Ordering::Relaxed) {
+                let job = rx.lock().await.recv().await;
+                match job {
+                    Some((message, url)) => {
+                        url_media(&ctx, message.chat.id.0, message.id.0 as i64, &url).await
+                    }
+                    None => break,
+                }
+            }
+        }));
+    }
+    *URL_WORKER_HANDLES.lock() = Some(handles);
+}
+
+/// Stops the URL workers: sets the stop flag, drops the job channel (so
+/// workers blocked in \`recv()\` wake with \`None\` and exit) and awaits the
+/// worker tasks. Each worker finishes its in-flight job first; jobs still
+/// queued in the channel are abandoned (the old implementation neither
+/// drained them nor woke blocked workers — it only set a flag checked
+/// between jobs).
+pub async fn stop_url_workers() {
+    URL_STOP.store(true, std::sync::atomic::Ordering::Relaxed);
+    // Dropping the sender makes every worker's recv() return None.
+    *URL_JOBS.lock() = None;
+    // Take the handles first so the lock guard drops before the awaits.
+    let handles = URL_WORKER_HANDLES.lock().take();
+    if let Some(handles) = handles {
+        for handle in handles {
+            let _ = handle.await;
+        }
+    }
+}
+
+/// Extracts URL and text-link entities (text + caption), deduped in order.
+pub fn extract_urls(message: &Message) -> Vec<String> {
+    let mut urls = Vec::new();
+    for entity in message.parse_entities().into_iter().flatten() {
+        match entity.kind() {
+            MessageEntityKind::Url => urls.push(entity.text().to_string()),
+            MessageEntityKind::TextLink { url } => urls.push(url.to_string()),
+            _ => {}
+        }
+    }
+    for entity in message.parse_caption_entities().into_iter().flatten() {
+        match entity.kind() {
+            MessageEntityKind::Url => urls.push(entity.text().to_string()),
+            MessageEntityKind::TextLink { url } => urls.push(url.to_string()),
+            _ => {}
+        }
+    }
+    let mut seen = HashSet::new();
+    // Dedup by the normalized post id so variant URLs of the same post
+    // (/status/1 vs /status/1/photo/1) are sent once; unsupported URLs fall
+    // back to exact-string dedup.
+    urls.retain(|url| seen.insert(x_media::site::cache_key(url).unwrap_or_else(|| url.clone())));
+    urls
+}
+
+/// For locally produced media (encoded ugoira MP4) the thumbnail URL is a
+/// hotlink-protected remote URL Telegram may not fetch; let Telegram generate
+/// its own thumbnail instead.
+fn thumbnail_for(media: &Media) -> Option<String> {
+    let url = media.url();
+    if url.starts_with("http://") || url.starts_with("https://") {
+        media.thumbnail_url().map(str::to_string)
+    } else {
+        None
+    }
+}
+
+fn media_to_payload(media: &Media, sensitive: bool) -> MediaItemPayload {
+    let fallback_url = media.smaller_url().map(str::to_string);
+    match media {
+        // A gif inside a group becomes a video item; a lone gif takes the
+        // animation path (see url_media).
+        Media::Illustration { .. } => MediaItemPayload::Photo {
+            media: media.url().to_string(),
+            has_spoiler: sensitive,
+            fallback_url,
+            file_id: false,
+        },
+        Media::Video { .. } => MediaItemPayload::Video {
+            media: media.url().to_string(),
+            has_spoiler: sensitive,
+            thumbnail: thumbnail_for(media),
+            fallback_url,
+            file_id: false,
+        },
+        Media::Animated { .. } => MediaItemPayload::Video {
+            media: media.url().to_string(),
+            has_spoiler: sensitive,
+            thumbnail: thumbnail_for(media),
+            fallback_url,
+            file_id: false,
+        },
+    }
+}
+
+pub(crate) async fn enqueue_retry(queue: &PersistentTaskQueue, task: Task, delay_seconds: f64) {
+    let payload = serde_json::to_value(task).expect("task serializes");
+    let run_after = now_f64() + delay_seconds;
+    if let Err(e) = queue.enqueue(payload, run_after).await {
+        log::error!("failed to enqueue retry: {e}");
+    }
+}
+
+/// Sends a task and handles the outcome: post-send actions on success, retry
+/// enqueue on retryable failure, reply + link-cache invalidation on
+/// permanent failure (a stale cached file id must not repeat forever).
+async fn dispatch_send(
+    ctx: &AppContext<'_>,
+    chat_id: i64,
+    reply_to: MessageId,
+    task: &Task,
+    url: &str,
+) {
+    let result = match task {
+        Task::SendAnimation { .. } => send::send_animation(ctx.sender, task).await,
+        Task::SendMediaSequence { .. } => send::send_media_sequence(ctx.sender, task).await,
+        Task::ForwardMessages { .. } => unreachable!(),
+    };
+    match result {
+        Ok(message_ids) => {
+            log::info!(
+                "sent {} message(s) for [key={}]",
+                message_ids.len(),
+                log_key(url)
+            );
+            send::post_send_actions(ctx.sender, task, message_ids).await;
+            // The task settled: drop any keep-alive temp media.
+            send::release_keep_alive(task);
+        }
+        Err(send::SendError::Retryable {
+            delay_seconds,
+            task,
+        }) => {
+            log::info!(
+                "send for [key={}] failed, queued for retry in {delay_seconds:.1}s",
+                log_key(url)
+            );
+            enqueue_retry(ctx.task_queue, task, delay_seconds).await;
+            let _ = reply(
+                ctx.sender,
+                chat_id,
+                reply_to,
+                "Send failed. Task queued for retry.",
+            )
+            .await;
+        }
+        Err(send::SendError::Permanent {
+            message: err_message,
+            task,
+        }) => {
+            send::invalidate_cache_with(ctx.link_cache, &task).await;
+            send::release_keep_alive(&task);
+            log::error!("send for {url} failed permanently: {err_message}");
+            let _ = reply(
+                ctx.sender,
+                chat_id,
+                reply_to,
+                format!("Send failed: {err_message}"),
+            )
+            .await;
+        }
+    }
+}
+
+/// Builds the send task from ready-made items, sharing the payload shape
+/// between the fresh-fetch and link-cache paths.
+#[allow(clippy::too_many_arguments)]
+fn build_send_task(
+    chat_data: &ChatData,
+    chat_id: i64,
+    reply_to_message_id: i64,
+    source_url: String,
+    caption: String,
+    items: Vec<MediaItemPayload>,
+    cache_data: Option<CachedPost>,
+) -> Task {
+    if items.len() == 1 && matches!(items[0], MediaItemPayload::Animation { .. }) {
+        Task::SendAnimation {
+            chat_id,
+            reply_to_message_id,
+            caption,
+            animation: items.into_iter().next().unwrap(),
+            source_url,
+            edit_before_forward: chat_data.edit_before_forward,
+            forward_channel_id: chat_data.forward_channel_id,
+            notify_chat_id: Some(chat_id),
+            notify_message_id: Some(reply_to_message_id),
+            cache_data,
+        }
+    } else {
+        Task::SendMediaSequence {
+            chat_id,
+            reply_to_message_id,
+            caption,
+            // Photos first so a mixed photo+video group starts with a photo
+            // (Telegram's sendMediaGroup rule); order within each kind is kept.
+            media_batches: send::chunk_media_items(send::photos_first(items)),
+            batch_index: 0,
+            sent_message_ids: vec![],
+            source_url,
+            edit_before_forward: chat_data.edit_before_forward,
+            forward_channel_id: chat_data.forward_channel_id,
+            notify_chat_id: Some(chat_id),
+            notify_message_id: Some(reply_to_message_id),
+            cache_data,
+        }
+    }
+}
+
+async fn url_media(ctx: &AppContext<'_>, chat_id: i64, reply_to_message_id: i64, url: &str) {
+    let reply_to = MessageId(reply_to_message_id as i32);
+    if let Err(e) = ctx
+        .sender
+        .send_chat_action(ChatId(chat_id), ChatAction::Typing)
+        .await
+    {
+        log::error!("send_chat_action failed: {e}");
+    }
+
+    // Link cache: a post sent before is re-sent from Telegram file ids —
+    // no source-site request, no download, no upload. Keyed by the
+    // normalized post id so x.com / fxtwitter / /photo/N variants collide.
+    if let Some(key) = x_media::site::cache_key(url)
+        && let Some(cached) = ctx.link_cache.get(&key, ctx.config.link_cache_ttl).await
+    {
+        log::debug!("link cache hit for {key}");
+        let chat_data = ctx.chat_store.get(chat_id).await;
+        // Cache keys are prefixed with the site id ("twitter:…"), matching
+        // the value a fresh fetch would read from Fetched::site_id.
+        let site = x_media::site::site_id_from_key(&key);
+        let format = chat_data
+            .message_format
+            .get(site)
+            .cloned()
+            .unwrap_or_default();
+        let caption = if format.is_empty() {
+            x_media::site::truncate_caption(&cached.caption)
+        } else {
+            x_media::site::caption_from_fields(
+                &format,
+                "",
+                &cached.url,
+                &cached.author,
+                &cached.author_url,
+                &cached.title,
+                &cached.tags,
+            )
+        };
+        let items: Vec<MediaItemPayload> = cached
+            .media
+            .iter()
+            .map(|m| match m.kind {
+                CachedMediaKind::Photo => MediaItemPayload::Photo {
+                    media: m.file_id.clone(),
+                    has_spoiler: cached.sensitive,
+                    fallback_url: None,
+                    file_id: true,
+                },
+                CachedMediaKind::Video => MediaItemPayload::Video {
+                    media: m.file_id.clone(),
+                    has_spoiler: cached.sensitive,
+                    thumbnail: None,
+                    fallback_url: None,
+                    file_id: true,
+                },
+                CachedMediaKind::Animation => MediaItemPayload::Animation {
+                    media: m.file_id.clone(),
+                    has_spoiler: cached.sensitive,
+                    file_id: true,
+                },
+            })
+            .collect();
+        let task = build_send_task(
+            &chat_data,
+            chat_id,
+            reply_to_message_id,
+            cached.url.clone(),
+            caption,
+            items,
+            Some(cached),
+        );
+        dispatch_send(ctx, chat_id, reply_to, &task, url).await;
+        return;
+    }
+
+    log::debug!("fetching {url} [key={}]", log_key(url));
+    match x_media::site::fetch(url).await {
+        // Unsupported links are ignored silently (Python parity).
+        Ok(None) => {
+            log::debug!("no site pattern matches {url}; ignoring");
+        }
+        // Retries exhausted: notify the user (Rust-only requirement 3).
+        Err(e) => {
+            log::error!("fetch {url}: {e}");
+            let _ = reply(
+                ctx.sender,
+                chat_id,
+                reply_to,
+                "Failed to fetch media from this link.",
+            )
+            .await;
+        }
+        Ok(Some(mut fetched)) => {
+            if fetched.media.is_empty() {
+                let _ = reply(
+                    ctx.sender,
+                    chat_id,
+                    reply_to,
+                    "No media found or media type is not supported.",
+                )
+                .await;
+                return;
+            }
+            let chat_data = ctx.chat_store.get(chat_id).await;
+            // Per-site caption format override (empty -> built-in caption).
+            let format = chat_data
+                .message_format
+                .get(fetched.site_name())
+                .cloned()
+                .unwrap_or_default();
+            let caption = fetched.caption_with(&format);
+            // Raw render data for the link cache; the send fills in the
+            // Telegram file ids and persists the entry.
+            let cache_data = fetched
+                .render_fields()
+                .map(|(author, author_url, title, tags)| CachedPost {
+                    url: fetched.source_url.clone(),
+                    caption: fetched.caption.clone(),
+                    title: title.to_string(),
+                    author: author.to_string(),
+                    author_url: author_url.to_string(),
+                    tags: tags.to_string(),
+                    sensitive: fetched.sensitive,
+                    media: vec![],
+                });
+            let items: Vec<MediaItemPayload> = fetched
+                .media
+                .iter()
+                .map(|media| media_to_payload(media, fetched.sensitive))
+                .collect();
+            let task = build_send_task(
+                &chat_data,
+                chat_id,
+                reply_to_message_id,
+                fetched.source_url.clone(),
+                caption,
+                items,
+                cache_data,
+            );
+            // Hand the keep-alive temp dir (ugoira / bsky remux MP4) to the
+            // retry registry: a queued retry runs after this function returns
+            // and the fetch's own TempDir is dropped, so without this the
+            // local file would be gone by the time the retry sends it.
+            if let Some(dir) = fetched.take_keep_alive() {
+                send::KEEP_ALIVE.lock().push(dir);
+            }
+            dispatch_send(ctx, chat_id, reply_to, &task, url).await;
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::db;
+    use crate::link_cache::CachedMedia;
+    use crate::media_sender::test_support::{MockSender, Outcome};
+    use std::sync::Arc;
+    use std::time::Duration;
+    use teloxide::{ApiError, RequestError};
+
+    fn permanent_error() -> RequestError {
+        RequestError::Api(ApiError::Unknown(
+            "Bad Request: message is not modified".into(),
+        ))
+    }
+
+    fn cached_photo_entry() -> CachedPost {
+        CachedPost {
+            url: "https://x.com/u/status/1".into(),
+            caption: "cap".into(),
+            title: "t".into(),
+            author: "a".into(),
+            author_url: "au".into(),
+            tags: "".into(),
+            sensitive: false,
+            media: vec![CachedMedia {
+                kind: CachedMediaKind::Photo,
+                file_id: "file-1".into(),
+            }],
+        }
+    }
+
+    #[tokio::test]
+    async fn cache_hit_sends_file_ids_and_invalidates_on_permanent_failure() {
+        let dir = tempfile::tempdir().unwrap();
+        let pool = db::open_store(dir.path().join("t.db").to_str().unwrap()).unwrap();
+        let chat_store = ChatStore::new(Arc::clone(&pool));
+        let task_queue = PersistentTaskQueue::new(Arc::clone(&pool));
+        let link_cache = LinkCache::new(Arc::clone(&pool));
+        let config = Config::load();
+        let sender = MockSender::scripted(
+            vec![Outcome::GroupErr, Outcome::MessageErr],
+            permanent_error,
+        );
+        let ctx = AppContext {
+            sender: &sender,
+            chat_store: &chat_store,
+            task_queue: &task_queue,
+            link_cache: &link_cache,
+            config: &config,
+        };
+        link_cache.put("twitter:1", &cached_photo_entry()).await;
+
+        url_media(&ctx, 1, 2, "https://x.com/u/status/1").await;
+
+        // The cached file id went out as a group send; the permanent failure
+        // then triggered the fire-and-forget reply (its mock error is fine).
+        assert_eq!(
+            sender.calls(),
+            vec!["send_chat_action", "send_media_group", "send_message"]
+        );
+        // The stale cache entry was invalidated so the next request re-fetches.
+        assert!(
+            link_cache
+                .get("twitter:1", Duration::from_secs(3600))
+                .await
+                .is_none()
+        );
+    }
+
+    #[tokio::test]
+    async fn cache_hit_success_keeps_the_cache_entry() {
+        let dir = tempfile::tempdir().unwrap();
+        let pool = db::open_store(dir.path().join("t.db").to_str().unwrap()).unwrap();
+        let chat_store = ChatStore::new(Arc::clone(&pool));
+        let task_queue = PersistentTaskQueue::new(Arc::clone(&pool));
+        let link_cache = LinkCache::new(Arc::clone(&pool));
+        let config = Config::load();
+        let sender = MockSender::scripted(vec![Outcome::GroupOk], permanent_error);
+        let ctx = AppContext {
+            sender: &sender,
+            chat_store: &chat_store,
+            task_queue: &task_queue,
+            link_cache: &link_cache,
+            config: &config,
+        };
+        link_cache.put("twitter:1", &cached_photo_entry()).await;
+
+        url_media(&ctx, 1, 2, "https://x.com/u/status/1").await;
+
+        assert_eq!(sender.calls(), vec!["send_chat_action", "send_media_group"]);
+        // Success must not evict the entry.
+        assert!(
+            link_cache
+                .get("twitter:1", Duration::from_secs(3600))
+                .await
+                .is_some()
+        );
+    }
+
+    #[tokio::test]
+    async fn unsupported_url_is_ignored_silently() {
+        let dir = tempfile::tempdir().unwrap();
+        let pool = db::open_store(dir.path().join("t.db").to_str().unwrap()).unwrap();
+        let chat_store = ChatStore::new(Arc::clone(&pool));
+        let task_queue = PersistentTaskQueue::new(Arc::clone(&pool));
+        let link_cache = LinkCache::new(Arc::clone(&pool));
+        let config = Config::load();
+        let sender = MockSender::scripted(vec![], permanent_error);
+        let ctx = AppContext {
+            sender: &sender,
+            chat_store: &chat_store,
+            task_queue: &task_queue,
+            link_cache: &link_cache,
+            config: &config,
+        };
+
+        // No cache key → the fetch dispatcher returns Ok(None) without any
+        // network; nothing is sent or replied.
+        url_media(&ctx, 1, 2, "https://example.com/not-a-post").await;
+        assert_eq!(sender.calls(), vec!["send_chat_action"]);
+    }
+}

crates/xmedia-bot/src/main.rs

@@ -11,8 +11,10 @@ mod config;
 mod db;
 mod handlers;
 mod link_cache;
+mod media_sender;
 mod photo;
 mod queue;
+mod rate_limit;
 mod send;
 mod state;
 

crates/xmedia-bot/src/media_sender.rs

@@ -0,0 +1,300 @@
+//! Send abstraction: the message-sending surface [`send`](crate::send)
+//! needs, so the send pipeline can be tested with a scripted mock instead of
+//! a live teloxide `Bot`.
+
+use std::future::Future;
+use std::pin::Pin;
+use teloxide::RequestError;
+use teloxide::prelude::Requester;
+use teloxide::prelude::*;
+use teloxide::types::{
+    ChatAction, ChatId, InlineKeyboardMarkup, InputFile, InputMedia, Message, MessageId, ParseMode,
+    ReplyParameters,
+};
+
+/// Boxed, `Send` future returned by a [`MediaSender`] method (`async fn` in
+/// traits is not dyn-compatible).
+type BoxFuture<'a, T> = Pin<Box<dyn Future<Output = T> + Send + 'a>>;
+
+/// The message-sending surface the send pipeline uses. The production
+/// implementation is teloxide's [`Bot`]; tests inject a scripted mock to
+/// cover the fallback and classification logic without touching the
+/// Telegram API.
+pub trait MediaSender: Send + Sync {
+    /// Sends a media group, replying to `reply_to`.
+    fn send_media_group(
+        &self,
+        chat_id: ChatId,
+        reply_to: MessageId,
+        items: Vec<InputMedia>,
+    ) -> BoxFuture<'_, Result<Vec<Message>, RequestError>>;
+
+    /// Sends a lone animation, replying to `reply_to`.
+    fn send_animation<'a>(
+        &'a self,
+        chat_id: ChatId,
+        reply_to: MessageId,
+        caption: &'a str,
+        spoiler: bool,
+        file: InputFile,
+    ) -> BoxFuture<'a, Result<Message, RequestError>>;
+
+    /// Copies messages between chats (forward to channel).
+    fn copy_messages(
+        &self,
+        to: ChatId,
+        from: ChatId,
+        ids: Vec<MessageId>,
+    ) -> BoxFuture<'_, Result<Vec<MessageId>, RequestError>>;
+
+    /// Sends a plain text message, optionally replying to `reply_to` and
+    /// attaching `reply_markup`.
+    fn send_message(
+        &self,
+        chat_id: ChatId,
+        text: String,
+        reply_to: Option<MessageId>,
+        reply_markup: Option<InlineKeyboardMarkup>,
+    ) -> BoxFuture<'_, Result<Message, RequestError>>;
+
+    /// Sets the chat's "typing / uploading …" indicator (cosmetic).
+    fn send_chat_action(
+        &self,
+        chat_id: ChatId,
+        action: ChatAction,
+    ) -> BoxFuture<'_, Result<(), RequestError>>;
+}
+
+impl MediaSender for Bot {
+    fn send_media_group(
+        &self,
+        chat_id: ChatId,
+        reply_to: MessageId,
+        items: Vec<InputMedia>,
+    ) -> BoxFuture<'_, Result<Vec<Message>, RequestError>> {
+        Box::pin(async move {
+            // Pace media sends per chat (one token per item) so bursts do not
+            // trip Telegram's flood control.
+            crate::rate_limit::limiter_for(chat_id.0)
+                .acquire(items.len() as f64)
+                .await;
+            // `<Bot as Requester>::` disambiguates from this trait's same-named
+            // method (teloxide's API lives in the `Requester` trait).
+            <Bot as Requester>::send_media_group(self, chat_id, items)
+                .reply_parameters(ReplyParameters::new(reply_to).allow_sending_without_reply())
+                .await
+        })
+    }
+
+    fn send_animation<'a>(
+        &'a self,
+        chat_id: ChatId,
+        reply_to: MessageId,
+        caption: &'a str,
+        spoiler: bool,
+        file: InputFile,
+    ) -> BoxFuture<'a, Result<Message, RequestError>> {
+        Box::pin(async move {
+            crate::rate_limit::limiter_for(chat_id.0).acquire(1.0).await;
+            let mut request = <Bot as Requester>::send_animation(self, chat_id, file)
+                .caption(caption)
+                .parse_mode(ParseMode::Html)
+                .reply_parameters(ReplyParameters::new(reply_to).allow_sending_without_reply());
+            if spoiler {
+                request = request.has_spoiler(true);
+            }
+            request.await
+        })
+    }
+
+    fn copy_messages(
+        &self,
+        to: ChatId,
+        from: ChatId,
+        ids: Vec<MessageId>,
+    ) -> BoxFuture<'_, Result<Vec<MessageId>, RequestError>> {
+        Box::pin(async move {
+            // Channel forwards are the burstiest path (batch copies); pace
+            // them per message against the channel's budget.
+            crate::rate_limit::limiter_for(to.0)
+                .acquire(ids.len() as f64)
+                .await;
+            <Bot as Requester>::copy_messages(self, to, from, ids).await
+        })
+    }
+
+    fn send_message(
+        &self,
+        chat_id: ChatId,
+        text: String,
+        reply_to: Option<MessageId>,
+        reply_markup: Option<InlineKeyboardMarkup>,
+    ) -> BoxFuture<'_, Result<Message, RequestError>> {
+        Box::pin(async move {
+            let mut request = <Bot as Requester>::send_message(self, chat_id, text);
+            if let Some(reply_to) = reply_to {
+                request = request
+                    .reply_parameters(ReplyParameters::new(reply_to).allow_sending_without_reply());
+            }
+            if let Some(markup) = reply_markup {
+                request = request.reply_markup(markup);
+            }
+            request.await
+        })
+    }
+
+    fn send_chat_action(
+        &self,
+        chat_id: ChatId,
+        action: ChatAction,
+    ) -> BoxFuture<'_, Result<(), RequestError>> {
+        Box::pin(async move {
+            // teloxide's `send_chat_action` returns `Result<True, _>` (its
+            // unit marker type); map the success to `()`.
+            <Bot as Requester>::send_chat_action(self, chat_id, action)
+                .await
+                .map(|_| ())
+        })
+    }
+}
+
+/// Test support: a scripted [`MediaSender`] mock (no Telegram API involved).
+#[cfg(test)]
+pub(crate) mod test_support {
+    use super::*;
+    use std::sync::Mutex;
+
+    /// One scripted outcome, consumed front-to-back; the last entry repeats
+    /// for further calls of the same method kind.
+    #[derive(Clone, Copy, Debug, PartialEq, Eq)]
+    pub(crate) enum Outcome {
+        GroupOk,
+        GroupErr,
+        AnimationErr,
+        CopyOk,
+        CopyErr,
+        /// An error from `send_message` (replies are fire-and-forget, so an
+        /// error is fine for tests).
+        MessageErr,
+    }
+
+    /// Replays a script and records the method names that were called.
+    pub(crate) struct MockSender {
+        script: Mutex<Vec<Outcome>>,
+        cursor: Mutex<usize>,
+        calls: Mutex<Vec<&'static str>>,
+        /// Builds the error every `*Err` outcome returns (RequestError is not
+        /// cloneable, so the factory recreates it per call).
+        error: Box<dyn Fn() -> RequestError + Send + Sync>,
+    }
+
+    impl MockSender {
+        pub(crate) fn scripted(
+            script: Vec<Outcome>,
+            error: impl Fn() -> RequestError + Send + Sync + 'static,
+        ) -> Self {
+            MockSender {
+                script: Mutex::new(script),
+                cursor: Mutex::new(0),
+                calls: Mutex::new(Vec::new()),
+                error: Box::new(error),
+            }
+        }
+
+        /// Method names in call order (e.g. `["send_media_group",
+        /// "send_media_group"]` proves the fallback re-sent).
+        pub(crate) fn calls(&self) -> Vec<&'static str> {
+            self.calls.lock().unwrap().clone()
+        }
+
+        fn next(&self, kind: &'static str) -> Outcome {
+            self.calls.lock().unwrap().push(kind);
+            let script = self.script.lock().unwrap();
+            let mut cursor = self.cursor.lock().unwrap();
+            if script.is_empty() {
+                panic!("mock script exhausted: {kind}");
+            }
+            let idx = (*cursor).min(script.len() - 1);
+            *cursor = idx + 1;
+            script[idx]
+        }
+
+        fn error(&self) -> RequestError {
+            (self.error)()
+        }
+    }
+
+    impl MediaSender for MockSender {
+        fn send_media_group(
+            &self,
+            _chat_id: ChatId,
+            _reply_to: MessageId,
+            _items: Vec<InputMedia>,
+        ) -> BoxFuture<'_, Result<Vec<Message>, RequestError>> {
+            Box::pin(async move {
+                match self.next("send_media_group") {
+                    Outcome::GroupOk => Ok(Vec::new()),
+                    Outcome::GroupErr => Err(self.error()),
+                    other => panic!("unexpected outcome {other:?} for send_media_group"),
+                }
+            })
+        }
+
+        fn send_animation<'a>(
+            &'a self,
+            _chat_id: ChatId,
+            _reply_to: MessageId,
+            _caption: &'a str,
+            _spoiler: bool,
+            _file: InputFile,
+        ) -> BoxFuture<'a, Result<Message, RequestError>> {
+            Box::pin(async move {
+                match self.next("send_animation") {
+                    Outcome::AnimationErr => Err(self.error()),
+                    other => panic!("unexpected outcome {other:?} for send_animation"),
+                }
+            })
+        }
+
+        fn copy_messages(
+            &self,
+            _to: ChatId,
+            _from: ChatId,
+            _ids: Vec<MessageId>,
+        ) -> BoxFuture<'_, Result<Vec<MessageId>, RequestError>> {
+            Box::pin(async move {
+                match self.next("copy_messages") {
+                    Outcome::CopyOk => Ok(vec![MessageId(1)]),
+                    Outcome::CopyErr => Err(self.error()),
+                    other => panic!("unexpected outcome {other:?} for copy_messages"),
+                }
+            })
+        }
+
+        fn send_message(
+            &self,
+            _chat_id: ChatId,
+            _text: String,
+            _reply_to: Option<MessageId>,
+            _reply_markup: Option<InlineKeyboardMarkup>,
+        ) -> BoxFuture<'_, Result<Message, RequestError>> {
+            Box::pin(async move {
+                match self.next("send_message") {
+                    Outcome::MessageErr => Err(self.error()),
+                    other => panic!("unexpected outcome {other:?} for send_message"),
+                }
+            })
+        }
+
+        fn send_chat_action(
+            &self,
+            _chat_id: ChatId,
+            _action: ChatAction,
+        ) -> BoxFuture<'_, Result<(), RequestError>> {
+            Box::pin(async move {
+                self.calls.lock().unwrap().push("send_chat_action");
+                Ok(())
+            })
+        }
+    }
+}

crates/xmedia-bot/src/rate_limit.rs

@@ -0,0 +1,136 @@
+//! Per-chat token-bucket rate limiting.
+//!
+//! Telegram throttles bots that burst past a chat's message budget
+//! (roughly 20 messages/min for channels/groups); today the bot absorbs
+//! those 429s with queue retries. This limiter smooths the burst *before*
+//! it reaches the API: media sends to a chat consume one token per
+//! message, refilled at [`REFILL_PER_SEC`], so a batch forward paces itself
+//! instead of tripping flood control. The queue retry stays as the safety
+//! net for limits this bucket does not model (global per-bot limits etc.).
+
+use parking_lot::Mutex;
+use std::collections::HashMap;
+use std::sync::Arc;
+use std::sync::LazyLock;
+use std::time::Duration;
+
+/// Burst capacity: how many messages may be sent at once without waiting.
+const CAPACITY: f64 = 20.0;
+/// Sustained refill: ~20 messages per minute.
+const REFILL_PER_SEC: f64 = 20.0 / 60.0;
+
+struct State {
+    /// Current token balance; may go negative (debt from an acquire larger
+    /// than the capacity, repaid by subsequent refills).
+    tokens: f64,
+    last_refill: tokio::time::Instant,
+}
+
+/// A token bucket: at most `CAPACITY` tokens accumulate, refilled at
+/// `REFILL_PER_SEC`. [`TokenBucket::acquire`] consumes `n` tokens, waiting
+/// for the deficit (a single acquire may exceed the capacity and goes into
+/// debt, which the refill repays).
+pub struct TokenBucket {
+    capacity: f64,
+    refill_per_sec: f64,
+    state: Mutex<State>,
+}
+
+impl TokenBucket {
+    fn new(capacity: f64, refill_per_sec: f64) -> Self {
+        TokenBucket {
+            capacity,
+            refill_per_sec,
+            state: Mutex::new(State {
+                tokens: capacity,
+                last_refill: tokio::time::Instant::now(),
+            }),
+        }
+    }
+
+    /// Waits until `n` tokens are available, consuming them. The wait is
+    /// bounded: the deficit is committed as debt and repaid over time, so a
+    /// large acquire returns once its share of the refill budget has passed.
+    pub async fn acquire(&self, n: f64) {
+        // The parking_lot guard is confined to this block: only the plain
+        // `wait` duration crosses the await (a guard across an await point
+        // would make the future !Send).
+        let wait = {
+            let mut state = self.state.lock();
+            let now = tokio::time::Instant::now();
+            let elapsed = now
+                .saturating_duration_since(state.last_refill)
+                .as_secs_f64();
+            // Refill up to the capacity; a debt (negative balance) is repaid
+            // before any surplus accumulates.
+            state.tokens = (state.tokens + elapsed * self.refill_per_sec).min(self.capacity);
+            state.last_refill = now;
+            if state.tokens >= n {
+                state.tokens -= n;
+                return;
+            }
+            // Commit the whole consumption now; the caller proceeds once the
+            // deficit's worth of refill time has passed.
+            let debt = n - state.tokens;
+            state.tokens = -debt;
+            debt / self.refill_per_sec
+        };
+        tokio::time::sleep(Duration::from_secs_f64(wait)).await;
+    }
+}
+
+/// One limiter per chat, created on first use. Per-chat so one chat's burst
+/// never throttles another.
+static LIMITERS: LazyLock<Mutex<HashMap<i64, Arc<TokenBucket>>>> =
+    LazyLock::new(|| Mutex::new(HashMap::new()));
+
+/// Returns the shared limiter for a chat, creating it on first use.
+pub fn limiter_for(chat_id: i64) -> Arc<TokenBucket> {
+    LIMITERS
+        .lock()
+        .entry(chat_id)
+        .or_insert_with(|| Arc::new(TokenBucket::new(CAPACITY, REFILL_PER_SEC)))
+        .clone()
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn limiter_for_reuses_the_per_chat_bucket() {
+        let a = limiter_for(1);
+        let b = limiter_for(1);
+        let c = limiter_for(2);
+        assert!(Arc::ptr_eq(&a, &b), "same chat → same bucket");
+        assert!(!Arc::ptr_eq(&a, &c), "different chat → different bucket");
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn burst_is_consumed_instantly_then_refill_waits() {
+        let bucket = TokenBucket::new(3.0, 1.0);
+        // A burst within capacity passes without waiting.
+        bucket.acquire(3.0).await;
+        // The bucket is empty now; one token needs 1s of refill.
+        let start = tokio::time::Instant::now();
+        bucket.acquire(1.0).await;
+        assert!(
+            start.elapsed() >= Duration::from_secs(1),
+            "elapsed {:?}",
+            start.elapsed()
+        );
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn acquire_larger_than_capacity_waits_for_the_deficit() {
+        let bucket = TokenBucket::new(2.0, 1.0);
+        // 5 tokens with a capacity of 2: the 3-token deficit takes 3s.
+        let start = tokio::time::Instant::now();
+        bucket.acquire(5.0).await;
+        assert!(
+            start.elapsed() >= Duration::from_secs(3),
+            "elapsed {:?}",
+            start.elapsed()
+        );
+    }
+}

crates/xmedia-bot/src/send.rs

@@ -4,7 +4,8 @@
 //! and uploads it via multipart).
 
 use crate::handlers::{CHAT_STORE, LINK_CACHE, TASK_QUEUE, log_key};
-use crate::link_cache::{CachedMedia, CachedMediaKind, CachedPost};
+use crate::link_cache::{CachedMedia, CachedMediaKind, CachedPost, LinkCache};
+use crate::media_sender::MediaSender;
 use crate::photo::{self, MAX_UPLOAD_BYTES, PhotoPrep};
 use crate::queue::QueueError;
 use crate::state::{EditMessage, unix_now};
@@ -15,7 +16,7 @@ use std::sync::LazyLock;
 use teloxide::prelude::*;
 use teloxide::types::{
     ChatId, InlineKeyboardButton, InlineKeyboardMarkup, InputFile, InputMedia, InputMediaAnimation,
-    InputMediaPhoto, InputMediaVideo, Message, MessageId, ParseMode, ReplyParameters,
+    InputMediaPhoto, InputMediaVideo, Message, MessageId, ParseMode,
 };
 use teloxide::{ApiError, RequestError};
 use tempfile::NamedTempFile;
@@ -253,12 +254,17 @@ async fn cache_animation_send(task: &Task, message: &Message) {
 /// A cached Telegram file id failed permanently (stale/expired); drop the
 /// cache entry so the next request re-fetches instead of repeating it.
 pub async fn invalidate_cache(task: &Task) {
+    invalidate_cache_with(&LINK_CACHE, task).await;
+}
+
+/// [`invalidate_cache`] against an injected cache (tests pass a tempdir one).
+pub async fn invalidate_cache_with(cache: &LinkCache, task: &Task) {
     if task.is_cached_send()
         && let Some(url) = task.source_url()
         && let Some(key) = x_media::site::cache_key(url)
     {
         log::debug!("removing stale link cache entry for [key={}]", log_key(url));
-        LINK_CACHE.remove(&key).await;
+        cache.remove(&key).await;
     }
 }
 
@@ -380,6 +386,7 @@ pub fn classify_request_error(e: &RequestError) -> Classification {
     }
 }
 
+#[derive(Debug)]
 pub enum SendError {
     Retryable { delay_seconds: f64, task: Task },
     Permanent { message: String, task: Task },
@@ -807,7 +814,7 @@ async fn prepare_upload_item(
 /// original order. Returns the fallback-error without the task attached;
 /// callers wrap it with the updated task state.
 async fn send_batch_via_upload(
-    bot: &Bot,
+    sender: &dyn MediaSender,
     chat_id: i64,
     reply_to: i64,
     batch: &[MediaItemPayload],
@@ -859,11 +866,8 @@ async fn send_batch_via_upload(
         .map(|m| m.expect("every upload item was prepared"))
         .collect();
     // `keep_alive` holds the temp files until the group request completes.
-    let result = bot
-        .send_media_group(ChatId(chat_id), items)
-        .reply_parameters(
-            ReplyParameters::new(MessageId(reply_to as i32)).allow_sending_without_reply(),
-        )
+    let result = sender
+        .send_media_group(ChatId(chat_id), MessageId(reply_to as i32), items)
         .await;
     drop(keep_alive);
     match result {
@@ -918,7 +922,10 @@ fn updated_sequence_task(task: &Task, batch_index: usize, sent_message_ids: Vec<
 /// Sends the media batches starting at `task.batch_index`, extending
 /// `sent_message_ids`. Returns all sent message ids on full success; on
 /// failure returns a [`SendError`] whose task carries the resumed state.
-pub async fn send_media_sequence(bot: &Bot, task: &Task) -> Result<Vec<i64>, SendError> {
+pub async fn send_media_sequence(
+    sender: &dyn MediaSender,
+    task: &Task,
+) -> Result<Vec<i64>, SendError> {
     let Task::SendMediaSequence {
         chat_id,
         reply_to_message_id,
@@ -954,11 +961,8 @@ pub async fn send_media_sequence(bot: &Bot, task: &Task) -> Result<Vec<i64>, Sen
                 });
             }
         };
-        match bot
-            .send_media_group(ChatId(chat_id), items)
-            .reply_parameters(
-                ReplyParameters::new(MessageId(reply_to as i32)).allow_sending_without_reply(),
-            )
+        match sender
+            .send_media_group(ChatId(chat_id), MessageId(reply_to as i32), items)
             .await
         {
             Ok(messages) => {
@@ -980,7 +984,7 @@ pub async fn send_media_sequence(bot: &Bot, task: &Task) -> Result<Vec<i64>, Sen
                         .unwrap_or_else(|| "?".into())
                 );
                 match send_batch_via_upload(
-                    bot,
+                    sender,
                     chat_id,
                     reply_to,
                     batch,
@@ -1011,28 +1015,26 @@ pub async fn send_media_sequence(bot: &Bot, task: &Task) -> Result<Vec<i64>, Sen
 }
 
 async fn send_animation_inner(
-    bot: &Bot,
+    sender: &dyn MediaSender,
     chat_id: i64,
     reply_to: i64,
     caption: &str,
     spoiler: bool,
     file: InputFile,
 ) -> Result<Message, RequestError> {
-    let mut request = bot
-        .send_animation(ChatId(chat_id), file)
-        .caption(caption)
-        .parse_mode(ParseMode::Html)
-        .reply_parameters(
-            ReplyParameters::new(MessageId(reply_to as i32)).allow_sending_without_reply(),
-        );
-    if spoiler {
-        request = request.has_spoiler(true);
-    }
-    request.await
+    sender
+        .send_animation(
+            ChatId(chat_id),
+            MessageId(reply_to as i32),
+            caption,
+            spoiler,
+            file,
+        )
+        .await
 }
 
 /// Sends a lone animation (gif), URL first with the download fallback.
-pub async fn send_animation(bot: &Bot, task: &Task) -> Result<Vec<i64>, SendError> {
+pub async fn send_animation(sender: &dyn MediaSender, task: &Task) -> Result<Vec<i64>, SendError> {
     let Task::SendAnimation {
         chat_id,
         reply_to_message_id,
@@ -1062,7 +1064,7 @@ pub async fn send_animation(bot: &Bot, task: &Task) -> Result<Vec<i64>, SendErro
             });
         }
     };
-    match send_animation_inner(bot, chat_id, reply_to, caption, has_spoiler, url_file).await {
+    match send_animation_inner(sender, chat_id, reply_to, caption, has_spoiler, url_file).await {
         Ok(message) => {
             let id = message.id.0 as i64;
             cache_animation_send(task, &message).await;
@@ -1089,7 +1091,7 @@ pub async fn send_animation(bot: &Bot, task: &Task) -> Result<Vec<i64>, SendErro
                     // Hold the temp file until the request completes.
                     let _keep_alive = keep_alive;
                     match send_animation_inner(
-                        bot,
+                        sender,
                         chat_id,
                         reply_to,
                         caption,
@@ -1115,7 +1117,7 @@ pub async fn send_animation(bot: &Bot, task: &Task) -> Result<Vec<i64>, SendErro
 
 /// Copies already-sent messages to the forward channel. No download fallback:
 /// the files are already on Telegram's servers.
-pub async fn forward_messages(bot: &Bot, task: &Task) -> Result<(), SendError> {
+pub async fn forward_messages(sender: &dyn MediaSender, task: &Task) -> Result<(), SendError> {
     let Task::ForwardMessages {
         from_chat_id,
         to_chat_id,
@@ -1129,7 +1131,7 @@ pub async fn forward_messages(bot: &Bot, task: &Task) -> Result<(), SendError> {
         .iter()
         .map(|id| MessageId(*id as i32))
         .collect::<Vec<_>>();
-    match bot
+    match sender
         .copy_messages(
             ChatId(*to_chat_id),
             ChatId(*from_chat_id),
@@ -1169,26 +1171,24 @@ pub fn build_edit_markup(templates: &HashMap<String, String>) -> InlineKeyboardM
 /// Notifies a chat about a dead-lettered task (skips when `notify_chat_id` is
 /// absent).
 pub async fn notify_failure(
-    bot: &Bot,
+    sender: &dyn MediaSender,
     chat_id: Option<i64>,
     message_id: Option<i64>,
     message: &str,
 ) {
     let Some(chat_id) = chat_id else { return };
-    let mut request = bot.send_message(ChatId(chat_id), message);
-    if let Some(message_id) = message_id {
-        request = request.reply_parameters(
-            ReplyParameters::new(MessageId(message_id as i32)).allow_sending_without_reply(),
-        );
-    }
-    if let Err(e) = request.await {
+    let reply_to = message_id.map(|id| MessageId(id as i32));
+    if let Err(e) = sender
+        .send_message(ChatId(chat_id), message.to_string(), reply_to, None)
+        .await
+    {
         log::error!("failed to notify about failed task: {e}");
     }
 }
 
 /// After a successful send: either open the edit-before-forward prompt or
 /// forward to the configured channel (with retry/queue handling).
-pub async fn post_send_actions(bot: &Bot, task: &Task, message_ids: Vec<i64>) {
+pub async fn post_send_actions(sender: &dyn MediaSender, task: &Task, message_ids: Vec<i64>) {
     let (
         chat_id,
         reply_to,
@@ -1231,14 +1231,15 @@ pub async fn post_send_actions(bot: &Bot, task: &Task, message_ids: Vec<i64>) {
 
     if edit_before_forward {
         let keyboard = build_edit_markup(&CHAT_STORE.get(chat_id).await.template);
-        match bot
-            .send_message(ChatId(chat_id), "Reply to edit message.")
-            .reply_markup(keyboard)
-            .reply_parameters(
-                ReplyParameters::new(MessageId(reply_to as i32)).allow_sending_without_reply(),
+        let prompt = sender
+            .send_message(
+                ChatId(chat_id),
+                "Reply to edit message.".to_string(),
+                Some(MessageId(reply_to as i32)),
+                Some(keyboard),
             )
-            .await
-        {
+            .await;
+        match prompt {
             Ok(prompt) => {
                 log::info!(
                     "edit-before-forward prompt {} opened for {} message(s)",
@@ -1279,7 +1280,7 @@ pub async fn post_send_actions(bot: &Bot, task: &Task, message_ids: Vec<i64>) {
             notify_chat_id,
             notify_message_id,
         };
-        match forward_messages(bot, &forward_task).await {
+        match forward_messages(sender, &forward_task).await {
             Ok(()) => {}
             Err(SendError::Retryable {
                 delay_seconds,
@@ -1297,7 +1298,7 @@ pub async fn post_send_actions(bot: &Bot, task: &Task, message_ids: Vec<i64>) {
             }
             Err(SendError::Permanent { message, .. }) => {
                 notify_failure(
-                    bot,
+                    sender,
                     notify_chat_id,
                     notify_message_id,
                     &format!("Task failed after retries: {message}"),
@@ -1381,10 +1382,13 @@ pub async fn handle_task(payload: serde_json::Value) -> Result<(), QueueError> {
     }
 }
 
-async fn send_media_or_animation(bot: &Bot, task: &Task) -> Result<Vec<i64>, SendError> {
+async fn send_media_or_animation(
+    sender: &dyn MediaSender,
+    task: &Task,
+) -> Result<Vec<i64>, SendError> {
     match task {
-        Task::SendMediaSequence { .. } => send_media_sequence(bot, task).await,
-        Task::SendAnimation { .. } => send_animation(bot, task).await,
+        Task::SendMediaSequence { .. } => send_media_sequence(sender, task).await,
+        Task::SendAnimation { .. } => send_animation(sender, task).await,
         Task::ForwardMessages { .. } => unreachable!(),
     }
 }
@@ -1656,4 +1660,169 @@ mod tests {
         assert_eq!(sniff_ext(b"\x00\x00\x00\x18ftypisom"), "mp4");
         assert_eq!(sniff_ext(b"something else"), "bin");
     }
+
+    // ── MediaSender-mock tests: fallback trigger + error classification ──
+
+    use crate::media_sender::test_support::{MockSender, Outcome};
+
+    /// Telegram's "I could not fetch this URL" error, which triggers the
+    /// download-and-reupload fallback.
+    fn media_fetch_error() -> RequestError {
+        RequestError::Api(ApiError::Unknown("Bad Request: WEBPAGE_MEDIA_EMPTY".into()))
+    }
+
+    fn sequence_task(media: &str) -> Task {
+        Task::SendMediaSequence {
+            chat_id: 1,
+            reply_to_message_id: 2,
+            caption: "cap".into(),
+            media_batches: vec![vec![MediaItemPayload::Photo {
+                media: media.to_string(),
+                has_spoiler: false,
+                fallback_url: None,
+                file_id: false,
+            }]],
+            batch_index: 0,
+            sent_message_ids: vec![],
+            source_url: "https://x.com/u/status/1".into(),
+            edit_before_forward: false,
+            forward_channel_id: None,
+            notify_chat_id: Some(1),
+            notify_message_id: Some(2),
+            cache_data: None,
+        }
+    }
+
+    #[tokio::test]
+    async fn media_group_fetch_failure_falls_back_then_permanent() {
+        // A local file avoids any network in the fallback (the prep pipeline
+        // uploads local paths directly). The first group send fails with a
+        // media-fetch error → the download-reupload fallback runs → the
+        // reupload also fails → Permanent.
+        let dir = tempfile::tempdir().unwrap();
+        let file = dir.path().join("media.jpg");
+        std::fs::write(&file, b"not-a-real-jpeg").unwrap();
+        let sender = MockSender::scripted(
+            vec![Outcome::GroupErr, Outcome::GroupErr],
+            media_fetch_error,
+        );
+        let task = sequence_task(file.to_str().unwrap());
+        let result = send_media_sequence(&sender, &task).await;
+        assert!(
+            matches!(result, Err(SendError::Permanent { .. })),
+            "got {result:?}"
+        );
+        // Two group sends: the original + the fallback reupload.
+        assert_eq!(sender.calls(), vec!["send_media_group", "send_media_group"]);
+    }
+
+    #[tokio::test]
+    async fn media_group_retry_after_classifies_retryable_without_fallback() {
+        use teloxide::types::Seconds;
+        // RetryAfter is not a media-fetch failure: no fallback, straight to a
+        // retryable error carrying the Telegram delay.
+        let dir = tempfile::tempdir().unwrap();
+        let file = dir.path().join("media.jpg");
+        std::fs::write(&file, b"not-a-real-jpeg").unwrap();
+        let sender = MockSender::scripted(vec![Outcome::GroupErr], || {
+            RequestError::RetryAfter(Seconds::from_seconds(7))
+        });
+        let task = sequence_task(file.to_str().unwrap());
+        let result = send_media_sequence(&sender, &task).await;
+        match result {
+            Err(SendError::Retryable { delay_seconds, .. }) => {
+                assert_eq!(delay_seconds, 7.0)
+            }
+            other => panic!("expected Retryable, got {other:?}"),
+        }
+        assert_eq!(sender.calls(), vec!["send_media_group"]);
+    }
+
+    #[tokio::test]
+    async fn animation_fetch_failure_falls_back_then_permanent() {
+        let dir = tempfile::tempdir().unwrap();
+        let file = dir.path().join("gif.mp4");
+        std::fs::write(&file, b"not-a-real-mp4").unwrap();
+        let sender = MockSender::scripted(
+            vec![Outcome::AnimationErr, Outcome::AnimationErr],
+            media_fetch_error,
+        );
+        let task = Task::SendAnimation {
+            chat_id: 1,
+            reply_to_message_id: 2,
+            caption: "cap".into(),
+            animation: MediaItemPayload::Animation {
+                media: file.to_string_lossy().into_owned(),
+                has_spoiler: false,
+                file_id: false,
+            },
+            source_url: "https://x.com/u/status/1".into(),
+            edit_before_forward: false,
+            forward_channel_id: None,
+            notify_chat_id: Some(1),
+            notify_message_id: Some(2),
+            cache_data: None,
+        };
+        let result = send_animation(&sender, &task).await;
+        assert!(
+            matches!(result, Err(SendError::Permanent { .. })),
+            "got {result:?}"
+        );
+        assert_eq!(sender.calls(), vec!["send_animation", "send_animation"]);
+    }
+
+    #[tokio::test]
+    async fn media_group_success_and_forward_ok() {
+        // GroupOk: the group send succeeds (empty message list → no file ids
+        // collected, the batch counts as sent). CopyOk: the forward succeeds.
+        let dir = tempfile::tempdir().unwrap();
+        let file = dir.path().join("media.jpg");
+        std::fs::write(&file, b"not-a-real-jpeg").unwrap();
+        let sender = MockSender::scripted(vec![Outcome::GroupOk], media_fetch_error);
+        let task = sequence_task(file.to_str().unwrap());
+        let result = send_media_sequence(&sender, &task).await;
+        assert!(result.is_ok(), "got {result:?}");
+
+        let sender = MockSender::scripted(vec![Outcome::CopyOk], media_fetch_error);
+        let task = Task::ForwardMessages {
+            from_chat_id: 1,
+            to_chat_id: 2,
+            message_ids: vec![3],
+            notify_chat_id: None,
+            notify_message_id: None,
+        };
+        assert!(forward_messages(&sender, &task).await.is_ok());
+    }
+
+    #[tokio::test]
+    async fn forward_classifies_retry_after_and_permanent() {
+        use teloxide::types::Seconds;
+        let task = Task::ForwardMessages {
+            from_chat_id: 1,
+            to_chat_id: 2,
+            message_ids: vec![3],
+            notify_chat_id: None,
+            notify_message_id: None,
+        };
+        // RetryAfter → Retryable with the Telegram delay.
+        let sender = MockSender::scripted(vec![Outcome::CopyErr], || {
+            RequestError::RetryAfter(Seconds::from_seconds(7))
+        });
+        match forward_messages(&sender, &task).await {
+            Err(SendError::Retryable { delay_seconds, .. }) => {
+                assert_eq!(delay_seconds, 7.0)
+            }
+            other => panic!("expected Retryable, got {other:?}"),
+        }
+        // A generic API error → Permanent.
+        let sender = MockSender::scripted(vec![Outcome::CopyErr], || {
+            RequestError::Api(ApiError::Unknown(
+                "Bad Request: message is not modified".into(),
+            ))
+        });
+        assert!(matches!(
+            forward_messages(&sender, &task).await,
+            Err(SendError::Permanent { .. })
+        ));
+    }
 }

docs/architecture-refactor.md

@@ -1,8 +1,9 @@
 # 架构优化设计：可测试性接缝 + handlers 拆分
 
-> 状态：设计稿（未实施）。目标：把仓库最大的测试空白（`handlers.rs`/`send.rs` 的
-> 发送与分派逻辑）补上可测试接缝，并把 ~1100 行的 handlers 单体拆成模块。
-> 每个阶段独立提交、独立回滚；全程 fmt / clippy / test 全绿，行为不变。
+> 状态：**阶段 A、B、C 已实施**（A: `c9e72fd`，B: `50206a9` + `ae69d72`，C:
+> rate_limit 提交）；**D 已延迟**——待下次数据库 schema 变化时实施（见 §5）。
+> 目标：把仓库最大的测试空白（`handlers.rs`/`send.rs` 的发送与分派逻辑）补上
+> 可测试接缝，并把 ~1100 行的 handlers 单体拆成模块。
 
 ---
 
@@ -74,26 +75,34 @@ impl MediaSender for Bot { /* 委托现有 teloxide 调用 */ }
 **风险**：中。动 `send.rs`/`handlers.rs` 签名（约 15 处调用点），行为不变。
 **不做**：`main.rs` 的 teloxide 装配不抽象（那是真正的胶水，无测试价值）。
 
-## 4. 阶段 C（可选）：主动限流
+## 4. 阶段 C：主动限流（已实施）
 
 批量转发时的突发会触发 Telegram 频道限速，现在靠 `RetryAfter → 队列重试` 被动
-应对。新增轻量令牌桶（`rate_limit.rs`，~50 行）：
+应对。新增轻量令牌桶（`rate_limit.rs`）：
 
 ```rust
-pub struct TokenBucket { /* capacity, refill_rate, state */ }
+pub struct TokenBucket { capacity, refill_per_sec, state: Mutex<State> }
 impl TokenBucket {
-    pub async fn acquire(&self, n: u64) -> Duration; // 等待时长（或 Notify 唤醒）
+    pub async fn acquire(&self, n: f64); // 按 n 个 token 等待并消费
 }
+pub fn limiter_for(chat_id: i64) -> Arc<TokenBucket>; // 每频道一个桶
 ```
 
-- 按频道粒度（`HashMap<ChatId, Arc<TokenBucket>>`），在 `send_media_group`/
-  `copy_messages` 前置 `acquire`。
-- 收益：减少 429 → 重试 → 死信；风险低，独立模块。
-- 不做的理由（若选不做）：当前重试链路已能自愈，容量可按需再加。
+- 默认 `CAPACITY = 20`、`REFILL_PER_SEC = 20/60`（约 20 msg/min）；
+  单次 acquire 可超出容量（记为债务，由后续 refill 偿还）。
+- 挂点：`MediaSender for Bot` 的 `send_media_group`（按 items 数）、
+  `copy_messages`（按 ids 数）、`send_animation`（1 token）前置 `acquire`；
+  MockSender 不受影响（测试不经过限流）。
+- 收益：减少 429 → 重试 → 死信；队列重试仍是全局限速的安全网。
+- 风险：低，独立模块；`tokio::time`（paused-clock 可测）。
 
-## 5. 阶段 D（可选）：DB 版本化迁移
+## 5. 阶段 D：DB 版本化迁移（**已延迟**）
 
-`schema_init` 是 `CREATE TABLE IF NOT EXISTS`，无版本概念。改为：
+> ⚠️ **待办提醒**：本阶段**推迟到下次数据库 schema 变化时实施**（给
+> `link_cache`/`chat_state`/`tasks` 加列、改结构等）。当前 `schema_init` 是
+> `CREATE TABLE IF NOT EXISTS`，无版本概念；一旦需要迁移已有线上库，必须先落地
+> 本方案（`PRAGMA user_version` 迁移链）再改 schema。`db.rs` 的 `schema_init`
+> 处已留注释指向这里。
 
 ```rust
 // db.rs
@@ -121,14 +130,14 @@ pub fn migrate(conn: &Connection) -> rusqlite::Result<()> {
 - **不引入 DI 框架**：仓库惯例是 LazyLock 静态 + 显式传参，保持。
 - **不抽象 main.rs 的 teloxide 装配**。
 
-## 7. 提交序列
+## 7. 实施记录
 
-| 阶段 | 提交消息（建议） |
-|---|---|
-| A | `refactor(handlers): split monolithic handlers.rs into modules` |
-| B | `refactor(send): introduce MediaSender seam for testable send paths` |
-| B+ | `test(send): cover fallback and classification via MockSender` |
-| C | `feat(send): add per-chat token bucket rate limiting` |
-| D | `refactor(db): versioned schema migrations` |
+| 阶段 | 提交 | 说明 |
+|---|---|---|
+| A | `c9e72fd` | handlers 拆为 `{mod, statics, commands, urls, inline, callback}` |
+| B | `50206a9` | `media_sender.rs`：`trait MediaSender` + `impl for Bot`（`<Bot as Requester>::` 消歧）；send.rs 8 处签名改 `&dyn MediaSender`；`MockSender` 测试覆盖兜底触发与错误分类（+5 测试） |
+| B | `ae69d72` | `AppContext` 注入 `url_media`（sender/store/queue/cache），url_media 全链路测试（缓存命中/失效/成功/不支持 URL，+3 测试） |
+| C | rate_limit 提交 | `rate_limit.rs` 令牌桶 + 每频道注册表；`MediaSender for Bot` 的 group/copy/animation 前置 `acquire`（+3 测试） |
+| D | — | **已延迟**：待下次数据库 schema 变化时实施（见 §5） |
 
-每阶段独立合入；A、B 为核心，C、D 可选。
+A、B、C 为核心并已实施；D 在 schema 变更时落地。