chore: bump version to 1.2.2

Deleted tweets answer the syndication endpoint with HTTP 200 and a
TweetTombstone (no `errors`, no `id_str`). The body classifier only
knew the `errors` shape, so tombstones fell through to the
`no id_str -> Sensitive` branch and degraded to an empty result,
making the bot reply "No media found" for a deleted tweet.

- extract parse_syndication_body(); tombstone shape -> NotFound
- propagate NotFound from the TWITTER_AUTH_TOKEN GraphQL fallback
  instead of swallowing it into empty_fetched
- unit tests for all body classes + live regression test on a real
  tombstoned tweet

AGENTS.md

@@ -4,7 +4,7 @@
 
 Telegram bot (teloxide) that turns post links from X/Twitter, Pixiv, and Bluesky into media messages (images, video, GIF) with the post's title, author, and tags. It supports batch media splitting, retry with persistence, inline queries, forward-channel rebinding with caption templates, and Pixiv ugoira→MP4 transcoding. README and user-facing strings are in Chinese. The project is a Rust port of a Python predecessor (see `queue.rs` comments referencing `utils/task_queue.py`).
 
-Two-crate Cargo workspace (both v1.2.0, edition 2024, resolver 3):
+Two-crate Cargo workspace (both v1.2.2, 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,15 +20,15 @@ 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.
 
-The `x-media` library: `site::fetch(url)` dispatches (in order) twitter → bsky → pixiv via per-site regex `PATTERN` and returns `Ok(None)` for unmatched URLs. `Fetched { source_url, caption, title, media: Vec<Media>, sensitive, … }`; `caption_with(format)` substitutes `{url} {author} {author_url} {title} {tags}`.
+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
 
 | 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()`, site struct, `From<SiteStruct> for Fetched`), `model.rs` (serde DTOs). Pixiv adds `api.rs` (auth + transport); twitter adds `auth.rs` (logged-in GraphQL `TweetDetail` fallback for NSFW tweets, gated on `TWITTER_AUTH_TOKEN`) |
-| `crates/xmedia-bot/src/main.rs` | Entry point: env/log init, command registration (`register_commands`), shared `send::BOT` force-init, queue worker start, pixiv validation, 300 s edit-expiry sweep, dptree handler tree, webhook vs polling dispatch |
+| `crates/x-media/src/site/<twitter\|pixiv\|bsky>/` | One directory per site: `mod.rs` (re-exports), `interface.rs` (PATTERN, `enabled()`, `fetch_from_url()`, `cache_key`/`is_retryable`/`media_headers`, unit struct `<Name>Site` implementing `site::Site`, `From<SiteStruct> for Fetched`), `model.rs` (serde DTOs). Pixiv adds `api.rs` (auth + transport); twitter adds `auth.rs` (logged-in GraphQL `TweetDetail` fallback for NSFW tweets, gated on `TWITTER_AUTH_TOKEN`) |
+| `crates/xmedia-bot/src/main.rs` | Entry point: env/log init, command registration (`register_commands`), shared `send::BOT` force-init, queue worker start, site login validation (`site::validate_all`), 300 s edit-expiry sweep, dptree handler tree, webhook vs polling dispatch |
 | `crates/xmedia-bot/src/config.rs` | Manual env parsing into `Config` |
 | `crates/xmedia-bot/src/handlers.rs` | `Command` enum (teloxide `BotCommands`), message/inline/callback handlers, URL extraction, global statics; per-URL work flows through a bounded job channel (256) drained by `URL_WORKERS = 8` workers (`start_url_workers`) — backpressure instead of unbounded spawns (teloxide's per-chat workers are sequential — batch-forwards need concurrency) |
 | `crates/xmedia-bot/src/state.rs` | `ChatStore`: parking_lot `Mutex<HashMap>` cache + SQLite write-through (`chat_state` table) |
@@ -52,15 +52,15 @@ Docker: `docker build -t tgxmb .` then `docker run --rm -d --name tgxmb --env-fi
 
 ## Code Conventions & Common Patterns
 
-- **No anyhow/thiserror.** Errors are hand-rolled enums with manual `Display`/`source()`/`From` impls: `QueueError` (`Retryable { delay_seconds, payload }` / `Permanent`), `SendError` (Retryable/Permanent), `FetchError` (`Http`/`Json`/`Pixiv`/`NotFound`/`Blocked`), `PixivError`, `Classification`. New errors should follow this pattern.
+- **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).
 - **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** (no trait, no enum dispatch — follow the existing convention): each site module exports `PATTERN: LazyLock<Regex>`, `enabled() -> bool`, `fetch_from_url(url) -> Result<Fetched, FetchError>`; `site/mod.rs` re-exports the site struct and `fetch_once` adds one guarded if-branch. Adding a site = new `site/<name>/{mod.rs,interface.rs,model.rs}` + one branch in `fetch_once`.
+- **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.
 - **Serde**: per-site `model.rs` are pure `Deserialize` DTOs mirroring API JSON; site structs in `interface.rs` have private fields, a `caption()` builder, and `impl From<SiteStruct> for Fetched`. Persisted payloads use internally-tagged enums (`#[serde(tag = "kind")]` / `type`).
 - **Naming**: module-per-concern, snake_case files, `CamelCase` types, `snake_case` fns. `//!` module docs and `///` docs on non-obvious logic (syndication token, ugoira encoding, `display_text_range`).
 - **Retries**: only `x-media::site::fetch` retries (3 attempts, `1 << attempt` backoff, HTTP errors only). Queue retries are explicit `QueueError::Retryable` with computed delay (`retry_delay_seconds`).
-- Logging via `log` macros (`pretty_env_logger`, level from `RUST_LOG`).
+- Logging via `log` macros (`pretty_env_logger`, level from `RUST_LOG`). Level convention: `info` = lifecycle + per-post business results (`sent`/`forwarded`/`copied`), admin/operator actions and anomalies (fallback, retry enqueue, dead-letter is `error`); `debug` = per-request detail (message/command/URL extraction, `fetching`/`fetched`, batch sends, queue processing, photo processing, inline queries). Full user-submitted URLs and message text only appear at `debug`; at `info` and above links are printed via the normalized cache key (`handlers::log_key`, e.g. `[key=twitter:123...]`) so logs stay short and do not echo user data.
 
 ## Important Files
 

Cargo.lock

@@ -2925,7 +2925,7 @@ checksum = "1ffae5123b2d3fc086436f8834ae3ab053a283cfac8fe0a0b8eaae044768a4c4"
 
 [[package]]
 name = "x-media"
