chore: bump version to 1.4.0

1.3.0 → 1.4.0: new features (DATA_DIR config, /test blockquote HTML report) plus the twitter entity-decode, post-send-actions and queue lease-heartbeat fixes.
feat(commands): wrap the /test caption in a blockquote (HTML report)
2026-09-23 23:32:05 +00:00 · 2026-08-16 17:43:54 +08:00 · 2026-08-16 17:31:54 +08:00 · 2026-08-16 17:07:33 +08:00 · 2026-08-16 17:06:28 +08:00 · 2026-08-16 17:02:05 +08:00
22 changed files with 2468 additions and 1206 deletions
@@ -4,8 +4,9 @@ name: CI
 # job that exercises the real source sites and the token-gated pixiv tests.
 #
 # Layering:
-#   test  — fmt + clippy + the full offline unit suite. Runs on every push
+#   test  — fmt + clippy + the full offline unit suite + cargo-audit
-#           and PR, including forks (it needs no secrets).
+#           dependency gate. Runs on every push and PR, including forks
 #           (it needs no secrets).
 #   live  — the #[ignore]d live-network tests plus the pixiv tests that are
 #           gated on PIXIV_REFRESH_TOKEN. Runs on schedule / manual dispatch
 #           / tag pushes only, because pull requests from forks cannot read
@@ -42,6 +43,12 @@ jobs:
        run: cargo clippy --workspace --all-targets -- -D warnings
      - name: Run offline tests
        run: cargo test --workspace
      # Dependency vulnerability gate: fails the build when a crate in
      # Cargo.lock has an unfixed security advisory. Unmaintained/unsound
      # *warnings* (dotenv, proc-macro-error2, anyhow transitive) do not fail
      # the build by default; the advisory DB is cached across runs.
      - name: Audit dependencies
        uses: actions-rust-lang/audit@v1
  live:
    needs: test
@@ -2,9 +2,9 @@
 ## Project Overview
-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`).
+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 is in Chinese; user-facing bot strings are in English. 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.4.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 and sent with HTML parse mode: raw fields are escaped, and the caption is wrapped in a `<blockquote>` so it renders exactly like the sent media caption (escaped text and links included). 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
@@ -27,14 +29,17 @@ The `x-media` library: `site::fetch(url)` dispatches through the `SITES` registr
 | Path | Purpose |
 |---|---|
 | `crates/x-media/src/` | Fetch library. `site/mod.rs` = dispatcher + `Fetched`/`FetchError`/`download_media`/`media_size`; `media.rs` = `Media` enum; `examples/fetch.rs` = end-to-end usage sample |
-| `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/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`). Twitter's `from_syndication_json` HTML-decodes the API text — syndication and GraphQL `full_text` both arrive pre-escaped (`&gt;` `&lt;` `&amp;` `&#39;`) — so the stored text is raw and the caption escapes exactly once |
 | `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/db.rs` | `DbPool`: per-store SQLite connection pool (`POOL_SIZE = 4`, WAL, busy_timeout) over `$DATA_DIR/task_queue.db` (default `data/`); `open_store` creates file + schema, `with_conn` runs all rusqlite I/O in `spawn_blocking` |
 | `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 |
 | `crates/xmedia-bot/src/send.rs` | Media senders, upload fallback, error classification, queue task handlers |
 | `crates/xmedia-bot/src/media_sender.rs` | `MediaSender` trait: the send surface (`send_media_group`/`send_animation`/`copy_messages`/`send_message`/`send_chat_action`) implemented by teloxide `Bot` (per-chat rate-limited) and by a scripted `MockSender` in tests |
 | `crates/xmedia-bot/src/rate_limit.rs` | Per-chat token bucket (`CAPACITY = 20`, ~20 msg/min refill) paced before sends reach the API so batch forwards don't trip flood control |
 ## Development Commands
@@ -53,7 +58,7 @@ Docker: `docker build -t tgxmb .` then `docker run --rm -d --name tgxmb --env-fi
 ## Code Conventions & Common Patterns
 - **Errors via `thiserror` derive** (no anyhow): the public, stringified errors — `FetchError` (`Http`/`Json`/`Pixiv`/`Site`/`NotFound`/`Blocked`) and `PixivError` — derive `thiserror::Error` with `#[from]` conversions; `Display`/`source()` come from the derive. The internal control-flow enums — `QueueError` (`Retryable { delay_seconds, payload }` / `Permanent`), `SendError` (Retryable/Permanent), `Classification`, `FallbackError` — carry no `Display` and are handled by direct variant matching. New errors should follow the same split: stringified/public errors derive `thiserror`, internal flow enums stay plain.
- **Global state via `std::sync::LazyLock` statics**, not DI: `CONFIG`, `CHAT_STORE`, `TASK_QUEUE` in `handlers.rs`; shared reqwest `CLIENT` in `x-media/src/site/mod.rs`. `Bot` is passed/cloned into handlers; queue workers share the process-wide `send::BOT` (`LazyLock<Bot>`, force-initialized in `main` so a missing token fails at startup).
+- **Global state via `std::sync::LazyLock` statics**, not DI: `CONFIG`, `CHAT_STORE`, `TASK_QUEUE` in `handlers/statics.rs`; shared reqwest `CLIENT` in `x-media/src/site/mod.rs`. `Bot` is passed/cloned into handlers; queue workers share the process-wide `send::BOT` (`LazyLock<Bot>`, force-initialized in `main` so a missing token fails at startup).
 - **Async**: tokio multi-thread runtime (`#[tokio::main]` default). All rusqlite I/O inside `tokio::task::spawn_blocking`. Long loops use `tokio::select!` with `tokio::sync::{watch, Notify}` stop/wake channels. No streams.
 - **Blocking sync primitives**: `parking_lot::Mutex` for hot caches, `tokio::sync::Mutex` for async-shared state (pixiv token cache), `AtomicBool` for feature gates.
 - **Site adapter convention**: each site module exports `PATTERN: LazyLock<Regex>`, `enabled() -> bool`, `fetch_from_url(url) -> Result<Fetched, FetchError>`, plus `cache_key`/`is_retryable`/`media_headers`, and a unit struct `<Name>Site` implementing `site::Site`; the central dispatcher (`site/mod.rs`) only iterates the `SITES` registry. Adding a site = new `site/<name>/{mod.rs,interface.rs,model.rs}` + one `Box::new(...)` entry in `SITES` — the bot crate never lists sites (SetFormat whitelist, cache-key site lookup and startup validation all derive from the registry). Async trait methods return `SiteFuture` (a boxed `Pin<Box<dyn Future + Send>>`) because `async fn` in traits is not dyn-compatible.
@@ -67,7 +72,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_DIR/task_queue.db`, default `data/` **relative to CWD**, dir auto-created); `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) |
@@ -83,18 +88,18 @@ Docker: `docker build -t tgxmb .` then `docker run --rm -d --name tgxmb --env-fi
 - **Rust, stable, edition 2024**, workspace resolver 3. No `rust-version`/MSRV pin, no `rust-toolchain.toml` — recent stable is assumed. No nightly features.
 - Package manager: **Cargo** (workspace with path dep `x-media` ← `xmedia-bot`). No `[workspace.package]`/shared deps — each crate lists deps independently.
 - **TLS is rustls end-to-end** (no native-tls/openssl in the tree, no libssl in the Docker runtime image): `teloxide` is declared `default-features = false` with `["webhooks-axum", "macros", "rustls", "ctrlc_handler"]` (the removed `default` also carried `native-tls` and `ctrlc_handler` — the latter must stay); x-media's reqwest is `default-features = false` with `["json", "rustls-tls"]` (webpki-roots baked in, so the image ships no CA bundle). One reqwest 0.12.28 in the lock.
- **Versioning**: bump the version in all three places (`crates/x-media/Cargo.toml`, `crates/xmedia-bot/Cargo.toml`, `Cargo.lock`) and **keep `README.md` and `AGENTS.md` in sync with the code on every bump**, then commit (`chore: bump version to X.Y.Z`), create an annotated tag `vX.Y.Z`, and push branch + tag (the tag push triggers the Docker Hub build).
+- **Versioning**: bump the version in all three places (`crates/x-media/Cargo.toml`, `crates/xmedia-bot/Cargo.toml`, `Cargo.lock`) and **keep `README.md`, `README.en.md` and `AGENTS.md` in sync with the code on every bump**, then commit (`chore: bump version to X.Y.Z`), create an annotated tag `vX.Y.Z`, and push branch + tag (the tag push triggers the Docker Hub build).
- Config is **environment-variable driven** (dotenv loads `.env`, gitignored; no `.env.example` exists). Key vars: `TELOXIDE_TOKEN` (required), `PIXIV_REFRESH_TOKEN`, `TWITTER_AUTH_TOKEN` (optional; x.com `auth_token` cookie — enables the logged-in GraphQL fallback that fetches NSFW tweets syndication withholds), `BOT_ADMIN` (comma-separated ids), `EDIT_MESSAGE_TTL_SECONDS` (default 86400), `LINK_CACHE_TTL_SECONDS` (default 604800), `WEBHOOK`/`WEBHOOK_URL`/`WEBHOOK_LISTEN`/`WEBHOOK_PORT`/`WEBHOOK_CERT`/`WEBHOOK_SECRET_TOKEN` (webhook mode requires URL/listen/port, `.expect`ed; `WEBHOOK_CERT` is Telegram-facing self-signed validation only — TLS must be terminated by a reverse proxy), `RUST_LOG`, `TELOXIDE_PROXY`, `LOCAL_USER_ID` (entrypoint only).
+- Config is **environment-variable driven** (dotenv loads `.env`, gitignored; no `.env.example` exists). Key vars: `TELOXIDE_TOKEN` (required), `PIXIV_REFRESH_TOKEN`, `TWITTER_AUTH_TOKEN` (optional; x.com `auth_token` cookie — enables the logged-in GraphQL fallback that fetches NSFW tweets syndication withholds), `BOT_ADMIN` (comma-separated ids), `EDIT_MESSAGE_TTL_SECONDS` (default 86400), `LINK_CACHE_TTL_SECONDS` (default 604800), `DATA_DIR` (default `data`, CWD-relative; the SQLite dir, auto-created), `WEBHOOK`/`WEBHOOK_URL`/`WEBHOOK_LISTEN`/`WEBHOOK_PORT`/`WEBHOOK_CERT`/`WEBHOOK_SECRET_TOKEN` (webhook mode requires URL/listen/port, `.expect`ed; `WEBHOOK_CERT` is Telegram-facing self-signed validation only — TLS must be terminated by a reverse proxy), `RUST_LOG`, `TELOXIDE_PROXY`, `LOCAL_USER_ID` (entrypoint only).
- SQLite via `rusqlite` with `bundled` feature (no system libsqlite needed). DB file `data/task_queue.db` is CWD-relative — run from the workspace root, or `/app` in Docker. Mount `./data` and `./cert` volumes.
+- SQLite via `rusqlite` with `bundled` feature (no system libsqlite needed). DB file `$DATA_DIR/task_queue.db` (default `data/task_queue.db`, CWD-relative — run from the workspace root, or `/app` in Docker; set `DATA_DIR` to pin state anywhere). Mount `./data` and `./cert` volumes.
 - `.gitattributes` enforces LF for `*.sh` (CRLF breaks shebangs in containers). `.gitignore`: `.env`, `data/`, `cert/`, `docker-compose.yml`, `/target`, `.idea/`.
- Docs are in Chinese; user-facing bot strings too. Keep that convention when editing captions/templates/docs.
+- Docs are in Chinese (README, AGENTS.md); user-facing bot strings are in English. Keep that split when editing user-facing strings and docs.
 ## Testing & QA
- **~80 tests, all inline `#[cfg(test)] mod tests`** — no `tests/` integration directories. Framework: built-in Rust test + `#[tokio::test]` (dev-deps only in `x-media`: tokio macros/rt-multi-thread, dotenv).
+- **~115 tests, all inline `#[cfg(test)] mod tests`** — no `tests/` integration directories. Framework: built-in Rust test + `#[tokio::test]` (dev-deps only in `x-media`: tokio macros/rt-multi-thread, dotenv).
 - No mocking framework anywhere (no mockito/wiremock/mockall). Conventions: pure-function units (regex parsing, serde round-trips, chunking, retry math) tested synchronously; async tests use real dependencies — file-backed SQLite via `tempfile` (`queue.rs::new_queue()` helper), live network fetches.