-version = "1.2.0"
+version = "1.2.2"
 dependencies = [
  "bytes",
  "dotenv",
@@ -2937,6 +2937,7 @@ dependencies = [
  "serde",
  "serde_json",
  "tempfile",
+ "thiserror",
  "tokio",
  "url",
  "zip",
@@ -2944,7 +2945,7 @@ dependencies = [
 
 [[package]]
 name = "xmedia-bot"
-version = "1.2.0"
+version = "1.2.2"
 dependencies = [
  "bytes",
  "dotenv",

crates/x-media/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "x-media"
-version = "1.2.0"
+version = "1.2.2"
 edition = "2024"
 
 [dependencies]
@@ -13,6 +13,7 @@ url = "2.5.2"
 bytes = "1"
 zip = "2"
 tempfile = "3"
+thiserror = "2"
 rand = "0.8"
 log = "0.4"
 tokio = { version = "1.40", features = ["time"] }

crates/x-media/src/site/bsky/interface.rs

@@ -1,10 +1,31 @@
 use super::model;
 use crate::media::Media;
-use crate::site::{FetchError, Fetched};
+use crate::site::{FetchError, Fetched, Site, SiteFuture};
 use html_escape::{encode_double_quoted_attribute, encode_text};
 use regex::Regex;
 use std::sync::LazyLock;
 
+/// Registry entry for the bluesky adapter (see [`crate::site::Site`]).
+pub struct BskySite;
+
+impl Site for BskySite {
+    fn id(&self) -> &'static str {
+        "bsky"
+    }
+
+    fn pattern(&self) -> &'static Regex {
+        &PATTERN
+    }
+
+    fn cache_key(&self, url: &str) -> Option<String> {
+        cache_key(url)
+    }
+
+    fn fetch_from_url<'a>(&'a self, url: &'a str) -> SiteFuture<'a, Fetched> {
+        Box::pin(async move { fetch_from_url(url).await })
+    }
+}
+
 pub static PATTERN: LazyLock<Regex> = LazyLock::new(|| {
     Regex::new(r"^(?:https?://)?bsky\.app/profile/([\w.\-:]+)/post/([\w.\-~]+)").unwrap()
 });
@@ -59,6 +80,25 @@ pub async fn fetch_from_url(url: &str) -> Result<Fetched, FetchError> {
     Ok(fetched)
 }
 
+/// Cache key for a bsky URL: `"bsky:<handle>/<rkey>"`. The prefix is the
+/// site id used for caption-format lookup and link-cache keys.
+pub fn cache_key(url: &str) -> Option<String> {
+    PATTERN
+        .captures(url)
+        .map(|caps| format!("bsky:{}/{}", &caps[1], &caps[2]))
+}
+
+/// Bluesky's fetch-retry policy: transient classes only. Not-found, blocked
+/// and parse failures are permanent.
+pub fn is_retryable(err: &FetchError) -> bool {
+    matches!(err, FetchError::Http(_) | FetchError::Transient(_))
+}
+
+/// bsky media (cdn.bsky.app) needs no extra headers.
+pub fn media_headers(_url: &str) -> Option<Vec<(&'static str, String)>> {
+    None
+}
+
 /// Downloads an HLS playlist (master or media) and remuxes its segments to a
 /// single MP4 via ffmpeg. Returns the MP4 path plus the temp dir that must
 /// stay alive until the file is uploaded. `Ok(None)` when ffmpeg is missing.
@@ -299,6 +339,7 @@ impl From<Post> for Fetched {
             title: post.text.clone(),
             media: post.media,
             sensitive: post.sensitive,
+            site_id: "bsky",
             render_data,
             _keep_alive: None,
         }

crates/x-media/src/site/bsky/mod.rs

@@ -1,4 +1,6 @@
 mod interface;
 mod model;
 
-pub use interface::{PATTERN, Post, enabled, fetch_from_url};
+pub use interface::{
+    BskySite, PATTERN, Post, cache_key, enabled, fetch_from_url, is_retryable, media_headers,
+};

crates/x-media/src/site/mod.rs

@@ -4,11 +4,15 @@
 //! `PATTERN`, `enabled()` and `fetch_from_url()`; a future site plugs in by
 //! adding one guarded entry in [`fetch_once`].
 
-use std::fmt;
+use std::future::Future;
+use std::pin::Pin;
 use std::sync::LazyLock;
 use std::sync::atomic::{AtomicBool, Ordering};
 use std::time::Duration;
 
+use regex::Regex;
+use thiserror::Error;
+
 pub mod bsky;
 pub mod pixiv;
 pub mod twitter;
@@ -30,6 +34,10 @@ pub struct Fetched {
     pub media: Vec<crate::media::Media>,
     /// Spoiler flag for all media of this post.
     pub sensitive: bool,
+    /// Site id (`"twitter"` / `"bsky"` / `"pixiv"`): the single source of
+    /// truth for site identity — caption-format lookup, cache-key prefix and
+    /// the SetFormat whitelist all derive from it. Set by the producing site.
+    pub site_id: &'static str,
     /// Raw values (pre-escaped) for user-customizable caption formats.
     pub(crate) render_data: Option<RenderData>,
     /// Keeps temp files (e.g. an encoded ugoira MP4) alive until the caller
@@ -50,16 +58,10 @@ pub(crate) struct RenderData {
 
 impl Fetched {
     /// The site this post came from (used for per-site format overrides).
+    /// A thin alias over [`Fetched::site_id`] kept for callers that read the
+    /// site off a fetched post.
     pub fn site_name(&self) -> &'static str {
-        if self.source_url.contains("x.com") || self.source_url.contains("twitter.com") {
-            "twitter"
-        } else if self.source_url.contains("bsky.app") {
-            "bsky"
-        } else if self.source_url.contains("pixiv.net") {
-            "pixiv"
-        } else {
-            "unknown"
-        }
+        self.site_id
     }
 
     /// Renders a user-supplied caption format. The format string is
@@ -162,88 +164,64 @@ pub fn caption_from_fields(
 
 /// Stable per-post cache key derived from any supported URL, so variant
 /// domains (x.com / twitter.com / fxtwitter.com, mobile, `/photo/N`
-/// suffixes) map to the same post. Returns `"twitter:<id>"`,
-/// `"pixiv:<id>"` or `"bsky:<handle>/<rkey>"`.
+/// suffixes) map to the same post. Delegates to each registered site's
+/// `cache_key` (dispatch order twitter → bsky → pixiv).
 pub fn cache_key(url: &str) -> Option<String> {
-    if let Some(caps) = twitter::PATTERN.captures(url) {
-        return Some(format!("twitter:{}", &caps[1]));
-    }
-    if let Some(caps) = pixiv::PATTERN.captures(url) {
-        return Some(format!("pixiv:{}", &caps[1]));
-    }
-    if let Some(caps) = bsky::PATTERN.captures(url) {
-        return Some(format!("bsky:{}/{}", &caps[1], &caps[2]));
-    }
-    None
+    SITES.iter().find_map(|site| site.cache_key(url))
 }
 
-#[derive(Debug)]
+/// The site id carried by a cache key (`"twitter:123"` → `"twitter"`).
+/// Unknown prefixes fall back to `"unknown"`. The bot uses this on the
+/// link-cache hit path, where no [`Fetched`] is available — the same value
+/// a fresh fetch would read from [`Fetched::site_id`].
+pub fn site_id_from_key(key: &str) -> &'static str {
+    let prefix = key.split(':').next().unwrap_or("");
+    SITES
+        .iter()
+        .map(|site| site.id())
+        .find(|id| *id == prefix)
+        .unwrap_or("unknown")
+}
+
+#[derive(Debug, Error)]
 pub enum FetchError {
-    Http(reqwest::Error),
-    Json(serde_json::Error),
-    Pixiv(PixivError),
+    #[error("http error: {0}")]
+    Http(#[from] reqwest::Error),
+    #[error("json error: {0}")]
+    Json(#[from] serde_json::Error),
+    #[error("pixiv error: {0}")]
+    Pixiv(#[from] PixivError),
+    /// A site-specific error from a site that keeps its own error type.
+    /// Permanent by default (sites that need retryable site errors convert
+    /// them to [`FetchError::Http`] / [`FetchError::Transient`] before
+    /// returning). Pixiv predates this and keeps the dedicated
+    /// [`FetchError::Pixiv`] variant.
+    #[error("{site} error: {error}")]
+    Site {
+        site: &'static str,
+        #[source]
+        error: Box<dyn std::error::Error + Send + Sync>,
+    },
+    #[error("not found")]
     NotFound,
+    #[error("blocked")]
     Blocked,
     /// The post exists but its content is withheld (twitter NSFW /
     /// age-restricted tweets come back as an empty `{}` from syndication).
+    #[error("content withheld (sensitive)")]
     Sensitive,
     /// A download exceeded the caller's size cap (see [`download_media_limited`]).
+    #[error("media too large")]
     TooLarge,
     /// A transient server-side failure (429 / 5xx); [`fetch`] retries these.
+    #[error("transient: {0}")]
     Transient(String),
     /// A local I/O failure while streaming a download to disk
     /// (see [`download_media_to_file`]).
+    #[error("io error: {0}")]
     Io(std::io::Error),
 }
 
-impl fmt::Display for FetchError {
-    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-        match self {
-            FetchError::Http(e) => write!(f, "http error: {e}"),
-            FetchError::Json(e) => write!(f, "json error: {e}"),
-            FetchError::Pixiv(e) => write!(f, "pixiv error: {e}"),
-            FetchError::NotFound => write!(f, "not found"),
-            FetchError::Blocked => write!(f, "blocked"),
-            FetchError::Sensitive => write!(f, "content withheld (sensitive)"),
-            FetchError::TooLarge => write!(f, "media too large"),
-            FetchError::Transient(message) => write!(f, "transient: {message}"),
-            FetchError::Io(e) => write!(f, "io error: {e}"),
-        }
-    }
-}
-
-impl std::error::Error for FetchError {
-    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
-        match self {
-            FetchError::Http(e) => Some(e),
-            FetchError::Json(e) => Some(e),
-            FetchError::Pixiv(e) => Some(e),
-            FetchError::NotFound | FetchError::Blocked | FetchError::Sensitive => None,
-            FetchError::TooLarge => None,
-            FetchError::Transient(_) => None,
-            FetchError::Io(e) => Some(e),
-        }
-    }
-}
-
-impl From<reqwest::Error> for FetchError {
-    fn from(e: reqwest::Error) -> Self {
-        FetchError::Http(e)
-    }
-}
-
-impl From<serde_json::Error> for FetchError {
-    fn from(e: serde_json::Error) -> Self {
-        FetchError::Json(e)
-    }
-}
-
-impl From<PixivError> for FetchError {
-    fn from(e: PixivError) -> Self {
-        FetchError::Pixiv(e)
-    }
-}
-
 /// Shared HTTP client (browser User-Agent) for twitter/bsky fetches and
 /// [`download_media`].
 pub(crate) static CLIENT: LazyLock<reqwest::Client> = LazyLock::new(|| {
@@ -296,65 +274,158 @@ pub(crate) fn log_once_ffmpeg_missing() {
     }
 }
 
+/// Site adapter: one impl per supported site (twitter / bsky / pixiv),
+/// registered in [`SITES`]. All site-specific knowledge — URL pattern,
+/// cache-key format, fetch, retry policy, media-host headers, startup
+/// validation — lives in the site module; the central dispatcher only
+/// iterates the registry.
+///
+/// Async methods return a boxed future (see [`SiteFuture`]): `async fn` /
+/// RPITIT in traits are not dyn-compatible (verified on rustc 1.95), and
+/// `+ Send` is required since URL/queue workers spawn these futures. The
+/// site structs are stateless unit structs, so the boxed futures never
+/// borrow from `self` beyond the call's scope.
+pub trait Site: Send + Sync {
+    /// Stable site id (`"twitter"` / `"bsky"` / `"pixiv"`): caption-format
+    /// lookup, cache-key prefixes and the SetFormat whitelist derive from it.
+    fn id(&self) -> &'static str;
+    /// URL pattern; the dispatcher's first match wins (dispatch order).
+    fn pattern(&self) -> &'static Regex;
+    /// Whether the site is usable (env token present, not disabled).
+    fn enabled(&self) -> bool {
+        true
+    }
+    /// Normalized cache key for a URL of this site (`None` when the URL does
+    /// not match this site).
+    fn cache_key(&self, url: &str) -> Option<String>;
+    /// Fetches and normalizes a post.
+    fn fetch_from_url<'a>(&'a self, url: &'a str) -> SiteFuture<'a, Fetched>;
+    /// Retry policy for fetch errors: transient classes only.
+    fn is_retryable(&self, err: &FetchError) -> bool {
+        matches!(err, FetchError::Http(_) | FetchError::Transient(_))
+    }
+    /// Extra headers for downloading this site's media (hotlink protection,
+    /// e.g. pixiv's Referer for pximg.net). Matched on the media URL, not
+    /// the site pattern.
+    fn media_headers(&self, _url: &str) -> Option<Vec<(&'static str, String)>> {
+        None
+    }
+    /// Startup validation (token check etc.); failures are surfaced by
+    /// [`validate_all`]. The default is a no-op.
+    fn validate(&self) -> SiteFuture<'static, (), String> {
+        Box::pin(async { Ok(()) })
+    }
+}
+
+/// A boxed, `Send` future produced by a [`Site`] async method. Boxed so the
+/// trait stays dyn-compatible; `Send` because URL/queue workers `tokio::spawn`
+/// these futures.
+type SiteFuture<'a, T, E = FetchError> = Pin<Box<dyn Future<Output = Result<T, E>> + Send + 'a>>;
+
+/// The one registry of supported sites, in dispatch order (twitter → bsky →
+/// pixiv). Adding a site = new module + one `Box::new(...)` entry here; the
+/// bot crate never lists sites itself.
+static SITES: LazyLock<Vec<Box<dyn Site>>> = LazyLock::new(|| {
+    vec![
+        Box::new(twitter::TwitterSite),
+        Box::new(bsky::BskySite),
+        Box::new(pixiv::PixivSite),
+    ]
+});
+
+/// The first enabled site whose pattern matches `url`, in dispatch order.
+fn find_site(url: &str) -> Option<&'static dyn Site> {
+    SITES
+        .iter()
+        .find(|site| site.enabled() && site.pattern().is_match(url))
+        .map(|site| site.as_ref())
+}
+
+/// Every supported site id, in dispatch order. The bot's SetFormat whitelist
+/// derives from this list.
+pub fn site_ids() -> Vec<&'static str> {
+    SITES.iter().map(|site| site.id()).collect()
+}
+
+/// Runs every enabled site's startup validation and returns the failures
+/// (site id + message). The caller logs / notifies; failing sites disable
+/// themselves (pixiv disables on a bad token).
+pub async fn validate_all() -> Vec<(&'static str, String)> {
+    let mut failures = Vec::new();
+    for site in SITES.iter() {
+        if !site.enabled() {
+            continue;
+        }
+        if let Err(e) = site.validate().await {
+            failures.push((site.id(), e));
+        }
+    }
+    failures
+}
+
 /// Fetches a post from its URL. Returns `Ok(None)` when no site pattern
 /// matches (unsupported links are silently ignored by the bot).
 ///
-/// Transient network failures are retried: 3 total attempts with 1s then 2s
-/// delays. Retried classes: bare HTTP errors, [`FetchError::Transient`]
-/// (429/5xx from any site), and pixiv errors (its network failures arrive
-/// wrapped as `PixivError`). Non-retried: Json/NotFound/Blocked/Sensitive.
+/// Transient failures are retried: 3 total attempts with 1s then 2s delays.
+/// What counts as transient is the matched site's own policy (`is_retryable`
+/// — e.g. pixiv retries only network errors and 429/5xx). Permanent classes
+/// (not-found, blocked, sensitive, parse failures, pixiv 4xx/auth errors)
+/// are returned immediately; retrying them only wastes attempts against the
+/// source site.
 pub async fn fetch(url: &str) -> Result<Option<Fetched>, FetchError> {
+    let Some(site) = find_site(url) else {
+        return Ok(None);
+    };
     for attempt in 0..3u32 {
-        match fetch_once(url).await {
-            Ok(Some(fetched)) => {
-                log::info!(
-                    "fetched {url}: site {} returned {} media",
+        match site.fetch_from_url(url).await {
+            Ok(fetched) => {
+                // Per-request detail: debug only, keyed by the post id.
+                log::debug!(
+                    "fetched [key={}]: site {} returned {} media",
+                    cache_key(url).unwrap_or_else(|| "?".into()),
                     fetched.site_name(),
                     fetched.media.len()
                 );
                 return Ok(Some(fetched));
             }
-            Ok(None) => return Ok(None),
-            Err(e @ (FetchError::Http(_) | FetchError::Transient(_) | FetchError::Pixiv(_))) => {
-                if attempt < 2 {
+            Err(err) => {
+                if site.is_retryable(&err) && attempt < 2 {
                     tokio::time::sleep(Duration::from_secs(1 << attempt)).await;
                 } else {
-                    return Err(e);
+                    return Err(err);
                 }
             }
-            Err(other) => return Err(other),
         }
     }
     unreachable!("retry loop always returns")
 }
 
-async fn fetch_once(url: &str) -> Result<Option<Fetched>, FetchError> {
-    if twitter::enabled() && twitter::PATTERN.is_match(url) {
-        return Ok(Some(twitter::fetch_from_url(url).await?));
+/// Applies every site's media-header rule to a download request (pixiv's
+/// `Referer` for pximg.net hotlink protection). Sites contribute via their
+/// `media_headers(url)` — the central download code carries no per-site logic.
+fn apply_media_headers(mut request: reqwest::RequestBuilder, url: &str) -> reqwest::RequestBuilder {
+    for site in SITES.iter() {
+        if let Some(headers) = site.media_headers(url) {
+            for (name, value) in headers {
+                request = request.header(name, value);
+            }
+        }
     }
-    if bsky::enabled() && bsky::PATTERN.is_match(url) {
-        return Ok(Some(bsky::fetch_from_url(url).await?));
-    }
-    if pixiv::enabled() && pixiv::PATTERN.is_match(url) {
-        return Ok(Some(pixiv::fetch_from_url(url).await?));
-    }
-    Ok(None)
+    request
 }
 
 /// Downloads media bytes for the bot's upload fallback: when Telegram's own
 /// fetch of a media URL is blocked (hotlink protection), the bot downloads
-/// the file itself and uploads it via multipart. Site-appropriate headers:
-/// pixiv image hosts need the `Referer` header.
+/// the file itself and uploads it via multipart. Site-appropriate headers
+/// come from each site's `media_headers` (pixiv image hosts need `Referer`).
 /// Returns the Content-Length of a media URL, or `None` when the server does
 /// not report one. Used to check whether a file fits Telegram's size limits
 /// before downloading/uploading it.
 pub async fn media_size(url: &str) -> Result<Option<u64>, FetchError> {
-    let mut request = CLIENT.get(url);
-    let lower = url.to_ascii_lowercase();
-    if lower.contains("pximg.net") {
-        request = request.header("Referer", "https://www.pixiv.net/");
-    }
-    let response = request.send().await?.error_for_status()?;
+    let response = apply_media_headers(CLIENT.get(url), url)
+        .send()
+        .await?
+        .error_for_status()?;
     Ok(response.content_length())
 }
 
@@ -363,12 +434,10 @@ pub async fn media_size(url: &str) -> Result<Option<u64>, FetchError> {
 /// crossed (or when a declared Content-Length already exceeds it). Keeps the
 /// bot from buffering arbitrarily large bodies into memory.
 pub async fn download_media_limited(url: &str, max_bytes: u64) -> Result<bytes::Bytes, FetchError> {
-    let mut request = CLIENT.get(url);
-    let lower = url.to_ascii_lowercase();
-    if lower.contains("pximg.net") {
-        request = request.header("Referer", "https://www.pixiv.net/");
-    }
-    let response = request.send().await?.error_for_status()?;
+    let response = apply_media_headers(CLIENT.get(url), url)
+        .send()
+        .await?
+        .error_for_status()?;
     if let Some(len) = response.content_length()
         && len > max_bytes
     {
@@ -401,12 +470,10 @@ pub async fn download_media_to_file(
     out: &mut std::fs::File,
 ) -> Result<u64, FetchError> {
     use std::io::Write;
-    let mut request = CLIENT.get(url);
-    let lower = url.to_ascii_lowercase();
-    if lower.contains("pximg.net") {
-        request = request.header("Referer", "https://www.pixiv.net/");
-    }
-    let response = request.send().await?.error_for_status()?;
+    let response = apply_media_headers(CLIENT.get(url), url)
+        .send()
+        .await?
+        .error_for_status()?;
     if let Some(len) = response.content_length()
         && len > max_bytes
     {
@@ -453,6 +520,43 @@ mod tests {
         assert_eq!(cache_key("https://example.com/not-a-post"), None);
     }
 
+    #[test]
+    fn site_id_from_key_parses_prefix() {
+        assert_eq!(site_id_from_key("twitter:123"), "twitter");
+        assert_eq!(site_id_from_key("pixiv:123"), "pixiv");
+        assert_eq!(site_id_from_key("bsky:handle.example/3lorem"), "bsky");
+        assert_eq!(site_id_from_key("unknown:1"), "unknown");
+        assert_eq!(site_id_from_key("no-colon"), "unknown");
+    }
+
+    #[test]
+    fn registry_lists_all_sites_in_dispatch_order() {
+        assert_eq!(site_ids(), vec!["twitter", "bsky", "pixiv"]);
+        // Enabled sites dispatch; unsupported URLs never match.
+        assert!(find_site("https://x.com/u/status/1").is_some());
+        assert!(find_site("https://bsky.app/profile/u/post/3x").is_some());
+        assert!(find_site("https://example.com/x").is_none());
+        // Cache keys are pattern-driven, independent of the enabled() gate
+        // (pixiv is disabled in tests without PIXIV_REFRESH_TOKEN).
+        assert_eq!(
+            cache_key("https://www.pixiv.net/artworks/1"),
+            Some("pixiv:1".into())
+        );
+    }
+
+    #[test]
+    fn site_error_variant_displays_and_sources() {
+        use std::error::Error as _;
+        let err = FetchError::Site {
+            site: "example",
+            error: Box::new(std::io::Error::other("boom")),
+        };
+        assert_eq!(err.to_string(), "example error: boom");
+        assert!(err.source().is_some());
+        // Permanent by default: no site's is_retryable matches it.
+        assert!(!twitter::is_retryable(&err));
+    }
+
     #[test]
     fn caption_from_fields_substitutes_and_escapes() {
         // The format string is escaped, the field values are substituted

crates/x-media/src/site/pixiv/api.rs

@@ -8,11 +8,11 @@ use super::model::{IllustrationModel, TypeModel, UgoiraMetadataModel};
 use crate::media::Media;
 use crate::site::FetchError;
 use std::env;
-use std::fmt;
 use std::io::Read;
 use std::sync::LazyLock;
 use std::sync::atomic::{AtomicBool, Ordering};
 use std::time::{Duration, SystemTime};
+use thiserror::Error;
 
 const AUTH_TOKEN_URL: &str = "https://oauth.secure.pixiv.net/auth/token";
 const APP_API_URL: &str = "https://app-api.pixiv.net";
@@ -23,48 +23,24 @@ const APP_USER_AGENT: &str = "PixivIOSApp/7.13.3 (iOS 14.6; iPhone13,2)";
 /// Token refresh safe margin (seconds).
 const TOKEN_REFRESH_SAFE_MARGIN: u64 = 300;
 
-#[derive(Debug)]
+#[derive(Debug, Error)]
 pub enum PixivError {
     /// No refresh token available (PIXIV_REFRESH_TOKEN unset).
+    #[error("pixiv: no authentication")]
     NoAuth,
-    Http(reqwest::Error),
-    Json(serde_json::Error),
+    #[error("pixiv http error: {0}")]
+    Http(#[from] reqwest::Error),
+    #[error("pixiv json error: {0}")]
+    Json(#[from] serde_json::Error),
+    /// Non-2xx HTTP status from the app API. The code lets [`crate::site::fetch`]
+    /// retry only transient classes (429 / 5xx) instead of burning attempts on
+    /// permanent 4xx (bad token, forbidden, not found).
+    #[error("pixiv status {0}")]
+    Status(u16),
+    #[error("pixiv api error: {0}")]
     Api(String),
 }
 
-impl fmt::Display for PixivError {
-    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-        match self {
-            PixivError::NoAuth => write!(f, "pixiv: no authentication"),
-            PixivError::Http(e) => write!(f, "pixiv http error: {e}"),
-            PixivError::Json(e) => write!(f, "pixiv json error: {e}"),
-            PixivError::Api(message) => write!(f, "pixiv api error: {message}"),
-        }
-    }
-}
-
-impl std::error::Error for PixivError {
-    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
-        match self {
-            PixivError::Http(e) => Some(e),
-            PixivError::Json(e) => Some(e),
-            _ => None,
-        }
-    }
-}
-
-impl From<reqwest::Error> for PixivError {
-    fn from(e: reqwest::Error) -> Self {
-        PixivError::Http(e)
-    }
-}
-
-impl From<serde_json::Error> for PixivError {
-    fn from(e: serde_json::Error) -> Self {
-        PixivError::Json(e)
-    }
-}
-
 /// Native pixiv app-API client.
 pub struct PixivAPI {
     refresh_token: String,
@@ -137,7 +113,7 @@ impl PixivAPI {
             .send()
             .await?;
         if !response.status().is_success() {
-            return Err(PixivError::Api(format!("status {}", response.status())));
+            return Err(PixivError::Status(response.status().as_u16()));
         }
         let json: serde_json::Value = serde_json::from_str(&response.text().await?)?;
         if json.get("error").is_some() {
@@ -190,7 +166,7 @@ impl PixivAPI {
             .send()
             .await?;
         if !response.status().is_success() {
-            return Err(PixivError::Api(format!("status {}", response.status())));
+            return Err(PixivError::Status(response.status().as_u16()));
         }
         let json: serde_json::Value = serde_json::from_str(&response.text().await?)?;
         if json.get("error").is_some() {

crates/x-media/src/site/pixiv/interface.rs

@@ -1,6 +1,6 @@
 use super::model::{IllustrationModel, TypeModel};
 use crate::media::Media;
-use crate::site::{FetchError, Fetched};
+use crate::site::{FetchError, Fetched, PixivError, Site, SiteFuture};
 use html_escape::{encode_double_quoted_attribute, encode_text};
 use regex::Regex;
 use std::sync::LazyLock;
@@ -13,6 +13,53 @@ pub fn enabled() -> bool {
     super::api::enabled()
 }
 
+/// Registry entry for the pixiv adapter (see [`crate::site::Site`]).
+pub struct PixivSite;
+
+impl Site for PixivSite {
+    fn id(&self) -> &'static str {
+        "pixiv"
+    }
+
+    fn pattern(&self) -> &'static Regex {
+        &PATTERN
+    }
+
+    fn enabled(&self) -> bool {
+        enabled()
+    }
+
+    fn cache_key(&self, url: &str) -> Option<String> {
+        cache_key(url)
+    }
+
+    fn fetch_from_url<'a>(&'a self, url: &'a str) -> SiteFuture<'a, Fetched> {
+        Box::pin(async move { fetch_from_url(url).await })
+    }
+
+    fn is_retryable(&self, err: &FetchError) -> bool {
+        is_retryable(err)
+    }
+
+    fn media_headers(&self, url: &str) -> Option<Vec<(&'static str, String)>> {
+        media_headers(url)
+    }
+
+    fn validate(&self) -> SiteFuture<'static, (), String> {
+        Box::pin(async {
+            match super::api::validate().await {
+                Ok(()) => Ok(()),
+                Err(e) => {
+                    // Keep the old behavior: a failed login disables pixiv
+                    // for the rest of this process.
+                    super::api::disable();
+                    Err(format!("{e}"))
+                }
+            }
+        })
+    }
+}
+
 pub async fn fetch_from_url(url: &str) -> Result<Fetched, FetchError> {
     let id = PATTERN
         .captures(url)
@@ -23,6 +70,43 @@ pub async fn fetch_from_url(url: &str) -> Result<Fetched, FetchError> {
     Ok(super::api::fetch(id).await?.into())
 }
 
+/// Cache key for a pixiv URL: `"pixiv:<id>"`. The prefix is the site id used
+/// for caption-format lookup and link-cache keys.
+pub fn cache_key(url: &str) -> Option<String> {
+    PATTERN
+        .captures(url)
+        .map(|caps| format!("pixiv:{}", &caps[1]))
+}
+
+/// Pixiv's fetch-retry policy: transient classes only — network errors and
+/// HTTP 429/5xx. Permanent 4xx (bad/expired token, forbidden, not found),
+/// API/auth errors, unparseable bodies and missing auth are not retried.
+pub fn is_retryable(err: &FetchError) -> bool {
+    match err {
+        FetchError::Http(_) | FetchError::Transient(_) => true,
+        FetchError::Pixiv(e) => match e {
+            PixivError::Http(_) => true,
+            PixivError::Status(code) if *code == 429 || *code >= 500 => true,
+            PixivError::Status(_)
+            | PixivError::Api(_)
+            | PixivError::Json(_)
+            | PixivError::NoAuth => false,
+        },
+        _ => false,
+    }
+}
+
+/// pximg.net is hotlink-protected: downloads must carry the pixiv Referer.
+/// The match is on the media host, not the site PATTERN — pixiv's PATTERN
+/// only matches `pixiv.net/artworks/...`, never `i.pximg.net`.
+pub fn media_headers(url: &str) -> Option<Vec<(&'static str, String)>> {
+    if url.to_ascii_lowercase().contains("pximg.net") {
+        Some(vec![("Referer", "https://www.pixiv.net/".to_string())])
+    } else {
+        None
+    }
+}
+
 #[derive(Debug)]
 pub struct Illustration {
     id: String,
@@ -142,6 +226,7 @@ impl From<Illustration> for Fetched {
             title: illustration.title.clone(),
             media: illustration.media,
             sensitive: illustration.nsfw,
+            site_id: "pixiv",
             render_data,
             _keep_alive: illustration._keep_alive,
         }
@@ -232,6 +317,43 @@ mod tests {
         }
     }
 
+    #[test]
+    fn is_retryable_classifies_transient_and_permanent() {
+        // Transient: network errors, explicit transient, pixiv 429/5xx.
+        assert!(is_retryable(&FetchError::Transient("429".into())));
+        assert!(is_retryable(&FetchError::Pixiv(PixivError::Status(429))));
+        assert!(is_retryable(&FetchError::Pixiv(PixivError::Status(500))));
+        assert!(is_retryable(&FetchError::Pixiv(PixivError::Status(503))));
+        // Permanent: pixiv 4xx (bad/expired token, forbidden, not found),
+        // api/auth errors, unparseable bodies, not-found/blocked/sensitive.
+        assert!(!is_retryable(&FetchError::Pixiv(PixivError::Status(400))));
+        assert!(!is_retryable(&FetchError::Pixiv(PixivError::Status(401))));
+        assert!(!is_retryable(&FetchError::Pixiv(PixivError::Status(403))));
+        assert!(!is_retryable(&FetchError::Pixiv(PixivError::Status(404))));
+        assert!(!is_retryable(&FetchError::Pixiv(PixivError::Api(
+            "invalid_grant".into()
+        ))));
+        assert!(!is_retryable(&FetchError::Pixiv(PixivError::NoAuth)));
+        let json_err = serde_json::from_str::<serde_json::Value>("x").unwrap_err();
+        assert!(!is_retryable(&FetchError::Pixiv(PixivError::Json(
+            json_err
+        ))));
+        assert!(!is_retryable(&FetchError::NotFound));
+        assert!(!is_retryable(&FetchError::Blocked));
+        assert!(!is_retryable(&FetchError::Sensitive));
+        assert!(!is_retryable(&FetchError::TooLarge));
+    }
+
+    #[test]
+    fn media_headers_adds_referer_only_for_pximg() {
+        assert_eq!(
+            media_headers("https://i.pximg.net/img-original/img/1.png"),
+            Some(vec![("Referer", "https://www.pixiv.net/".to_string())])
+        );
+        assert_eq!(media_headers("https://www.pixiv.net/artworks/1"), None);
+        assert_eq!(media_headers("https://x.com/u/status/1"), None);
+    }
+
     #[test]
     fn ugoira_yields_empty_media() {
         let v = illust_json(

crates/x-media/src/site/pixiv/mod.rs

@@ -3,4 +3,7 @@ mod interface;
 mod model;
 
 pub use api::{PixivAPI, PixivError, disable, fetch, validate};
-pub use interface::{Illustration, PATTERN, enabled, fetch_from_url};
+pub use interface::{
+    Illustration, PATTERN, PixivSite, cache_key, enabled, fetch_from_url, is_retryable,
+    media_headers,
+};

crates/x-media/src/site/twitter/interface.rs

@@ -1,10 +1,31 @@
 use super::model;
 use crate::media::Media;
-use crate::site::{FetchError, Fetched};
+use crate::site::{FetchError, Fetched, Site, SiteFuture};
 use html_escape::{encode_double_quoted_attribute, encode_text};
 use regex::Regex;
 use std::sync::LazyLock;
 
+/// Registry entry for the twitter adapter (see [`crate::site::Site`]).
+pub struct TwitterSite;
+
+impl Site for TwitterSite {
+    fn id(&self) -> &'static str {
+        "twitter"
+    }
+
+    fn pattern(&self) -> &'static Regex {
+        &PATTERN
+    }
+
+    fn cache_key(&self, url: &str) -> Option<String> {
+        cache_key(url)
+    }
+
+    fn fetch_from_url<'a>(&'a self, url: &'a str) -> SiteFuture<'a, Fetched> {
+        Box::pin(async move { fetch_from_url(url).await })
+    }
+}
+
 pub static PATTERN: LazyLock<Regex> = LazyLock::new(|| {
     Regex::new(r"^(?:https?://)?(?:www\.|mobile\.)?(?:x|twitter|fixvx|vxtwitter|fixupx|fxtwitter)\.com/[^.]+/status/(\d+)").unwrap()
 });
@@ -29,13 +50,19 @@ pub async fn fetch_from_url(url: &str) -> Result<Fetched, FetchError> {
             if super::auth::enabled() {
                 match super::auth::fetch(id).await {
                     Ok(tweet) => Ok(tweet.into()),
+                    // The tweet is genuinely gone (deleted / suspended /
+                    // tombstoned): report it instead of degrading to an
+                    // empty result ("No media found"). Only unexpected
+                    // fallback failures (network, parse) keep the NSFW
+                    // placeholder.
+                    Err(FetchError::NotFound) => Err(FetchError::NotFound),
                     Err(e) => {
                         log::warn!("twitter auth fallback failed for {id}: {e}");
                         Ok(empty_fetched(url))
                     }
                 }
             } else {
-                log::info!("tweet {id} is sensitive; set TWITTER_AUTH_TOKEN to fetch NSFW media");
+                log::debug!("tweet {id} is sensitive; set TWITTER_AUTH_TOKEN to fetch NSFW media");
                 Ok(empty_fetched(url))
             }
         }
@@ -43,6 +70,26 @@ pub async fn fetch_from_url(url: &str) -> Result<Fetched, FetchError> {
     }
 }
 
+/// Cache key for a twitter URL: `"twitter:<id>"`. The prefix is the site id
+/// used for caption-format lookup and link-cache keys.
+pub fn cache_key(url: &str) -> Option<String> {
+    PATTERN
+        .captures(url)
+        .map(|caps| format!("twitter:{}", &caps[1]))
+}
+
+/// Twitter's fetch-retry policy: transient classes only. Not-found, blocked,
+/// sensitive (NSFW withholding) and parse failures are permanent — retrying
+/// them only wastes attempts against the syndication endpoint.
+pub fn is_retryable(err: &FetchError) -> bool {
+    matches!(err, FetchError::Http(_) | FetchError::Transient(_))
+}
+
+/// twimg URLs need no extra headers (no hotlink protection).
+pub fn media_headers(_url: &str) -> Option<Vec<(&'static str, String)>> {
+    None
+}
+
 /// A Fetched with no media for withheld tweets: the bot replies
 /// "No media found" and moves on instead of erroring.
 fn empty_fetched(url: &str) -> Fetched {
@@ -54,13 +101,15 @@ fn empty_fetched(url: &str) -> Fetched {
         title: String::new(),
         media: vec![],
         sensitive: true,
+        site_id: "twitter",
         render_data: None,
         _keep_alive: None,
     }
 }
 
 /// Fetches a tweet from the syndication endpoint. Deleted/blocked tweets
-/// surface as `FetchError::NotFound`.
+/// surface as `FetchError::NotFound`; withheld content (empty tombstone,
+/// age-restricted) as `FetchError::Sensitive`.
 pub async fn fetch(id: &str) -> Result<Tweet, FetchError> {
     let id_num = id.parse::<u64>().map_err(|_| FetchError::NotFound)?;
     let response = crate::site::CLIENT
@@ -79,23 +128,46 @@ pub async fn fetch(id: &str) -> Result<Tweet, FetchError> {
         };
     }
     let text = response.text().await?;
-    // Deleted tweets answer with {"errors": [...]} instead of a tweet.
-    if serde_json::from_str::<serde_json::Value>(&text)
-        .map(|v| v.get("errors").is_some())
-        .unwrap_or(false)
-    {
+    // Classify before parsing the tweet (see [`parse_syndication_body`]).
+    parse_syndication_body(&text)?;
+    Tweet::from_syndication_json(&text).map_err(FetchError::Json)
+}
+
+/// Parses and classifies a syndication response body. `Ok` means the body is
+/// a real tweet payload; `Err` carries the permanent error class:
+/// - `NotFound`: an `errors` array (deleted/blocked) or a `TweetTombstone`
+///   **with a reason** — "This Post was deleted by the Post author." /
+///   "This Post is from a suspended account." (the tweet is gone).
+/// - `Sensitive`: content withheld **without a deletion reason** — the empty
+///   `{}` shape or an *empty* `TweetTombstone` (`{"__typename":
+///   "TweetTombstone","tombstone":{}}`). Live tweets in restricted contexts
+///   surface this way; treating them as deleted is a regression (a normal
+///   tweet must not report "deleted"). Age-restricted tombstones route here
+///   too so the logged-in GraphQL fallback can fetch the real tweet.
+/// - `Json`: an unparseable body.
+fn parse_syndication_body(text: &str) -> Result<serde_json::Value, FetchError> {
+    let body: serde_json::Value = serde_json::from_str(text)?;
+    if body.get("errors").is_some() {
         return Err(FetchError::NotFound);
     }
-    // NSFW / age-restricted tweets exist but are served as an empty `{}` —
-    // they surface as FetchError::Sensitive so the caller can retry as a
-    // logged-in user.
-    if serde_json::from_str::<serde_json::Value>(&text)
-        .map(|v| v.get("id_str").is_none())
-        .unwrap_or(false)
-    {
+    if let Some(tombstone) = body.get("tombstone") {
+        // Only a tombstone with an explicit reason means the tweet is gone;
+        // a missing reason (empty `tombstone: {}`) or an age-restricted
+        // reason means the tweet exists but is withheld.
+        let reason = tombstone
+            .get("text")
+            .and_then(|t| t.get("text"))
+            .and_then(|t| t.as_str())
+            .unwrap_or("");
+        if reason.is_empty() || reason.to_ascii_lowercase().contains("age-restricted") {
+            return Err(FetchError::Sensitive);
+        }
+        return Err(FetchError::NotFound);
+    }
+    if body.get("id_str").is_none() {
         return Err(FetchError::Sensitive);
     }
-    Tweet::from_syndication_json(&text).map_err(FetchError::Json)
+    Ok(body)
 }
 
 /// The syndication token: JS `((id / 1e15) * PI).toString(36)` (the
@@ -285,6 +357,7 @@ impl From<Tweet> for Fetched {
             title: tweet.text.clone(),
             media: tweet.media,
             sensitive: tweet.sensitive,
+            site_id: "twitter",
             render_data,
             _keep_alive: None,
         }
@@ -336,6 +409,29 @@ mod tests {
         }
     }
 
+    #[test]
+    fn cache_key_prefixes_tweet_id() {
+        assert_eq!(
+            cache_key("https://x.com/user/status/1234567890"),
+            Some("twitter:1234567890".into())
+        );
+        assert_eq!(cache_key("https://example.com/1"), None);
+    }
+
+    #[test]
+    fn is_retryable_classifies_transient_and_permanent() {
+        // Transient: network errors and explicit transient statuses (the
+        // `Http` arm shares this match arm with `Transient`).
+        assert!(is_retryable(&FetchError::Transient("429".into())));
+        // Permanent: gone, blocked, withheld, oversized, unparseable.
+        assert!(!is_retryable(&FetchError::NotFound));
+        assert!(!is_retryable(&FetchError::Blocked));
+        assert!(!is_retryable(&FetchError::Sensitive));
+        assert!(!is_retryable(&FetchError::TooLarge));
+        let json_err = serde_json::from_str::<serde_json::Value>("x").unwrap_err();
+        assert!(!is_retryable(&FetchError::Json(json_err)));
+    }
+
     #[test]
     fn syndication_json_converts_to_fetched() {
         let raw = fixture(serde_json::json!([
@@ -572,6 +668,77 @@ mod tests {
         assert!(token.starts_with("236.v"), "got {token}");
     }
 
+    #[test]
+    fn syndication_tombstone_maps_to_not_found() {
+        // Deleted tweets answer HTTP 200 with a TweetTombstone carrying a
+        // reason (no `errors`, no `id_str`); they must not fall through to
+        // Sensitive, which would make the bot reply "No media found" for a
+        // deleted tweet.
+        let raw = serde_json::json!({
+            "__typename": "TweetTombstone",
+            "tombstone": {
+                "text": { "rtl": false, "text": "This Post was deleted by the Post author. Learn more" }
+            }
+        });
+        assert!(matches!(
+            parse_syndication_body(&raw.to_string()),
+            Err(FetchError::NotFound)
+        ));
+    }
+
+    #[test]
+    fn syndication_empty_tombstone_maps_to_sensitive() {
+        // Regression: live tweets in restricted contexts answer with an
+        // EMPTY tombstone (`{"__typename":"TweetTombstone","tombstone":{}}`)
+        // — no deletion reason. They must not be reported as deleted.
+        let raw = serde_json::json!({ "__typename": "TweetTombstone", "tombstone": {} });
+        assert!(matches!(
+            parse_syndication_body(&raw.to_string()),
+            Err(FetchError::Sensitive)
+        ));
+    }
+
+    #[test]
+    fn syndication_age_restricted_tombstone_maps_to_sensitive() {
+        // An age-restricted tombstone withholds a live tweet; route it to
+        // the logged-in fallback instead of reporting it as gone.
+        let raw = serde_json::json!({
+            "__typename": "TweetTombstone",
+            "tombstone": {
+                "text": { "rtl": false, "text": "Age-restricted adult content" }
+            }
+        });
+        assert!(matches!(
+            parse_syndication_body(&raw.to_string()),
+            Err(FetchError::Sensitive)
+        ));
+    }
+
+    #[test]
+    fn syndication_errors_maps_to_not_found() {
+        // The classic gone shape: {"errors": [...]}.
+        let raw = serde_json::json!({ "errors": [{ "message": "Couldn't find Tweet" }] });
+        assert!(matches!(
+            parse_syndication_body(&raw.to_string()),
+            Err(FetchError::NotFound)
+        ));
+    }
+
+    #[test]
+    fn syndication_empty_object_maps_to_sensitive() {
+        // NSFW / age-restricted withholding: an empty `{}`.
+        assert!(matches!(
+            parse_syndication_body("{}"),
+            Err(FetchError::Sensitive)
+        ));
+    }
+
+    #[test]
+    fn syndication_tweet_body_passes() {
+        let raw = fixture(serde_json::json!([]));
+        assert!(parse_syndication_body(&raw.to_string()).is_ok());
+    }
+
     #[tokio::test]
     #[ignore = "live network: requires outbound HTTPS to cdn.syndication.twimg.com"]
     async fn live_fetch_with_photos() {
@@ -596,4 +763,30 @@ mod tests {
             "got {result:?}"
         );
     }
+
+    #[tokio::test]
+    #[ignore = "live network: requires outbound HTTPS to cdn.syndication.twimg.com"]
+    async fn live_fetch_tombstone_deleted_tweet_is_not_found() {
+        // Regression: a real deleted tweet answering with a TweetTombstone
+        // (HTTP 200, no errors/id_str) used to surface as Sensitive and
+        // degrade to an empty result ("No media found").
+        let result = fetch("2085948045967986859").await;
+        assert!(
+            matches!(result, Err(FetchError::NotFound)),
+            "got {result:?}"
+        );
+    }
+
+    #[tokio::test]
+    #[ignore = "live network: requires outbound HTTPS to cdn.syndication.twimg.com"]
+    async fn live_fetch_empty_tombstone_is_sensitive() {
+        // Regression: a LIVE tweet (verified via a third-party API) answers
+        // syndication with an empty TweetTombstone; it must surface as
+        // Sensitive (withheld), never as NotFound (deleted).
+        let result = fetch("2087851366253555752").await;
+        assert!(
+            matches!(result, Err(FetchError::Sensitive)),
+            "got {result:?}"
+        );
+    }
 }

crates/x-media/src/site/twitter/mod.rs

@@ -2,4 +2,6 @@ mod auth;
 mod interface;
 mod model;
 
-pub use interface::{PATTERN, Tweet, enabled, fetch_from_url};
+pub use interface::{
+    PATTERN, Tweet, TwitterSite, cache_key, enabled, fetch_from_url, is_retryable, media_headers,
+};

crates/xmedia-bot/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "xmedia-bot"
-version = "1.2.0"
+version = "1.2.2"
 edition = "2024"
 
 [dependencies]

crates/xmedia-bot/src/db.rs

@@ -113,6 +113,39 @@ pub fn open_db(path: &str) -> rusqlite::Result<Connection> {
     Ok(conn)
 }
 
+/// Opens the shared DB file, runs the merged schema for all three tables and
+/// returns a pool for it. One call per process in production (the stores
+/// share the returned pool); tests call it per tempdir.
+pub fn open_store(path: &str) -> rusqlite::Result<Arc<DbPool>> {
+    if let Some(parent) = std::path::Path::new(path).parent()
+        && !parent.as_os_str().is_empty()
+    {
+        std::fs::create_dir_all(parent).map_err(rusqlite_error)?;
+    }
+    let conn = open_db(path)?;
+    schema_init(&conn)?;
+    Ok(Arc::new(DbPool::new(path)))
+}
+
+fn rusqlite_error(e: std::io::Error) -> rusqlite::Error {
+    rusqlite::Error::ToSqlConversionFailure(Box::new(e))
+}
+
+/// 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.
+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, \
+         run_after REAL NOT NULL, attempts INTEGER NOT NULL, status TEXT NOT NULL, \
+         locked_until REAL NOT NULL, created_at REAL NOT NULL); \
+         CREATE INDEX IF NOT EXISTS idx_tasks_pending ON tasks(status, run_after); \
+         CREATE TABLE IF NOT EXISTS chat_state (chat_id TEXT PRIMARY KEY, payload TEXT NOT NULL); \
+         CREATE TABLE IF NOT EXISTS link_cache (url TEXT PRIMARY KEY, payload TEXT NOT NULL, \
+         created_at REAL NOT NULL);",
+    )
+}
+
 /// Unix timestamp in fractional seconds. Shared by the queue, chat store and
 /// link cache (previously four private copies).
 pub fn now_f64() -> f64 {

crates/xmedia-bot/src/handlers.rs

@@ -1,11 +1,11 @@
 use crate::config::Config;
-use crate::db::now_f64;
+use crate::db::{self, now_f64};
 use crate::link_cache::{CachedMediaKind, CachedPost, LinkCache};
 use crate::queue::PersistentTaskQueue;
 use crate::send::{self, MediaItemPayload, Task};
 use crate::state::{ChatData, ChatStore, unix_now};
 use std::collections::HashSet;
-use std::sync::LazyLock;
+use std::sync::{Arc, LazyLock};
 use teloxide::RequestError;
 use teloxide::prelude::*;
 use teloxide::types::{
@@ -26,6 +26,10 @@ static URL_JOBS: LazyLock<parking_lot::Mutex<Option<tokio::sync::mpsc::Sender<Ur
 /// Set by main's shutdown sequence; workers stop pulling new jobs.
 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;
@@ -40,9 +44,10 @@ 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);
-        tokio::spawn(async move {
+        handles.push(tokio::spawn(async move {
             while !URL_STOP.load(std::sync::atomic::Ordering::Relaxed) {
                 let job = rx.lock().await.recv().await;
                 match job {
@@ -50,21 +55,41 @@ pub async fn start_url_workers() {
                     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;
+        }
     }
 }
 
-/// Stops URL workers (drains up to the 256 queued jobs, then exits).
-pub fn stop_url_workers() {
-    URL_STOP.store(true, std::sync::atomic::Ordering::Relaxed);
-}
+/// One shared SQLite pool for the three stores (chat state, task queue, link
+/// cache): a single pool bounds concurrent DB work on `data/task_queue.db`
+/// instead of three independent pools competing for the same file. The schema
+/// for all three tables is initialized once, here.
+static DB: LazyLock<Arc<db::DbPool>> =
+    LazyLock::new(|| db::open_store("data/task_queue.db").expect("failed to open database"));
 
-pub static CHAT_STORE: LazyLock<ChatStore> =
-    LazyLock::new(|| ChatStore::open("data/task_queue.db").expect("failed to open chat store"));
+pub static CHAT_STORE: LazyLock<ChatStore> = LazyLock::new(|| ChatStore::new(Arc::clone(&DB)));
 pub static TASK_QUEUE: LazyLock<PersistentTaskQueue> =
-    LazyLock::new(|| PersistentTaskQueue::new("data/task_queue.db"));
-pub static LINK_CACHE: LazyLock<LinkCache> =
-    LazyLock::new(|| LinkCache::open("data/task_queue.db"));
+    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);
 
 #[derive(BotCommands, Clone)]
@@ -111,6 +136,14 @@ where
         .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())
+}
+
 /// Extracts URL and text-link entities (text + caption), deduped in order.
 pub fn extract_urls(message: &Message) -> Vec<String> {
     let mut urls = Vec::new();
@@ -374,7 +407,7 @@ async fn execute_command(
                     return Ok(());
                 }
             };
-            if !["twitter", "bsky", "pixiv"].contains(&site) {
+            if !x_media::site::site_ids().contains(&site) {
                 reply(
                     bot.clone(),
                     message.clone(),
@@ -514,7 +547,11 @@ async fn dispatch_send(bot: Bot, message: &Message, task: &Task, url: &str) {
     };
     match result {
         Ok(message_ids) => {
-            log::info!("sent {} message(s) for {url}", message_ids.len());
+            log::info!(
+                "sent {} message(s) for [key={}]",
+                message_ids.len(),
+                log_key(url)
+            );
             send::post_send_actions(&bot, task, message_ids).await;
             // The task settled: drop any keep-alive temp media.
             send::release_keep_alive(task);
@@ -523,7 +560,10 @@ async fn dispatch_send(bot: Bot, message: &Message, task: &Task, url: &str) {
             delay_seconds,
             task,
         }) => {
-            log::info!("send for {url} failed, queued for retry in {delay_seconds:.1}s");
+            log::info!(
+                "send for [key={}] failed, queued for retry in {delay_seconds:.1}s",
+                log_key(url)
+            );
             enqueue_retry(task, delay_seconds).await;
             let _ = reply(bot, message.clone(), "Send failed. Task queued for retry.").await;
         }
@@ -599,9 +639,11 @@ async fn url_media(bot: Bot, message: &Message, url: &str) {
     if let Some(key) = x_media::site::cache_key(url)
         && let Some(cached) = LINK_CACHE.get(&key, CONFIG.link_cache_ttl).await
     {
-        log::info!("link cache hit for {url}");
+        log::debug!("link cache hit for {key}");
         let chat_data = CHAT_STORE.get(chat_id).await;
-        let site = key.split(':').next().unwrap_or("unknown");
+        // 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)
@@ -656,11 +698,11 @@ async fn url_media(bot: Bot, message: &Message, url: &str) {
         return;
     }
 
-    log::info!("fetching {url}");
+    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::info!("no site pattern matches {url}; ignoring");
+            log::debug!("no site pattern matches {url}; ignoring");
         }
         // Retries exhausted: notify the user (Rust-only requirement 3).
         Err(e) => {
@@ -743,7 +785,8 @@ pub async fn message_handler(bot: Bot, message: Message) -> Result<(), RequestEr
             &t[..end]
         })
         .unwrap_or("<no text>");
-    log::info!(
+    // Per-request detail: debug only (message text is user data).
+    log::debug!(
         "message from {sender} in {} (private={is_private}): {text_preview}",
         message.chat.id
     );
@@ -754,14 +797,16 @@ pub async fn message_handler(bot: Bot, message: Message) -> Result<(), RequestEr
     if let Some(text) = message.text()
         && let Ok(command) = Command::parse(text, "")
     {
-        log::info!("command from {}: {text_preview}", message.chat.id);
+        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() {
-            log::info!("extracted {} URL(s): {urls:?}", urls.len());
+            // 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
@@ -855,7 +900,11 @@ pub async fn inline_query_handler(bot: Bot, query: InlineQuery) -> Result<(), Re
 /// 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::info!("inline query: {}", query.query);
+    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();
@@ -932,7 +981,7 @@ pub async fn callback_query_handler(bot: Bot, query: CallbackQuery) -> Result<()
     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::info!(
+        log::debug!(
             "callback from {}: no edit record for prompt {prompt_message_id}",
             chat_id
         );
@@ -1008,7 +1057,7 @@ pub async fn callback_query_handler(bot: Bot, query: CallbackQuery) -> Result<()
                 }
             }
             None => {
-                log::info!("forward callback without a forward channel set");
+                log::debug!("forward callback without a forward channel set");
                 bot.answer_callback_query(callback_query_id)
                     .text("No forward channel set.")
                     .await?;

crates/xmedia-bot/src/link_cache.rs

@@ -9,8 +9,9 @@
 //! by the periodic prune in `main`.
 
 use crate::db::now_f64;
-use rusqlite::{Connection, params};
+use rusqlite::params;
 use serde::{Deserialize, Serialize};
+use std::sync::Arc;
 use std::time::Duration;
 
 #[derive(Serialize, Deserialize, Clone, Debug, PartialEq)]
@@ -45,24 +46,16 @@ pub struct CachedPost {
 }
 
 /// SQLite-backed cache sharing `data/task_queue.db` with the queue and chat
-/// state (same `open_db` pattern: busy timeout, `spawn_blocking` I/O).
+/// state (same shared pool, see [`crate::db::open_store`]).
 pub struct LinkCache {
-    pool: crate::db::DbPool,
+    pool: Arc<crate::db::DbPool>,
 }
 
 impl LinkCache {
-    pub fn open(db_path: &str) -> Self {
-        if let Ok(conn) = Connection::open(db_path)
-            && let Err(e) = conn.execute_batch(
-                "CREATE TABLE IF NOT EXISTS link_cache (url TEXT PRIMARY KEY, \
-                 payload TEXT NOT NULL, created_at REAL NOT NULL);",
-            )
-        {
-            log::error!("failed to initialize link cache schema: {e}");
-        }
-        Self {
-            pool: crate::db::DbPool::new(db_path),
-        }
+    /// Wraps the shared DB pool (the `link_cache` table lives in the merged
+    /// schema alongside `tasks` and `chat_state`).
+    pub fn new(pool: Arc<crate::db::DbPool>) -> Self {
+        LinkCache { pool }
     }
 
     /// Returns the cached post if present and not expired; a stale entry is
@@ -197,7 +190,9 @@ mod tests {
     #[tokio::test]
     async fn put_get_roundtrip() {
         let dir = tempfile::tempdir().unwrap();
-        let cache = LinkCache::open(dir.path().join("c.db").to_str().unwrap());
+        let cache = LinkCache::new(
+            crate::db::open_store(dir.path().join("c.db").to_str().unwrap()).unwrap(),
+        );
         cache.put("twitter:1", &entry()).await;
         let got = cache.get("twitter:1", Duration::from_secs(3600)).await;
         assert!(got.is_some());
@@ -209,11 +204,13 @@ mod tests {
     #[tokio::test]
     async fn expired_entry_removed_on_read() {
         let dir = tempfile::tempdir().unwrap();
-        let cache = LinkCache::open(dir.path().join("c.db").to_str().unwrap());
+        let cache = LinkCache::new(
+            crate::db::open_store(dir.path().join("c.db").to_str().unwrap()).unwrap(),
+        );
         cache.put("twitter:1", &entry()).await;
         // Force the row into the past so a 1s TTL expires it.
         {
-            let conn = Connection::open(dir.path().join("c.db")).unwrap();
+            let conn = rusqlite::Connection::open(dir.path().join("c.db")).unwrap();
             conn.execute("UPDATE link_cache SET created_at = created_at - 100", [])
                 .unwrap();
         }
@@ -234,7 +231,9 @@ mod tests {
     #[tokio::test]
     async fn remove_and_prune() {
         let dir = tempfile::tempdir().unwrap();
-        let cache = LinkCache::open(dir.path().join("c.db").to_str().unwrap());
+        let cache = LinkCache::new(
+            crate::db::open_store(dir.path().join("c.db").to_str().unwrap()).unwrap(),
+        );
         cache.put("twitter:1", &entry()).await;
         cache.put("pixiv:2", &entry()).await;
         cache.remove("twitter:1").await;
@@ -251,7 +250,7 @@ mod tests {
                 .is_some()
         );
         {
-            let conn = Connection::open(dir.path().join("c.db")).unwrap();
+            let conn = rusqlite::Connection::open(dir.path().join("c.db")).unwrap();
             conn.execute("UPDATE link_cache SET created_at = created_at - 100", [])
                 .unwrap();
         }
@@ -267,7 +266,9 @@ mod tests {
     #[tokio::test]
     async fn clear_one_entry_or_all() {
         let dir = tempfile::tempdir().unwrap();
-        let cache = LinkCache::open(dir.path().join("c.db").to_str().unwrap());
+        let cache = LinkCache::new(
+            crate::db::open_store(dir.path().join("c.db").to_str().unwrap()).unwrap(),
+        );
         cache.put("twitter:1", &entry()).await;
         cache.put("pixiv:2", &entry()).await;
         // By key: only the matching row is removed.

crates/xmedia-bot/src/main.rs

@@ -69,19 +69,18 @@ async fn main() {
     handlers::start_url_workers().await;
     log::info!("url workers started");
 
-    // Pixiv login validation (user request): a failed login notifies the
-    // admin and disables pixiv for this process.
-    if site::pixiv::enabled() {
-        match site::pixiv::validate().await {
-            Ok(()) => log::info!("pixiv login validated"),
-            Err(e) => {
-                log::error!("pixiv login failed: {e}");
-                if let Some(admin) = CONFIG.admin_ids.first() {
-                    let _ = bot
-                        .send_message(ChatId(*admin), format!("Pixiv login failed: {e}"))
-                        .await;
-                }
-                site::pixiv::disable();
+    // Site login validation (user request): a failed login notifies the
+    // admin and the site disables itself for this process (pixiv).
+    let failures = site::validate_all().await;
+    if failures.is_empty() {
+        log::info!("site logins validated");
+    } else {
+        for (site_id, message) in &failures {
+            log::error!("{site_id} login failed: {message}");
+            if let Some(admin) = CONFIG.admin_ids.first() {
+                let _ = bot
+                    .send_message(ChatId(*admin), format!("{site_id} login failed: {message}"))
+                    .await;
             }
         }
     }
@@ -185,7 +184,7 @@ async fn main() {
     const SHUTDOWN_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(30);
     let shutdown = async {
         let _ = stop_tx.send(true);
-        handlers::stop_url_workers();
+        handlers::stop_url_workers().await;
         if let Some(admin) = CONFIG.admin_ids.first() {
             let _ = bot.send_message(ChatId(*admin), "Shutting down...").await;
         }

crates/xmedia-bot/src/photo.rs

@@ -223,7 +223,7 @@ fn prepare_png(file: NamedTempFile, bytes: &[u8]) -> Result<PhotoPrep, String> {
     if w + h <= PHOTO_MAX_DIMENSION_SUM && !size_over {
         return Ok(PhotoPrep::Upload(file));
     }
-    log::info!(
+    log::debug!(
         "photo {w}x{h} ({_bit_depth:?} {color_type:?}, {} bytes) needs processing",
         bytes.len()
     );
@@ -269,7 +269,7 @@ fn prepare_png(file: NamedTempFile, bytes: &[u8]) -> Result<PhotoPrep, String> {
         let (nw, nh) = target_dims(w, h);
         pix = resize_pix(pix, w, h, nw, nh)?;
         (w, h) = (nw, nh);
-        log::info!("downscaled photo to {w}x{h} (Lanczos3)");
+        log::debug!("downscaled photo to {w}x{h} (Lanczos3)");
     }
 
     let mut png_bytes = Vec::new();
@@ -277,7 +277,7 @@ fn prepare_png(file: NamedTempFile, bytes: &[u8]) -> Result<PhotoPrep, String> {
     if png_bytes.len() as u64 <= MAX_UPLOAD_BYTES {
         return Ok(PhotoPrep::Upload(write_temp(&png_bytes, "png")?));
     }
-    log::info!("PNG still over the upload cap after processing; transcoding to JPEG");
+    log::debug!("PNG still over the upload cap after processing; transcoding to JPEG");
     let jpeg_bytes = encode_jpeg(&pix, w, h)?;
     if jpeg_bytes.len() as u64 <= MAX_UPLOAD_BYTES {
         return Ok(PhotoPrep::Upload(write_temp(&jpeg_bytes, "jpg")?));
@@ -311,7 +311,7 @@ fn prepare_jpeg(file: NamedTempFile, bytes: &[u8]) -> Result<PhotoPrep, String>
         let (nw, nh) = target_dims(w, h);
         pix = resize_pix(pix, w, h, nw, nh)?;
         (w, h) = (nw, nh);
-        log::info!("downscaled jpeg to {w}x{h} (Lanczos3)");
+        log::debug!("downscaled jpeg to {w}x{h} (Lanczos3)");
     }
     let jpeg_bytes = encode_jpeg(&pix, w, h)?;
     if jpeg_bytes.len() as u64 <= MAX_UPLOAD_BYTES {

crates/xmedia-bot/src/queue.rs

@@ -7,7 +7,7 @@
 
 use crate::db::now_f64;
 use parking_lot::Mutex;
-use rusqlite::{Connection, TransactionBehavior, params};
+use rusqlite::{TransactionBehavior, params};
 use serde_json::Value;
 use std::pin::Pin;
 use std::sync::Arc;
@@ -81,34 +81,12 @@ fn scaled_retry_delay(base: f64, attempts: i32) -> f64 {
     (base * 2f64.powi(attempts)).min(300.0)
 }
 
-fn ensure_schema(conn: &rusqlite::Connection) -> rusqlite::Result<()> {
-    conn.execute_batch(
-        "PRAGMA journal_mode=WAL; \
-         CREATE TABLE IF NOT EXISTS tasks (id TEXT PRIMARY KEY, payload TEXT NOT NULL, \
-         run_after REAL NOT NULL, attempts INTEGER NOT NULL, status TEXT NOT NULL, \
-         locked_until REAL NOT NULL, created_at REAL NOT NULL); \
-         CREATE INDEX IF NOT EXISTS idx_tasks_pending ON tasks(status, run_after);",
-    )
-}
-
 impl PersistentTaskQueue {
-    pub fn new(db_path: &str) -> Self {
-        // Ensure the parent dir and table exist even if only the queue (not
-        // ChatStore) is used — a fresh container without a mounted data dir
-        // must still be able to open the DB.
-        if let Some(parent) = std::path::Path::new(db_path).parent()
-            && !parent.as_os_str().is_empty()
-            && let Err(e) = std::fs::create_dir_all(parent)
-        {
-            log::error!("failed to create queue dir: {e}");
-        }
-        if let Ok(conn) = Connection::open(db_path)
-            && let Err(e) = ensure_schema(&conn)
-        {
-            log::error!("failed to initialize queue schema: {e}");
-        }
+    /// Wraps the shared DB pool; the schema is initialized once by
+    /// [`crate::db::open_store`] (all three stores share the pool).
+    pub fn new(pool: std::sync::Arc<crate::db::DbPool>) -> Self {
         Self {
-            pool: std::sync::Arc::new(crate::db::DbPool::new(db_path)),
+            pool,
             notify: Arc::new(Notify::new()),
             stop: Arc::new(AtomicBool::new(false)),
             worker: Mutex::new(Vec::new()),
@@ -188,7 +166,7 @@ impl PersistentTaskQueue {
             self.counter.fetch_add(1, Ordering::Relaxed)
         );
         let payload = payload.to_string();
-        log::info!("enqueued {id} (run_after {run_after:.1})");
+        log::debug!("enqueued {id} (run_after {run_after:.1})");
         self.pool.with_conn(move |conn| {
             conn.execute(
                 "INSERT OR REPLACE INTO tasks (id, payload, run_after, attempts, status, locked_until, created_at) \
@@ -339,10 +317,10 @@ impl QueueWorker {
                 return;
             }
         };
-        log::info!("processing {} (attempt {})", row.id, row.attempts + 1);
+        log::debug!("processing {} (attempt {})", row.id, row.attempts + 1);
         match (self.handler)(payload).await {
             Ok(()) => {
-                log::info!("task {} completed", row.id);
+                log::debug!("task {} completed", row.id);
                 self.delete_row(&row.id).await;
             }
             Err(QueueError::Retryable {
@@ -356,7 +334,7 @@ impl QueueWorker {
                     (self.dead_letter)(payload, message).await;
                 } else {
                     let delay = scaled_retry_delay(delay_seconds, row.attempts);
-                    log::info!(
+                    log::debug!(
                         "task {} rescheduled in {delay:.1}s (attempt {})",
                         row.id,
                         row.attempts + 1
@@ -424,7 +402,8 @@ mod tests {
     async fn new_queue() -> (PersistentTaskQueue, tempfile::TempDir) {
         let dir = tempfile::tempdir().unwrap();
         let path = dir.path().join("queue.db");
-        let queue = PersistentTaskQueue::new(path.to_str().unwrap());
+        let pool = crate::db::open_store(path.to_str().unwrap()).unwrap();
+        let queue = PersistentTaskQueue::new(pool);
         (queue, dir)
     }
 
@@ -531,10 +510,11 @@ mod tests {
     async fn stale_in_progress_row_is_recovered_on_start() {
         let dir = tempfile::tempdir().unwrap();
         let path = dir.path().join("queue.db");
-        // Insert a stale leased row directly (lease expired).
+        // Insert a stale leased row directly (lease expired). open_store runs
+        // the schema; the queue below shares the same pool.
+        let pool = crate::db::open_store(path.to_str().unwrap()).unwrap();
         {
-            let conn = Connection::open(&path).unwrap();
-            ensure_schema(&conn).unwrap();
+            let conn = rusqlite::Connection::open(&path).unwrap();
             conn.execute(
                 "INSERT INTO tasks (id, payload, run_after, attempts, status, locked_until, created_at) \
                  VALUES ('task_stale', '{\"s\":1}', 0, 0, 'in_progress', ?1, 0)",
@@ -542,7 +522,7 @@ mod tests {
             )
             .unwrap();
         }
-        let queue = PersistentTaskQueue::new(path.to_str().unwrap());
+        let queue = PersistentTaskQueue::new(pool);
         let calls = Arc::new(AtomicUsize::new(0));
         let c = calls.clone();
         queue
@@ -578,8 +558,7 @@ mod tests {
         // Insert a stale leased row AFTER startup: without a runtime sweep it
         // would stay `in_progress` forever (only start() used to recover).
         {
-            let conn = Connection::open(queue.pool.path()).unwrap();
-            ensure_schema(&conn).unwrap();
+            let conn = rusqlite::Connection::open(queue.pool.path()).unwrap();
             conn.execute(
                 "INSERT INTO tasks (id, payload, run_after, attempts, status, locked_until, created_at) \
                  VALUES ('task_stale_runtime', '{\"s\":1}', 0, 0, 'in_progress', ?1, 0)",

crates/xmedia-bot/src/send.rs

@@ -3,7 +3,7 @@
 //! URL is blocked by hotlink protection; the bot downloads the file itself
 //! and uploads it via multipart).
 
-use crate::handlers::{CHAT_STORE, LINK_CACHE, TASK_QUEUE};
+use crate::handlers::{CHAT_STORE, LINK_CACHE, TASK_QUEUE, log_key};
 use crate::link_cache::{CachedMedia, CachedMediaKind, CachedPost};
 use crate::photo::{self, MAX_UPLOAD_BYTES, PhotoPrep};
 use crate::queue::QueueError;
@@ -232,7 +232,7 @@ async fn cache_sent_task(task: &Task, media: Vec<CachedMedia>) {
     post.media = media;
     if let Some(key) = x_media::site::cache_key(&post.url) {
         LINK_CACHE.put(&key, &post).await;
-        log::info!("cached send for {}", post.url);
+        log::debug!("cached send for [key={}]", log_key(&post.url));
     }
 }
 
@@ -257,7 +257,7 @@ pub async fn invalidate_cache(task: &Task) {
         && let Some(url) = task.source_url()
         && let Some(key) = x_media::site::cache_key(url)
     {
-        log::info!("removing stale link cache entry for {url}");
+        log::debug!("removing stale link cache entry for [key={}]", log_key(url));
         LINK_CACHE.remove(&key).await;
     }
 }
@@ -399,6 +399,23 @@ fn classify_to_send_error(e: &RequestError, task: Task) -> SendError {
     }
 }
 
+impl SendError {
+    /// Attaches the (updated) task to a task-free [`FallbackError`] from the
+    /// download/upload pipeline. [`FallbackError::MediaTooLarge`] never
+    /// escapes the pipeline (it is handled by falling back to the smaller
+    /// URL), so it is unreachable here.
+    fn from_fallback(f: FallbackError, task: Task) -> SendError {
+        match f {
+            FallbackError::Retryable { delay_seconds } => SendError::Retryable {
+                delay_seconds,
+                task,
+            },
+            FallbackError::Permanent { message } => SendError::Permanent { message, task },
+            FallbackError::MediaTooLarge => unreachable!("handled inside the upload fallback"),
+        }
+    }
+}
+
 fn parse_media_url(s: &str) -> Result<url::Url, String> {
     url::Url::parse(s).map_err(|e| format!("invalid media URL: {e}"))
 }
@@ -795,7 +812,8 @@ async fn send_batch_via_upload(
     reply_to: i64,
     batch: &[MediaItemPayload],
     caption: Option<&str>,
-) -> Result<Vec<Message>, FallbackError> {
+    task: Task,
+) -> Result<Vec<Message>, SendError> {
     let sem = std::sync::Arc::new(tokio::sync::Semaphore::new(3));
     let mut set = tokio::task::JoinSet::new();
     for (i, item) in batch.iter().enumerate() {
@@ -818,10 +836,11 @@ async fn send_batch_via_upload(
             Ok(Ok(item)) => item,
             // Dropping the JoinSet aborts the remaining prep tasks; their
             // temp files are cleaned up on drop (short-circuit like before).
-            Ok(Err(e)) => return Err(e),
+            Ok(Err(e)) => return Err(SendError::from_fallback(e, task.clone())),
             Err(e) => {
-                return Err(FallbackError::Permanent {
+                return Err(SendError::Permanent {
                     message: format!("upload worker panicked: {e}"),
+                    task,
                 });
             }
         };
@@ -850,12 +869,14 @@ async fn send_batch_via_upload(
     match result {
         Ok(messages) => Ok(messages),
         Err(e) => Err(match classify_request_error(&e) {
-            Classification::Retryable { delay_seconds } => {
-                FallbackError::Retryable { delay_seconds }
-            }
-            Classification::Permanent { message } => FallbackError::Permanent { message },
-            Classification::MediaFetchFailure => FallbackError::Permanent {
+            Classification::Retryable { delay_seconds } => SendError::Retryable {
+                delay_seconds,
+                task: task.clone(),
+            },
+            Classification::Permanent { message } => SendError::Permanent { message, task },
+            Classification::MediaFetchFailure => SendError::Permanent {
                 message: "upload failed".into(),
+                task,
             },
         }),
     }
@@ -941,7 +962,7 @@ pub async fn send_media_sequence(bot: &Bot, task: &Task) -> Result<Vec<i64>, Sen
             .await
         {
             Ok(messages) => {
-                log::info!(
+                log::debug!(
                     "media group batch {idx}/{} sent ({} item(s))",
                     media_batches.len(),
                     batch.len()
@@ -952,26 +973,27 @@ pub async fn send_media_sequence(bot: &Bot, task: &Task) -> Result<Vec<i64>, Sen
             Err(RequestError::Api(api)) if is_media_fetch_failure(&api) || is_size_error(&api) => {
                 log::info!(
                     "Telegram could not fetch media for batch {idx} ({}), downloading and reuploading",
-                    batch.first().map(item_url).unwrap_or("?")
+                    batch
+                        .first()
+                        .map(item_url)
+                        .map(log_key)
+                        .unwrap_or_else(|| "?".into())
                 );
-                match send_batch_via_upload(bot, chat_id, reply_to, batch, caption).await {
+                match send_batch_via_upload(
+                    bot,
+                    chat_id,
+                    reply_to,
+                    batch,
+                    caption,
+                    updated_sequence_task(task, idx, sent.clone()),
+                )
+                .await
+                {
                     Ok(messages) => {
                         collect_file_ids(&messages, batch, &mut cached_media);
                         sent.extend(messages.into_iter().map(|m| m.id.0 as i64));
                     }
-                    Err(FallbackError::Retryable { delay_seconds }) => {
-                        return Err(SendError::Retryable {
-                            delay_seconds,
-                            task: updated_sequence_task(task, idx, sent),
-                        });
-                    }
-                    Err(FallbackError::Permanent { message }) => {
-                        return Err(SendError::Permanent {
-                            message,
-                            task: updated_sequence_task(task, idx, sent),
-                        });
-                    }
-                    Err(FallbackError::MediaTooLarge) => unreachable!("handled inside upload"),
+                    Err(e) => return Err(e),
                 }
             }
             Err(e) => {
@@ -1048,19 +1070,31 @@ pub async fn send_animation(bot: &Bot, task: &Task) -> Result<Vec<i64>, SendErro
         }
         Err(RequestError::Api(api)) if is_media_fetch_failure(&api) || is_size_error(&api) => {
             log::info!(
-                "Telegram could not fetch animation URL, downloading and reuploading: {}",
-                media_url
+                "Telegram could not fetch animation URL, downloading and reuploading: [key={}]",
+                log_key(media_url)
             );
-            match download_to_temp(animation).await {
-                Ok((file, _bytes)) => {
-                    let path = file.path().to_path_buf();
+            // Single-item local preparation — the same pipeline the media
+            // group fallback uses (download with the upload-cap check,
+            // downscale/transcode photos, smaller-URL fallback). Animations
+            // have no smaller variant, so an oversized file surfaces as a
+            // permanent error here.
+            match prepare_upload_item(animation.clone(), 0, None).await {
+                Ok(prepared) => {
+                    let PreparedItem {
+                        media, keep_alive, ..
+                    } = prepared;
+                    let InputMedia::Animation(animation) = media else {
+                        unreachable!("an Animation payload prepares to InputMedia::Animation")
+                    };
+                    // Hold the temp file until the request completes.
+                    let _keep_alive = keep_alive;
                     match send_animation_inner(
                         bot,
                         chat_id,
                         reply_to,
                         caption,
                         has_spoiler,
-                        InputFile::file(path),
+                        animation.media,
                     )
                     .await
                     {
@@ -1072,46 +1106,7 @@ pub async fn send_animation(bot: &Bot, task: &Task) -> Result<Vec<i64>, SendErro
                         Err(e) => Err(classify_to_send_error(&e, task.clone())),
                     }
                 }
-                // Over the upload cap: fall back to the smaller URL.
-                Err(FallbackError::MediaTooLarge) => match animation.fallback_url() {
-                    Some(url) => match input_file_for(url) {
-                        Ok(file) => {
-                            match send_animation_inner(
-                                bot,
-                                chat_id,
-                                reply_to,
-                                caption,
-                                has_spoiler,
-                                file,
-                            )
-                            .await
-                            {
-                                Ok(message) => {
-                                    let id = message.id.0 as i64;
-                                    cache_animation_send(task, &message).await;
-                                    Ok(vec![id])
-                                }
-                                Err(e) => Err(classify_to_send_error(&e, task.clone())),
-                            }
-                        }
-                        Err(message) => Err(SendError::Permanent {
-                            message,
-                            task: task.clone(),
-                        }),
-                    },
-                    None => Err(SendError::Permanent {
-                        message: "media too large".into(),
-                        task: task.clone(),
-                    }),
-                },
-                Err(FallbackError::Retryable { delay_seconds }) => Err(SendError::Retryable {
-                    delay_seconds,
-                    task: task.clone(),
-                }),
-                Err(FallbackError::Permanent { message }) => Err(SendError::Permanent {
-                    message,
-                    task: task.clone(),
-                }),
+                Err(e) => Err(SendError::from_fallback(e, task.clone())),
             }
         }
         Err(e) => Err(classify_to_send_error(&e, task.clone())),

crates/xmedia-bot/src/state.rs

@@ -5,7 +5,6 @@ use parking_lot::Mutex;
 use rusqlite::params;
 use serde::{Deserialize, Serialize};
 use std::collections::HashMap;
-use std::path::Path;
 use std::sync::Arc;
 use std::time::{Duration, SystemTime, UNIX_EPOCH};
 
@@ -38,7 +37,7 @@ pub struct ChatStore {
     /// Per-chat async locks serializing get→mutate→set so concurrent handler
     /// tasks (batch-forwards, callbacks) cannot clobber each other's writes.
     locks: Mutex<HashMap<i64, Arc<tokio::sync::Mutex<()>>>>,
-    pool: crate::db::DbPool,
+    pool: Arc<crate::db::DbPool>,
 }
 
 pub fn unix_now() -> i64 {
@@ -49,26 +48,15 @@ pub fn unix_now() -> i64 {
 }
 
 impl ChatStore {
-    /// Creates the parent directory and the `chat_state` table (idempotent).
-    /// The shared `tasks` / `link_cache` tables are owned by `queue.rs` and
-    /// `link_cache.rs` respectively.
-    pub fn open(path: &str) -> rusqlite::Result<Self> {
-        if let Some(parent) = Path::new(path).parent()
-            && !parent.as_os_str().is_empty()
-        {
-            std::fs::create_dir_all(parent)
-                .map_err(|e| rusqlite::Error::ToSqlConversionFailure(Box::new(e)))?;
-        }
-        let conn = crate::db::open_db(path)?;
-        conn.execute_batch(
-            "CREATE TABLE IF NOT EXISTS chat_state (chat_id TEXT PRIMARY KEY, payload TEXT NOT NULL);",
-        )?;
-        drop(conn);
-        Ok(ChatStore {
+    /// Wraps the shared DB pool (schema initialized once by
+    /// [`crate::db::open_store`]; the `chat_state` table lives in the merged
+    /// schema alongside `tasks` and `link_cache`).
+    pub fn new(pool: Arc<crate::db::DbPool>) -> Self {
+        ChatStore {
             cache: Mutex::new(HashMap::new()),
             locks: Mutex::new(HashMap::new()),
-            pool: crate::db::DbPool::new(path),
-        })
+            pool,
+        }
     }
 
     pub async fn get(&self, chat_id: i64) -> ChatData {
@@ -209,9 +197,8 @@ mod tests {
     #[tokio::test]
     async fn concurrent_updates_do_not_lose_edit_records() {
         let dir = tempfile::tempdir().unwrap();
-        let store = std::sync::Arc::new(
-            ChatStore::open(dir.path().join("s.db").to_str().unwrap()).unwrap(),
-        );
+        let pool = crate::db::open_store(dir.path().join("s.db").to_str().unwrap()).unwrap();
+        let store = std::sync::Arc::new(ChatStore::new(pool));
         let mut handles = Vec::new();
         for i in 0..4 {
             let store = Arc::clone(&store);

docs/architecture-refactor.md

@@ -0,0 +1,134 @@
+# 架构优化设计：可测试性接缝 + handlers 拆分
+
+> 状态：设计稿（未实施）。目标：把仓库最大的测试空白（`handlers.rs`/`send.rs` 的
+> 发送与分派逻辑）补上可测试接缝，并把 ~1100 行的 handlers 单体拆成模块。
+> 每个阶段独立提交、独立回滚；全程 fmt / clippy / test 全绿，行为不变。
+
+---
+
+## 1. 现状与动机
+
+- `handlers.rs`（~1100 行）混装：命令解析/执行、URL 提取 + 任务通道、inline
+  debounce、callback、edit-before-forward、全部全局静态。
+- 关键路径零测试：`url_media` 的分派、`dispatch_send` 的失败分类、缓存命中路径、
+  edit-before-forward、转发重试——AGENTS.md 自认 "untested: handlers.rs"。
+- 根因：`handlers.rs`/`send.rs` 直接依赖 teloxide `Bot`（具体类型）与全局静态
+  （`CHAT_STORE`/`TASK_QUEUE`/`LINK_CACHE`/`CONFIG`），没有注入点。
+
+## 2. 阶段 A：handlers 拆分（纯组织，零风险，先行）
+
+把 `handlers.rs` 拆为模块（仅移动代码，不改签名）：
+
+```
+handlers/
+  mod.rs       — 入口：message/inline/callback 分发 + 公共类型（UrlJob、log_key）
+  statics.rs   — CHAT_STORE / TASK_QUEUE / LINK_CACHE / DB / CONFIG / URL_JOBS
+  commands.rs  — Command enum + execute_command + set_forward_channel_handler
+  urls.rs      — extract_urls + start/stop_url_workers + url_media + build_send_task + media_to_payload
+  inline.rs    — inline_query_handler + debounce 状态机 + answer_inline_query
+  callback.rs  — callback_query_handler + edit_message_handler
+```
+
+- `mod.rs` 用 `pub use` 重导出，bot 侧引用 `handlers::xxx` 不变。
+- 收益：每个模块独立审阅；后续阶段 B 的接缝改动落在明确的模块内。
+
+## 3. 阶段 B：MediaSender 接缝（核心）
+
+**动机**：`send.rs` 的所有发送入口（`send_media_group`/`send_animation`/
+`copy_messages`）都挂在具体 `Bot` 上；测试无法注入失败/成功。
+
+**设计**：新增 `crates/xmedia-bot/src/media_sender.rs`：
+
+```rust
+/// 发送抽象：生产用 teloxide Bot，测试用记录型 mock。
+/// 方法签名与 teloxide 调用点一一对应，返回 Result 以便注入任意失败。
+pub trait MediaSender: Send + Sync {
+    fn send_media_group(&self, chat_id: ChatId, items: Vec<InputMedia>)
+        -> BoxFuture<'_, Result<Vec<Message>, RequestError>>;
+    fn send_animation(&self, chat_id: ChatId, file: InputFile, caption: Option<&str>, spoiler: bool, reply_to: i64)
+        -> BoxFuture<'_, Result<Message, RequestError>>;
+    fn copy_messages(&self, to: ChatId, from: ChatId, ids: Vec<MessageId>)
+        -> BoxFuture<'_, Result<Vec<MessageId>, RequestError>>;
+    // 按需扩展：edit_message_caption / delete_message / answer_callback_query …
+}
+
+impl MediaSender for Bot { /* 委托现有 teloxide 调用 */ }
+```
+
+配套：`ChatStore`/`LinkCache`/`PersistentTaskQueue` 已是具体类型——给 `send.rs`/
+`url_media` 需要的最小面加 trait（`ChatStoreReader`/`LinkCacheReader` 等），或直接
+注入具体类型（它们已有内存态，测试用真实 tempdir 即可，见阶段 B-注）。
+
+**接入点**：
+- `dispatch_send` / `send_media_sequence` / `send_animation` / `forward_messages` /
+  `post_send_actions` / `notify_failure` 的 `bot: &Bot` 参数改为 `sender: &dyn MediaSender`。
+- `url_media` 由 `url_media(bot, message, url)` 改为 `url_media(sender, store, queue, cache, message, url)`（或聚合为一个 `AppContext` 结构传引用）。
+
+**测试策略**（仓库无 mock 框架，手写 mock）：
+- `MockSender` 记录调用序列、按脚本返回 Ok/Err（覆盖：URL 发送成功、media-fetch
+  失败触发兜底、RetryAfter 触发入队、Permanent 触发缓存失效）。
+- `ChatStore`/`LinkCache` 用真实 tempdir 实例（现有测试已这么做）。
+- 新增测试：`send_media_sequence` 分批续传、`send_animation` 兜底、`url_media`
+  缓存命中 vs 未命中、`dispatch_send` 三分支。
+
+**风险**：中。动 `send.rs`/`handlers.rs` 签名（约 15 处调用点），行为不变。
+**不做**：`main.rs` 的 teloxide 装配不抽象（那是真正的胶水，无测试价值）。
+
+## 4. 阶段 C（可选）：主动限流
+
+批量转发时的突发会触发 Telegram 频道限速，现在靠 `RetryAfter → 队列重试` 被动
+应对。新增轻量令牌桶（`rate_limit.rs`，~50 行）：
+
+```rust
+pub struct TokenBucket { /* capacity, refill_rate, state */ }
+impl TokenBucket {
+    pub async fn acquire(&self, n: u64) -> Duration; // 等待时长（或 Notify 唤醒）
+}
+```
+
+- 按频道粒度（`HashMap<ChatId, Arc<TokenBucket>>`），在 `send_media_group`/
+  `copy_messages` 前置 `acquire`。
+- 收益：减少 429 → 重试 → 死信；风险低，独立模块。
+- 不做的理由（若选不做）：当前重试链路已能自愈，容量可按需再加。
+
+## 5. 阶段 D（可选）：DB 版本化迁移
+
+`schema_init` 是 `CREATE TABLE IF NOT EXISTS`，无版本概念。改为：
+
+```rust
+// db.rs
+const MIGRATIONS: &[&str] = &[
+    // v1: 初始 schema（tasks / chat_state / link_cache）
+    "CREATE TABLE IF NOT EXISTS tasks (...); ...",
+];
+pub fn migrate(conn: &Connection) -> rusqlite::Result<()> {
+    let v: i64 = conn.query_row("PRAGMA user_version", [], |r| r.get(0))?;
+    for (i, sql) in MIGRATIONS.iter().enumerate().skip(v as usize) {
+        conn.execute_batch(sql)?;
+        conn.pragma_update(None, "user_version", (i + 1) as i64)?;
+    }
+    Ok(())
+}
+```
+
+- 低优先级：schema 未变时无收益；将来加列/改结构时必须有。
+- `open_store` 改用 `migrate` 替换 `schema_init` 调用。
+
+## 6. 明确不做
+
+- **不拆 xmedia-core**：`Task`/队列/发送抽成独立 lib crate 是大工程，除非出现
+  第二个客户端，否则收益不抵成本。
+- **不引入 DI 框架**：仓库惯例是 LazyLock 静态 + 显式传参，保持。
+- **不抽象 main.rs 的 teloxide 装配**。
+
+## 7. 提交序列
+
+| 阶段 | 提交消息（建议） |
+|---|---|
+| A | `refactor(handlers): split monolithic handlers.rs into modules` |
+| B | `refactor(send): introduce MediaSender seam for testable send paths` |
+| B+ | `test(send): cover fallback and classification via MockSender` |
+| C | `feat(send): add per-chat token bucket rate limiting` |
+| D | `refactor(db): versioned schema migrations` |
+
+每阶段独立合入；A、B 为核心，C、D 可选。

docs/site-registry-refactor.md

@@ -0,0 +1,237 @@
+# 站点适配器重构方案：让新增站点变成"新模块 + 注册一行"
+
+> 状态：**已实施**（阶段 1-5，提交 `7ca8fd1` / `5e23916` / `bf4e615` / `5679a8c` +
+> 本文档收尾）。目标：把"加一个新站点"从改 8-9 处收敛到 3 处，并让站点身份、
+> 重试策略、下载 header 等站点能力归位到站点模块自身。实施过程中的关键偏差
+> （async 形态）见 §3 的 "async 形态" 段——原生 AFIT 实测不可用于 dyn 分派，
+> 最终采用手写 `BoxFuture`（`SiteFuture` 别名）。
+
+---
+
+## 1. 现状摩擦清单
+
+以现有三站（twitter / bsky / pixiv）为基线，新增第 4 个站点（代号 `example`）
+今天需要触碰的位置：
+
+| # | 位置（当前行号） | 改动 | 必改? |
+|---|---|---|---|
+| 1 | 新目录 `crates/x-media/src/site/example/{mod,interface,model}.rs` | 新模块 | 必改 |
+| 2 | `site/mod.rs:354-365` `fetch_once` | 加一个 `if` 分派分支 | 必改 |
+| 3 | `site/mod.rs:167-178` `cache_key` | 加一个 `if` 分支 + 约定 key 前缀 `"example:..."` | 必改 |
+| 4 | `site/mod.rs:53-63` `site_name()` | 加一个 URL `contains` 嗅探分支 | 必改 |
+| 5 | `handlers.rs:405` `SetFormat` 白名单 | `["twitter","bsky","pixiv"]` 加字符串 | 必改 |
+| 6 | `config.rs` / `main.rs:74-84` | 仿 pixiv 加启动校验（token、`disable()`） | 视站点 |
+| 7 | `site/mod.rs:312-326` `fetch_error_is_retryable` | 若重试策略特殊，改中央分类函数 | 视站点 |
+| 8 | `site/mod.rs:377/391/430` 三个下载函数 | 若媒体有防盗链，加 header（现在是硬编码 pximg 判断） | 视站点 |
+| 9 | `site/mod.rs:16,180-197` `FetchError` | 若错误类型特殊，加嵌套 variant（仿 `Pixiv(PixivError)`） | 视站点 |
+
+**根因**：仓库里没有"站点"这个实体。站点的四类能力——URL 识别（PATTERN +
+cache_key）、抓取、重试策略、下载 header——分别散落在中央 if 链、URL 字符串嗅探、
+魔法字符串 key 和 bot crate 的白名单里。`AGENTS.md` 现行约定 "no trait, no enum
+dispatch" 是刻意的简单性选择；本方案的目标是在**不推翻它精神的前提下**收敛摩擦，
+并在阶段 3 提供完整的 trait 注册表选项。
+
+## 2. 目标架构
+
+```
+crates/x-media/src/site/mod.rs
+  ├─ SITES: LazyLock<Vec<Box<dyn Site>>>      ← 注册表（唯一的"站点列表"）
+  ├─ find_site(url) / fetch(url) / cache_key(url) / site_ids()
+  └─ 通用类型：Fetched { site_id, ... } / FetchError（通用类 + Site 变体）
+        │
+        ├─ site/twitter/{mod,interface,model}.rs  impl Site
+        ├─ site/bsky/…                            impl Site
+        └─ site/pixiv/…                           impl Site  (download_headers: pximg Referer)
+                                                    (validate: token 校验)
+
+crates/xmedia-bot
+  ├─ handlers.rs  SetFormat 白名单 ← x_media::site::ids()（不再写死）
+  ├─ handlers.rs  site 格式查找 ← fetched.site_id（缓存/新鲜两条路径同口径）
+  └─ main.rs      启动校验 ← site::validate_all()（不再特判 pixiv）
+```
+
+## 3. 分阶段迁移
+
+每个阶段是一个独立提交，保持 `cargo fmt` / `cargo clippy -- -D warnings` /
+`cargo test --workspace` 全绿；行为完全不变，只挪代码、不换语义。
+
+### 阶段 1：站点身份单一来源（低风险，推荐先做）
+
+**动机**：同一概念目前有两个来源——缓存命中路径用 `key.split(':').next()`
+（`handlers.rs:639`），新鲜抓取路径用 `fetched.site_name()`（`handlers.rs:724`）；
+`site_name()` 又是对 `source_url` 的 `contains` 字符串嗅探，还有 `"unknown"`
+兜底分支。
+
+**改动**：
+
+1. `site/mod.rs`：`Fetched` 增加字段 `site_id: &'static str`（由各站点的
+   `impl From<SiteStruct> for Fetched` 填充；`empty_fetched` 同步填）。
+   `Fetched::site_name()` 改为 `return self.site_id`（保留方法名，删除
+   `source_url.contains` 嗅探与 `"unknown"` 分支）。
+2. `site/mod.rs`：新增 `pub fn site_id_from_key(key: &str) -> &'static str`
+   （解析 `"example:..."` 前缀，未知前缀返回 `"unknown"`），bot 缓存命中路径改用它，
+   与 `fetched.site_id` 口径统一。
+3. `handlers.rs:405`：`SetFormat` 白名单改为 `x_media::site::ids()`——阶段 1 先实现
+   `ids()` 为 `["twitter","bsky","pixiv"]` 的常量函数（数据源仍集中，行为不变），
+   阶段 3 再改为遍历注册表。
+4. `twitter/interface.rs:48-60` / `bsky` / `pixiv` 的 `From<SiteStruct> for Fetched`
+   各补 `site_id` 字段。
+
+**风险**：低。纯增量字段；`site_name()` 语义不变（测试 `pixiv/interface.rs:355`
+  已断言 `"pixiv"`）。
+**验证**：现有全部单测；`cache_key_normalizes_domain_variants` 等不变。
+**回滚**：revert 该提交。
+
+### 阶段 2：站点能力下沉（不引入 trait，静态分派）
+
+**动机**：把"每个站点自己才知道"的逻辑搬回站点模块，中央只做迭代。这是
+`AGENTS.md` 现有约定（无 trait）与完整注册表之间的折中，可独立交付。
+
+**改动**：每个站点模块新增并 `mod.rs` 重新导出：
+
+```rust
+// site/twitter/interface.rs（bsky/pixiv 同构）
+pub fn cache_key(url: &str) -> Option<String>;   // 用自身 PATTERN，返回 "twitter:<id>"
+pub fn is_retryable(err: &FetchError) -> bool;   // 默认 Http|Transient；pixiv 覆盖 PixivError 分支
+pub fn media_headers(url: &str) -> Option<Vec<(&'static str, String)>>;
+                                                 // pixiv: url 含 "pximg.net" → Referer
+```
+
+`site/mod.rs` 相应改为迭代三站：
+
+- `cache_key`：逐个调 `site::cache_key`，不再自己写 key 格式；
+- `fetch_error_is_retryable`：删除，`fetch()` 重试循环改调 `current_site::is_retryable`
+  （`fetch_once` 已能确定站点，把站点传下去）；
+- `media_size` / `download_media_limited` / `download_media_to_file` 里的
+  `pximg.net → Referer` 硬编码删除，改为遍历 `SITES`（阶段 2 是遍历
+  `[twitter, bsky, pixiv]` 静态列表）取 `media_headers(url)` 合并。
+
+**注意**：Referer 判定依据是媒体 URL 的 host（`pximg.net`），**不是**站点
+PATTERN（pixiv 的 PATTERN 只匹配 `pixiv.net/artworks/...`），所以 `media_headers`
+不能挂在 PATTERN 匹配上，必须按 URL 独立匹配——这正是把它做成独立函数的原因。
+
+**风险**：中。下载函数签名不变，行为必须逐字节不变；新增单元测试覆盖
+`media_headers("https://i.pximg.net/...") == Some(Referer)` 与
+`cache_key` 等价性（对全部既有用例断言新旧结果一致）。
+**回滚**：revert。
+
+### 阶段 3：Site trait + SITES 注册表（完整方案，可选）
+
+**动机**：加站点时 bot crate 与中央分派零改动；站点列表成为唯一注册点。
+
+**新增**（`site/mod.rs`，按实施后的实际形态）：
+
+```rust
+/// Boxed, Send future produced by a Site async method. Boxed so the trait
+/// stays dyn-compatible; Send because URL/queue workers tokio::spawn these.
+type SiteFuture<'a, T, E = FetchError> =
+    Pin<Box<dyn Future<Output = Result<T, E>> + Send + 'a>>;
+
+pub trait Site: Send + Sync {
+    fn id(&self) -> &'static str;
+    fn pattern(&self) -> &'static Regex;
+    fn enabled(&self) -> bool { true }                 // 默认: true
+    fn cache_key(&self, url: &str) -> Option<String>;
+    fn fetch_from_url<'a>(&'a self, url: &'a str) -> SiteFuture<'a, Fetched>;
+    fn is_retryable(&self, err: &FetchError) -> bool;  // 默认: Http|Transient
+    fn media_headers(&self, url: &str) -> Option<Vec<(&'static str, String)>>; // 默认: None
+    fn validate(&self) -> SiteFuture<'static, (), String>;  // 默认: Ok(())
+}
+
+static SITES: LazyLock<Vec<Box<dyn Site>>> = LazyLock::new(|| vec![
+    Box::new(twitter::TwitterSite), Box::new(bsky::BskySite), Box::new(pixiv::PixivSite),
+]);
+```
+
+- `fetch` → `find_site(url)`（注册表中首个 PATTERN 命中且 `enabled()` 的站点，
+  返回 `&'static dyn Site`）→ `site.fetch_from_url(url).await`；
+- `cache_key` / `site_ids()` / `site_id_from_key()` / `apply_media_headers()` /
+  `validate_all()` 全部遍历 `SITES`；`validate_all` 返回失败列表，pixiv 的
+  `Site::validate` 失败时自行 `disable()`；
+- `match_site`/`SiteKind`（阶段 2 的静态分派）与中央 `fetch_error_is_retryable`
+  删除，重试判定走 `site.is_retryable`；
+- `main.rs` 的 pixiv 特判 → `site::validate_all()` + 通用失败通知；
+- 保留各站点的 `PATTERN`/`enabled()`/`fetch_from_url()` 顶层导出（兼容既有
+  测试），trait impl 只是薄壳。
+
+**async 形态**（实施结论）：**原生 AFIT 不可行**。
+
+- 实测（rustc 1.95.0，edition 2024；**1.97.1 复测一致**）：trait 里写
+  `async fn` 报 "method is `async`"（非 dyn 兼容）；写反糖
+  `-> impl Future<...> + Send + '_` 报 "references an `impl Trait` type in its
+  return type"（同样非 dyn 兼容）；纯 RPITIT（无 `+ Send`）也一样。即：
+  **RPITIT/AFIT 目前无法用于 `Vec<Box<dyn Site>>` 注册表**，与早期设计的
+  判断相反。
+- **为什么**：dyn 分派要求调用方在编译期知道返回值大小以分配空间，而
+  `async fn`/RPITIT 返回不透明的 Future——这是"非定长返回值走 dyn"的普遍问题，
+  与 async 无关。Rust 1.75 稳定的 AFIT 只覆盖**静态分派**，dyn 路径被排除；
+  原生 dyn 支持（AFIDT）是 2026-2027 的已接受项目目标，尚未进入 stable。
+  参见 <https://rust-lang.github.io/rust-project-goals/2026/afidt-box.html>。
+- **采用 (a) 手写 `Pin<Box<dyn Future + Send + '_>>`**（`SiteFuture` 别名）：
+  零新依赖、dyn 兼容、future 保证 Send。签名噪音靠别名缓解；生命周期坑因
+  站点是无状态单元结构体 + `'a` 同时约束 `&self` 与 `url` 而完全可控
+  （future 只借用调用域内的 url）。
+- **(b) `async-trait`** 仍是可行备选（语法更干净、同样 box），但新增依赖；
+  本仓库采用 (a) 后无需引入。
+- 若未来 Rust 稳定版落地 AFIDT（调用点 `dyn_box!`），可平滑迁移回原生
+  `async fn`，实现体几乎不动。
+
+**风险**：中。动中央分派，但每站点行为不变；注册表迭代 + `find_site` 补单测
+（`fetch`/`cache_key` 对既有 URL 集合的结果与阶段 2 完全一致）。
+**回滚**：revert。
+
+### 阶段 4：FetchError 泛化（已实施）
+
+**改动**：`FetchError` 新增 `Site { site: &'static str, error: Box<dyn std::error::Error + Send + Sync> }`
+变体（`Display`/`source()` 同步）。**`Pixiv(PixivError)` 变体保留**（未迁移）——
+它已有完整的 `Display`/`source()`/`is_retryable` 处理，替换纯属 churn。`Site`
+变体默认永久性（各站点 `is_retryable` 都不匹配它）；需要可重试站点错误的站点
+应自行转换为 `Http`/`Transient` 再返回。
+
+**风险**：低（纯增量变体）。测试：`site_error_variant_displays_and_sources`。
+
+### 阶段 5：收尾
+
+- 更新 `AGENTS.md` 的 "Site adapter convention" 段：写新约定（注册表 + `impl Site` +
+  每站点 `cache_key`/`is_retryable`/`media_headers`），删除 "no trait" 表述；
+- `examples/fetch.rs` 不变（走 `site::fetch`）；
+- 新增站点 checklist 见 §4。
+
+## 4. 重构后新增站点 checklist
+
+```
+1. crates/x-media/src/site/example/{mod,interface,model}.rs   // 新模块
+2. impl Site for ExampleSite 并注册进 SITES                   // 注册一行
+3. （可选）token 读取 + validate() 实现                         // 启动校验自动生效
+── bot crate 零改动 ──
+```
+
+对比现状的 8-9 处，bot crate 完全不碰：`SetFormat` 白名单、格式查找口径、
+缓存 key、启动校验全部自动跟随注册表。
+
+## 5. 权衡与明确不做的事
+
+- **不做**：Media 类型扩展（`media.rs` + `MediaItemPayload` + `CachedMediaKind` +
+  send.rs 约 10+ 处 match 的 blast radius）——这是"新增媒体类型"的摩擦，与"新增
+  站点"正交，优先级低，保持现状。
+- **不做**：DI/全局注入改造（`CHAT_STORE`/`TASK_QUEUE`/`CONFIG` 的 `LazyLock` 静态
+  模式是仓库惯例，与站点扩展无关）。
+- **不做**：schema 迁移——新站点只产生新的 cache key 前缀与 `message_format` JSON
+  key，`link_cache`/`chat_state` 表结构均无需变化。
+- **代价**：阶段 3 引入 `dyn Site` 与 boxed future 签名（`SiteFuture`，见 §3）；
+  `Send` 约束前移到 trait 边界，站点 impl 的 future 必须 Send（现仅在各
+  `tokio::spawn` 点检查，重构后在 impl 处即报错，提前暴露问题）。
+  若站点数量长期 ≤5 且无新增迹象，阶段 2 的折中方案已够用；本次已按完整方案
+  实施到阶段 4。
+
+## 6. 提交序列（已按此实施）
+
+| 阶段 | 提交 | hash |
+|---|---|---|
+| 1 | `refactor(site): carry site_id on Fetched; unify cache-key site lookup` | `7ca8fd1` |
+| 2 | `refactor(site): move cache_key/is_retryable/media_headers into site modules` | `5e23916` |
+| 3 | `refactor(site): introduce Site trait and SITES registry` | `bf4e615` |
+| 4 | `refactor(site): genericize FetchError::Site` | `5679a8c` |
+| 5 | `docs: update site adapter convention in AGENTS.md` | 本文档收尾提交 |
+
+每阶段独立合入、独立回滚；阶段 2 完成后"加站点"摩擦已收敛，3/4 为深化。