- Live-network tests exist in `site/twitter/interface.rs` (3), `site/bsky/interface.rs` (2), `site/pixiv/interface.rs`/`api.rs`. Test gating convention (enforced by `.github/workflows/ci.yml`): pure unit tests always run; live-network tests carry `#[ignore = "live network: ..."]` (run via `cargo test --workspace -- --ignored live`); token-gated pixiv tests early-return when `PIXIV_REFRESH_TOKEN` is absent **or empty** (an unset GitHub secret arrives as `""` — `is_err()` alone would run them tokenless and fail). Run the full offline suite with `cargo test --workspace`.
+- Live-network tests exist in `site/twitter/interface.rs` (5), `site/bsky/interface.rs` (2), `site/pixiv/api.rs` (1); `photo.rs` adds one `#[ignore = "heavy: …"]` test. Test gating convention (enforced by `.github/workflows/ci.yml`): pure unit tests always run; live-network tests carry `#[ignore = "live network: ..."]` (run via `cargo test --workspace -- --ignored live`); token-gated pixiv tests early-return when `PIXIV_REFRESH_TOKEN` is absent **or empty** (an unset GitHub secret arrives as `""` — `is_err()` alone would run them tokenless and fail). Run the full offline suite with `cargo test --workspace`.
 - Fixtures are inline `serde_json::json!` builder fns (`fixture()`, `thread_json()`, `illust_json()`), not files. The shared `CLIENT` sets `pool_max_idle_per_host(0)` under `#[cfg(test)]` to avoid cross-runtime `DispatchGone`.
- **CI** — `.github/workflows/ci.yml` runs `cargo fmt --check` + `cargo clippy --workspace --all-targets -- -D warnings` + `cargo test --workspace` (offline, no secrets, on every push/PR) and a `live` job (schedule/manual/tag only, `PIXIV_REFRESH_TOKEN`/`TWITTER_AUTH_TOKEN` from secrets, `continue-on-error`) for the `#[ignore]`d live + token tests. `.github/workflows/docker.yml` builds/pushes the image only.
+- **CI** — `.github/workflows/ci.yml` runs `cargo fmt --check` + `cargo clippy --workspace --all-targets -- -D warnings` + `cargo test --workspace` + a `actions-rust-lang/audit` dependency-vulnerability gate (offline, no secrets, on every push/PR) and a `live` job (schedule/manual/tag only, `PIXIV_REFRESH_TOKEN`/`TWITTER_AUTH_TOKEN` from secrets, `continue-on-error`) for the `#[ignore]`d live + token tests. `.github/workflows/docker.yml` builds/pushes the image only.
 - Untested and hard to test without a mock seam: `handlers.rs` (depends directly on teloxide `Bot`); `main.rs`, `config.rs`, `state.rs`; `media.rs`, `lib.rs`, all `model.rs`.
 - No coverage tracking.
@@ -2925,7 +2925,7 @@ checksum = "1ffae5123b2d3fc086436f8834ae3ab053a283cfac8fe0a0b8eaae044768a4c4"
 [[package]]
 name = "x-media"
-version = "1.2.2"
+version = "1.4.0"
 dependencies = [
 "bytes",
 "dotenv",
@@ -2945,7 +2945,7 @@ dependencies = [
 [[package]]
 name = "xmedia-bot"
-version = "1.2.2"
+version = "1.4.0"
 dependencies = [
 "bytes",
 "dotenv",
@@ -30,7 +30,7 @@ docker build -t tgxmb .
 docker run --rm -d --name tgxmb --env-file .env -v ./data:/app/data tgxmb
 ```
-Environment variables: `TELOXIDE_TOKEN` (required), `PIXIV_REFRESH_TOKEN`, `BOT_ADMIN`, `EDIT_MESSAGE_TTL_SECONDS`, `LINK_CACHE_TTL_SECONDS`, `RUST_LOG`, `WEBHOOK*`, `TWITTER_AUTH_TOKEN` (optional).
+Environment variables: `TELOXIDE_TOKEN` (required), `PIXIV_REFRESH_TOKEN`, `BOT_ADMIN`, `EDIT_MESSAGE_TTL_SECONDS`, `LINK_CACHE_TTL_SECONDS`, `RUST_LOG`, `TELOXIDE_PROXY`, `WEBHOOK*`, `TWITTER_AUTH_TOKEN` (optional).
 NSFW tweets: the public syndication endpoint does not return sensitive content. Setting `TWITTER_AUTH_TOKEN` (the `auth_token` cookie value of a logged-in x.com session) lets the bot fetch NSFW media in the logged-in state only when it hits a withheld tweet; without it, the bot reports no media.
@@ -84,6 +84,7 @@ Telegram only accepts ports 443/80/88/8443.
 | `BOT_ADMIN` | Admin chat IDs, comma-separated; receives start/stop notifications |
 | `EDIT_MESSAGE_TTL_SECONDS` | Edit-before-forward record expiry in seconds, default 86400 |
 | `LINK_CACHE_TTL_SECONDS` | Link-result cache expiry in seconds, default 604800 (7 days) |
 | `DATA_DIR` | Data directory (where the SQLite `task_queue.db` lives), default `data` (relative to the working directory, created automatically) |
 | `RUST_LOG` | Log level |
 | `TELOXIDE_PROXY` | HTTP proxy (e.g. `http://127.0.0.1:10808`); applies to both the Telegram Bot API and site fetches — required on restricted networks (e.g. behind the GFW) |
 | `LOCAL_USER_ID` | UID the container runs as, default 9001 |
@@ -113,6 +114,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.
@@ -30,7 +30,7 @@ docker build -t tgxmb .
 docker run --rm -d --name tgxmb --env-file .env -v ./data:/app/data tgxmb
 ```
-环境变量：`TELOXIDE_TOKEN`（必填）、`PIXIV_REFRESH_TOKEN`、`BOT_ADMIN`、`EDIT_MESSAGE_TTL_SECONDS`、`LINK_CACHE_TTL_SECONDS`、`RUST_LOG`、`WEBHOOK*`、`TWITTER_AUTH_TOKEN`（可选）。
+环境变量：`TELOXIDE_TOKEN`（必填）、`PIXIV_REFRESH_TOKEN`、`BOT_ADMIN`、`EDIT_MESSAGE_TTL_SECONDS`、`LINK_CACHE_TTL_SECONDS`、`RUST_LOG`、`TELOXIDE_PROXY`、`WEBHOOK*`、`TWITTER_AUTH_TOKEN`（可选）。
 NSFW 推文：公开的 syndication 接口不返回敏感内容。设置 `TWITTER_AUTH_TOKEN`（登录 x.com 后浏览器 Cookie 里的 `auth_token` 值）后，bot 会仅在遇到 NSFW 推文时以登录态获取媒体；未设置则提示无媒体。
@@ -84,6 +84,7 @@ Telegram 只接受 443/80/88/8443 端口。
 | `BOT_ADMIN` | 管理员聊天 ID，逗号分隔；接收启动/停止通知 |
 | `EDIT_MESSAGE_TTL_SECONDS` | 转发前编辑记录过期秒数，默认 86400 |
 | `LINK_CACHE_TTL_SECONDS` | 链接结果缓存过期秒数，默认 604800（7 天） |
 | `DATA_DIR` | 数据目录（SQLite 数据库 `task_queue.db` 所在目录），默认 `data`（相对工作目录，会自动创建） |
 | `RUST_LOG` | 日志级别 |
 | `TELOXIDE_PROXY` | HTTP 代理（如 `http://127.0.0.1:10808`）；同时作用于 Telegram Bot API 与站点抓取请求，网络受限环境（如 GFW）必需 |
 | `LOCAL_USER_ID` | 容器内运行用户 UID，默认 9001 |
@@ -113,6 +114,7 @@ Telegram 只接受 443/80/88/8443 端口。
 | `/set_format <站点> <格式>` | 自定义某站点的 caption 格式。站点：`twitter` / `bsky` / `pixiv`。占位符：`{url}` `{author}` `{author_url}` `{title}` `{tags}` |
 | `/clear_cache [链接]` | 清空链接缓存（仅管理员）；带链接只清该条，否则清空全部 |
 | `/bot_dict` | 查看当前聊天状态（调试用） |
 | `/test <链接>` | 调试：只解析链接并返回解析结果（站点、标题、作者、标签、媒体列表），不发送任何媒体 |
 链接处理仅限私聊；命令在任意聊天可用。
@@ -1,6 +1,6 @@
 [package]
 name = "x-media"
-version = "1.2.2"
+version = "1.4.0"
 edition = "2024"
 [dependencies]
@@ -1,7 +1,7 @@
 use super::model;
 use crate::media::Media;
 use crate::site::{FetchError, Fetched, Site, SiteFuture};
-use html_escape::{encode_double_quoted_attribute, encode_text};
+use html_escape::{decode_html_entities, encode_double_quoted_attribute, encode_text};
 use regex::Regex;
 use std::sync::LazyLock;
@@ -239,9 +239,17 @@ impl Tweet {
        // strip the appended media short link, mirroring FxEmbed's linkFixer
        // (no display_text_range arithmetic — see expand_links).
        let text = expand_links(&json.text, &json.entities.urls);
        // Twitter APIs (syndication AND GraphQL full_text) return the text
        // pre-escaped for HTML (`&gt;` `&lt;` `&amp;` `&#39;` …): decode it so
        // the stored text is raw. The caption's own escaping then produces
        // the rendered form exactly once — without this, `&gt;^ω^&lt;` would
        // be double-escaped to `&amp;gt;^ω^&amp;lt;` and the sent message
        // would show literal `&gt;^ω^&lt;`.
        let text = decode_html_entities(&text).into_owned();
        // `name` is the display name, `screen_name` the handle (Python's
        // vxtwitter mapping: author = display name, author_id = handle).
-        let author = json.user.name;
+        // Display names can carry the same pre-escaped entities.
        let author = decode_html_entities(&json.user.name).into_owned();
        let author_id = json.user.screen_name;
        let mut media = vec![];
        for item in json.media_details {
@@ -409,6 +417,41 @@ mod tests {
        }
    }
    #[test]
    fn syndication_text_is_unescaped_before_storing() {
        // Real API shape: the text arrives pre-escaped for HTML — e.g. the
        // tweet `>^ω^<` comes back as `&gt;^ω^&lt;` (fxtwitter's raw_text for
        // 2060196388252827954) and apostrophes as `&#39;`. Storing it raw and
        // escaping once at caption build avoids the double-escape that would
        // show literal `&gt;`/`&lt;`/`&amp;` in the sent message.
        let raw = serde_json::json!({
            "__typename": "Tweet",
            "id_str": "1",
            "text": "&gt;^ω^&lt; &amp; more &#39;quoted&#39; https://t.co/abc123",
            "user": { "name": "O&#39;Brien", "screen_name": "h" },
            "entities": { "urls": [] },
            "mediaDetails": []
        });
        let tweet = Tweet::from_syndication_json(&raw.to_string()).unwrap();
        // The appended media short link is stripped, then entities decoded.
        assert_eq!(tweet.text, ">^ω^< & more 'quoted'");
        assert_eq!(tweet.author, "O'Brien");
        let fetched: Fetched = tweet.into();
        assert_eq!(fetched.title, ">^ω^< & more 'quoted'");
        // The caption escapes the raw text exactly once (encode_text covers
        // & < >; apostrophes stay literal — they are harmless in text).
        assert!(
            fetched.caption.contains("&gt;^ω^&lt; &amp; more 'quoted'"),
            "caption: {}",
            fetched.caption
        );
        assert!(
            !fetched.caption.contains("&amp;gt;"),
            "double-escaped text: {}",
            fetched.caption
        );
    }
    #[test]
    fn cache_key_prefixes_tweet_id() {
        assert_eq!(
@@ -1,11 +1,11 @@
 [package]
 name = "xmedia-bot"
-version = "1.2.2"
+version = "1.4.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"] }
@@ -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, \
@@ -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(())
 }
@@ -0,0 +1,580 @@
 //! 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, reply_html};
 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,
                    );
                    // HTML report: the caption renders inside a <blockquote>
                    // exactly as it will appear in the sent media message.
                    reply_html(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 HTML 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. Sent with
 /// HTML parse mode: raw fields are escaped, the pre-escaped render fields are
 /// embedded as-is, and the caption is wrapped in a `<blockquote>` so it shows
 /// exactly as it will render in the sent media message. 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 {}", html_escape::encode_text(url)),
        format!("site: {site_id}"),
        format!(
            "key: {}",
            html_escape::encode_text(
                &x_media::site::cache_key(url).unwrap_or_else(|| "<unsupported>".to_string())
            )
        ),
    ];
    lines.push(format!(
        "source_url: {}",
        html_escape::encode_text(source_url)
    ));
    lines.push(format!("title: {}", html_escape::encode_text(title)));
    if let Some((author, author_url, _title, tags)) = render {
        // The render fields are already pre-escaped for HTML captions; embed
        // them as-is so the report renders them exactly like the final
        // caption. `author_url` is raw and gets escaped here.
        lines.push(format!("author: {author}"));
        lines.push(format!(
            "author_url: {}",
            html_escape::encode_text(author_url)
        ));
        lines.push(format!("tags: {tags}"));
    }
    lines.push(format!("sensitive: {sensitive}"));
    // The caption is wrapped in a <blockquote> so the report (an HTML
    // message) shows it exactly as it will render in the sent media caption
    // — escaped text and links included.
    lines.push(format!(
        "caption: <blockquote>{}</blockquote>",
        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,
            html_escape::encode_text(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_wraps_caption_in_blockquote() {
        // The report is an HTML message: raw fields are escaped, pre-escaped
        // render fields are embedded as-is, and the caption is wrapped in a
        // <blockquote> so it shows exactly as it will render in the sent
        // media caption (escaped text and links included).
        let report = test_parse_report(
            "https://x.com/u/status/1",
            "twitter",
            "https://x.com/u/status/1",
            "A & B <C>",
            Some((
                "A &amp; B",
                "https://x.com/u",
                "A &amp; B &lt;C&gt;",
                "#a &amp; #b",
            )),
            false,
            "<a href=\"https://x.com/u\">A &amp; B</a>: C &lt;D&gt; &amp; E",
            &[],
        );
        // Raw fields escaped (they render back to the original text in HTML).
        assert!(report.contains("title: A &amp; B &lt;C&gt;"), "{report}");
        assert!(
            report.contains("source_url: https://x.com/u/status/1"),
            "{report}"
        );
        // Pre-escaped render fields embedded as-is.
        assert!(report.contains("author: A &amp; B"), "{report}");
        assert!(report.contains("tags: #a &amp; #b"), "{report}");
        // Caption wrapped in a blockquote with its HTML preserved.
        assert!(
            report.contains(
                "caption: <blockquote><a href=\"https://x.com/u\">A &amp; B</a>: C &lt;D&gt; &amp; E</blockquote>"
            ),
            "{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}");
    }
 }
@@ -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)
 }
@@ -0,0 +1,164 @@
 //! 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, ReplyParameters};
 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
 }
 /// Reply to a message by id with HTML parse mode (same reply decoration as
 /// [`reply`]). Used by `/test`, whose report is an HTML message (the caption
 /// is wrapped in a `<blockquote>` to show it exactly as it will render).
 pub(crate) async fn reply_html(
    bot: &Bot,
    chat_id: i64,
    reply_to: MessageId,
    text: String,
 ) -> Result<Message, RequestError> {
    // `<Bot as Requester>::` disambiguates from the MediaSender trait's
    // same-named method (see media_sender.rs).
    <Bot as Requester>::send_message(bot, ChatId(chat_id), text)
        .parse_mode(ParseMode::Html)
        .reply_parameters(ReplyParameters::new(reply_to).allow_sending_without_reply())
        .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(())
 }
@@ -0,0 +1,38 @@
 //! 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(|| {
    let path = db_path();
    db::open_store(&path.to_string_lossy()).expect("failed to open database")
 });
 /// DB file location: `$DATA_DIR/task_queue.db` (default `data`, relative to
 /// the working directory — keeps the docker-compose `./data` mount and local
 /// runs unchanged). The directory is created if missing: SQLite does not
 /// create parent dirs, so the old hardcoded `data/task_queue.db` failed with
 /// a confusing error when started from a directory without `data/`, and a
 /// CWD-relative path is a footgun for systemd / cron deployments — `DATA_DIR`
 /// lets them pin the state anywhere.
 fn db_path() -> std::path::PathBuf {
    let dir = std::env::var("DATA_DIR").unwrap_or_else(|_| "data".to_string());
    let dir_path = std::path::Path::new(&dir);
    std::fs::create_dir_all(dir_path).expect("failed to create data directory");
    dir_path.join("task_queue.db")
 }
 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);
@@ -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"]);
    }
 }
@@ -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;
@@ -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(())
            })
        }
    }
 }
@@ -307,6 +307,11 @@ impl QueueWorker {
        }
    }
    /// Processes one leased row, keeping the lease alive while the handler
    /// runs. Without the heartbeat a task longer than [`LOCK_TTL_SECONDS`]
    /// (slow download, ugoira encode, rate-limited batch forward) would have
    /// its lease expire mid-run; the expiry sweep would flip the row back to
    /// `pending` and another worker would process it again — duplicate sends.
    async fn process(&self, row: LeasedRow) {
        let payload: Value = match serde_json::from_str(&row.payload) {
            Ok(value) => value,
@@ -318,7 +323,8 @@ impl QueueWorker {
            }
        };
        log::debug!("processing {} (attempt {})", row.id, row.attempts + 1);
-        match (self.handler)(payload).await {
+        let outcome = self.run_with_lease(&row.id, payload).await;
        match outcome {
            Ok(()) => {
                log::debug!("task {} completed", row.id);
                self.delete_row(&row.id).await;
@@ -351,6 +357,42 @@ impl QueueWorker {
        }
    }
    /// Drives the handler to completion, refreshing the row's `locked_until`
    /// every 30 s so the expiry sweep never re-leases a still-running task.
    /// The heartbeat is part of this future, not a separate spawned task: if
    /// the worker task dies (panic) the heartbeat dies with it and the sweep
    /// recovers the row exactly as before.
    async fn run_with_lease(&self, id: &str, payload: Value) -> Result<(), QueueError> {
        let fut = (self.handler)(payload);
        tokio::pin!(fut);
        let mut interval = tokio::time::interval(Duration::from_secs(30));
        // The first interval tick fires immediately; skip it (the lease was
        // just set by lease_next).
        interval.tick().await;
        let id_owned = id.to_string();
        loop {
            tokio::select! {
                result = &mut fut => return result,
                _ = interval.tick() => {
                    let now = now_f64();
                    let id = id_owned.clone();
                    let result = self
                        .pool
                        .with_conn(move |conn| {
                            conn.execute(
                                "UPDATE tasks SET locked_until=?1 WHERE id=?2 AND status='in_progress'",
                                params![now + LOCK_TTL_SECONDS, id],
                            )
                        })
                        .await;
                    if let Err(e) = result {
                        log::error!("queue lease heartbeat failed: {e}");
                    }
                }
            }
        }
    }
    async fn delete_row(&self, id: &str) {
        let id = id.to_string();
        let result = self
@@ -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()
        );
    }
 }
@@ -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
+    let result = sender
-        .send_media_group(ChatId(chat_id), items)
+        .send_media_group(ChatId(chat_id), MessageId(reply_to as i32), items)
        .reply_parameters(
            ReplyParameters::new(MessageId(reply_to as i32)).allow_sending_without_reply(),
        )
        .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
+        match sender
-            .send_media_group(ChatId(chat_id), items)
+            .send_media_group(ChatId(chat_id), MessageId(reply_to as i32), items)
            .reply_parameters(
                ReplyParameters::new(MessageId(reply_to as i32)).allow_sending_without_reply(),
            )
            .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
+    sender
-        .send_animation(ChatId(chat_id), file)
+        .send_animation(
-        .caption(caption)
+            ChatId(chat_id),
-        .parse_mode(ParseMode::Html)
+            MessageId(reply_to as i32),
-        .reply_parameters(
+            caption,
-            ReplyParameters::new(MessageId(reply_to as i32)).allow_sending_without_reply(),
+            spoiler,
-        );
+            file,
-    if spoiler {
+        )
-        request = request.has_spoiler(true);
+        .await
    }
    request.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);
+    let reply_to = message_id.map(|id| MessageId(id as i32));
-    if let Some(message_id) = message_id {
+    if let Err(e) = sender
-        request = request.reply_parameters(
+        .send_message(ChatId(chat_id), message.to_string(), reply_to, None)
-            ReplyParameters::new(MessageId(message_id as i32)).allow_sending_without_reply(),
+        .await
-        );
+    {
    }
    if let Err(e) = request.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
+        let prompt = sender
-            .send_message(ChatId(chat_id), "Reply to edit message.")
+            .send_message(
-            .reply_markup(keyboard)
+                ChatId(chat_id),
-            .reply_parameters(
+                "Reply to edit message.".to_string(),
-                ReplyParameters::new(MessageId(reply_to as i32)).allow_sending_without_reply(),
+                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}"),
@@ -1320,18 +1321,6 @@ pub async fn handle_task(payload: serde_json::Value) -> Result<(), QueueError> {
        }
    };
    let bot = BOT.clone();
    // A resumed multi-batch send already ran post_send_actions (edit prompt /
    // forward) when it first started; running them again on the resume would
    // open a duplicate edit prompt and double-forward. SendAnimation is
    // atomic (always a fresh run), so only SendMediaSequence can resume.
    let resumed = matches!(
        &task,
        Task::SendMediaSequence {
            batch_index,
            sent_message_ids,
            ..
        } if *batch_index > 0 || !sent_message_ids.is_empty()
    );
    match task {
        Task::SendMediaSequence { .. } | Task::SendAnimation { .. } => {
            let message_ids = match send_media_or_animation(&bot, &task).await {
@@ -1355,9 +1344,14 @@ pub async fn handle_task(payload: serde_json::Value) -> Result<(), QueueError> {
                    });
                }
            };
-            if !resumed {
+            // A task only reaches the queue after a failed send, so this
            // successful run is the first time post_send_actions can fire —
            // the fresh attempt failed before it ever got here. Run it
            // unconditionally: `post_send_actions` executes once, after the
            // whole sequence (every batch) completed, so the channel forward
            // and the edit-before-forward prompt must not be lost just
            // because the send needed a retry.
            post_send_actions(&bot, &task, message_ids).await;
            }
            release_keep_alive(&task);
            Ok(())
        }
@@ -1381,10 +1375,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::SendMediaSequence { .. } => send_media_sequence(sender, task).await,
-        Task::SendAnimation { .. } => send_animation(bot, task).await,
+        Task::SendAnimation { .. } => send_animation(sender, task).await,
        Task::ForwardMessages { .. } => unreachable!(),
    }
 }
@@ -1656,4 +1653,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 { .. })
        ));
    }
 }
@@ -1,8 +1,9 @@
 # 架构优化设计：可测试性接缝 + handlers 拆分
-> 状态：设计稿（未实施）。目标：把仓库最大的测试空白（`handlers.rs`/`send.rs` 的
+> 状态：**阶段 A、B、C 已实施**（A: `c9e72fd`，B: `50206a9` + `ae69d72`，C:
-> 发送与分派逻辑）补上可测试接缝，并把 ~1100 行的 handlers 单体拆成模块。
+> rate_limit 提交）；**D 已延迟**——待下次数据库 schema 变化时实施（见 §5）。
-> 每个阶段独立提交、独立回滚；全程 fmt / clippy / test 全绿，行为不变。
+> 目标：把仓库最大的测试空白（`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`/
+- 默认 `CAPACITY = 20`、`REFILL_PER_SEC = 20/60`（约 20 msg/min）；
-  `copy_messages` 前置 `acquire`。
+  单次 acquire 可超出容量（记为债务，由后续 refill 偿还）。
- 收益：减少 429 → 重试 → 死信；风险低，独立模块。
+- 挂点：`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` |
+| A | `c9e72fd` | handlers 拆为 `{mod, statics, commands, urls, inline, callback}` |
-| B | `refactor(send): introduce MediaSender seam for testable send paths` |
+| B | `50206a9` | `media_sender.rs`：`trait MediaSender` + `impl for Bot`（`<Bot as Requester>::` 消歧）；send.rs 8 处签名改 `&dyn MediaSender`；`MockSender` 测试覆盖兜底触发与错误分类（+5 测试） |
-| B+ | `test(send): cover fallback and classification via MockSender` |
+| B | `ae69d72` | `AppContext` 注入 `url_media`（sender/store/queue/cache），url_media 全链路测试（缓存命中/失效/成功/不支持 URL，+3 测试） |
-| C | `feat(send): add per-chat token bucket rate limiting` |
+| C | rate_limit 提交 | `rate_limit.rs` 令牌桶 + 每频道注册表；`MediaSender for Bot` 的 group/copy/animation 前置 `acquire`（+3 测试） |
-| D | `refactor(db): versioned schema migrations` |
+| D | — | **已延迟**：待下次数据库 schema 变化时实施（见 §5） |
-每阶段独立合入；A、B 为核心，C、D 可选。
+A、B、C 为核心并已实施；D 在 schema 变更时落地。
Author	SHA1	Message	Date
YoursFunny	3f6a0f034a	chore: bump version to 1.4.0 1.3.0 → 1.4.0: new features (DATA_DIR config, /test blockquote HTML report) plus the twitter entity-decode, post-send-actions and queue lease-heartbeat fixes.	2026-08-16 17:43:54 +08:00
YoursFunny	90a011e978	feat(commands): wrap the /test caption in a blockquote (HTML report) Replaces the strip-tags plain-text rendering: the /test reply is now an HTML message (reply_html helper with ParseMode::Html). Raw fields (url, source_url, title, author_url, media urls) are escaped, the pre-escaped render fields are embedded as-is, and the caption is wrapped in <blockquote>...</blockquote> so the report shows it exactly as it will render in the sent media caption — escaped text and clickable links included, no literal &/</> and no raw markup.	2026-08-16 17:31:54 +08:00
YoursFunny	12a065846c	feat(statics): make the SQLite path configurable via DATA_DIR The DB file was hardcoded to CWD-relative data/task_queue.db — a footgun for systemd/cron deployments and a confusing startup failure when the data/ dir did not exist (SQLite never creates parent dirs). db_path() now reads DATA_DIR (default data, CWD-relative, unchanged for local runs and the docker-compose ./data mount) and creates the directory automatically. README/README.en.md env tables and AGENTS.md document the new variable.	2026-08-16 17:07:33 +08:00
YoursFunny	dca1eff1c9	ci: add a cargo-audit dependency vulnerability gate Runs actions-rust-lang/audit after the offline tests in the test job: a crate in Cargo.lock with an unfixed security advisory fails the build. Verified locally against the current lockfile (0 vulnerabilities; the 3 warnings — unmaintained dotenv/proc-macro-error2 and transitive anyhow unsoundness — do not fail by default).	2026-08-16 17:06:28 +08:00
YoursFunny	894a9ebf4a	docs: correct the user-facing string language claim in AGENTS.md AGENTS.md claimed user-facing bot strings are Chinese, but every reply/send_message string in the code is English (Hello!, Send failed, No media found, Reply to edit message, ...). README stays Chinese; update both the overview line and the convention line to state the actual split.	2026-08-16 17:02:05 +08:00
YoursFunny	c968891ff6	fix(commands): render the /test caption as plain text The report's caption line still showed the raw HTML markup (<a href="...">...</a>). strip_html_tags now drops the tags (keeping the visible text; the links are already reported via source_url / author_url) and the remaining entity-encoded text is decoded — the strip runs on the escaped caption so a tweet text like >^ω^< survives instead of being eaten as markup. Custom-format captions contain no tags and pass through unchanged.	2026-08-16 17:01:47 +08:00
YoursFunny	0087bd01ac	fix(queue): heartbeat the lease so long tasks are not re-processed The lease was set once to now + LOCK_TTL_SECONDS (120 s) with no renewal. Tasks that legitimately take longer — slow CDN downloads, ugoira encodes, rate-limited batch forwards (a 100-message channel copy waits ~4 min on the per-chat token bucket) — had their lease expire mid-run; the 30 s expiry sweep flipped the row back to pending and another worker processed it again, double-sending. run_with_lease now drives the handler through tokio::select! and refreshes locked_until every 30 s while it runs. The heartbeat lives in the same future as the handler, so a panicking worker still lets the sweep recover the row (no leaked task keeping the lease fresh forever).	2026-08-16 17:00:38 +08:00
YoursFunny	4cb40909c5	fix(send): run post_send_actions after retried sends A task only reaches the queue after a failed send, so the fresh attempt never ran post_send_actions (edit-before-forward prompt / channel forward) — it failed before that point. The old guard skipped post_send_actions for resumed tasks (batch_index > 0 or sent ids present), which meant any send that needed a retry after partial progress silently lost its forward and edit prompt. post_send_actions is now run unconditionally on a successful queue send; it executes exactly once, after the whole sequence completed.	2026-08-16 17:00:31 +08:00
YoursFunny	f260f41755	docs: align AGENTS.md and READMEs with the current code AGENTS.md: document the twitter API entity decode and the /test report HTML-decoded display; add the missing db.rs / media_sender.rs / rate_limit.rs module rows; fix the statics location (handlers/statics.rs); refresh test counts (~115, twitter live 5, pixiv api.rs 1, photo heavy test); versioning convention now includes README.en.md. README.md / README.en.md: add TELOXIDE_PROXY to the env variable list.	2026-08-16 16:31:22 +08:00
YoursFunny	af901caddb	fix(twitter): decode API HTML entities so captions escape exactly once Twitter's syndication and GraphQL APIs return tweet text and display names pre-escaped for HTML (> < & '); the caption builder escaped the text again, so sent messages showed literal entities (e.g. >^ω^< came back as >^ω^<). from_syndication_json now decodes the API text before storing it — both the syndication path and the TWITTER_AUTH_TOKEN GraphQL fallback route through it — so the caption escapes exactly once and renders correctly. The /test report is a plain-text message but printed the pre-escaped caption and render fields; it now HTML-decodes them for display so the report shows the rendered text.	2026-08-16 16:31:16 +08:00
YoursFunny	c0af42b1cc	chore: bump version to 1.3.0	2026-08-15 15:49:40 +08:00
YoursFunny	d0810217b4	feat(commands): add /test debug command that reports link parse results only	2026-08-15 15:48:57 +08:00
YoursFunny	e6ba178983	feat(send): add per-chat token bucket rate limiting	2026-08-15 00:34:59 +08:00
YoursFunny	ae69d72930	refactor(handlers): inject AppContext into url_media; cover the full URL pipeline	2026-08-14 23:38:41 +08:00
YoursFunny	1e30815a10	docs: mark architecture refactor phases A and B implemented	2026-08-14 22:04:30 +08:00
YoursFunny	50206a9056	refactor(send): introduce MediaSender seam; add mock-based fallback tests	2026-08-14 22:04:16 +08:00
YoursFunny	c9e72fda70	refactor(handlers): split monolithic handlers.rs into modules	2026-08-14 21:50:40 +08:00