chore: bump version to 1.2.2

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.1, 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,11 +52,11 @@ 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`).

Cargo.lock

@@ -2925,7 +2925,7 @@ checksum = "1ffae5123b2d3fc086436f8834ae3ab053a283cfac8fe0a0b8eaae044768a4c4"
 
 [[package]]
 name = "x-media"
-version = "1.2.1"
+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.1"
+version = "1.2.2"
 dependencies = [
  "bytes",
  "dotenv",

crates/x-media/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "x-media"
-version = "1.2.1"
+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,39 +274,111 @@ 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 failures are retried: 3 total attempts with 1s then 2s delays.
-/// Retried classes: bare HTTP errors, [`FetchError::Transient`] (429/5xx
-/// from any site), pixiv network errors, and pixiv HTTP statuses that are
-/// actually transient (429 / 5xx). Permanent classes are returned
-/// immediately: Json, NotFound, Blocked, Sensitive, pixiv 4xx statuses
-/// (bad/expired token, forbidden, not found) and pixiv API/auth errors.
-/// Whether [`fetch`] should retry `err` (3 total attempts, 1s then 2s
-/// backoff). Permanent classes — 4xx statuses, invalid tokens, unparseable
-/// bodies, not-found/blocked/sensitive — are returned immediately; retrying
-/// them only wastes attempts against the source site.
-fn fetch_error_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,
-            // 4xx, invalid token, unparseable body: retrying cannot help.
-            PixivError::Status(_)
-            | PixivError::Api(_)
-            | PixivError::Json(_)
-            | PixivError::NoAuth => false,
-        },
-        _ => false,
-    }
-}
-
+/// 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)) => {
+        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",
@@ -338,9 +388,8 @@ pub async fn fetch(url: &str) -> Result<Option<Fetched>, FetchError> {
                 );
                 return Ok(Some(fetched));
             }
-            Ok(None) => return Ok(None),
             Err(err) => {
-                if fetch_error_is_retryable(&err) && attempt < 2 {
+                if site.is_retryable(&err) && attempt < 2 {
                     tokio::time::sleep(Duration::from_secs(1 << attempt)).await;
                 } else {
                     return Err(err);
@@ -351,33 +400,32 @@ pub async fn fetch(url: &str) -> Result<Option<Fetched>, FetchError> {
     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())
 }
 
@@ -386,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
     {
@@ -424,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
     {
@@ -477,48 +521,40 @@ mod tests {
     }
 
     #[test]
-    fn fetch_error_retryability_classification() {
-        // Transient: network errors, explicit transient, pixiv 429/5xx.
-        assert!(fetch_error_is_retryable(&FetchError::Transient(
-            "429".into()
-        )));
-        assert!(fetch_error_is_retryable(&FetchError::Pixiv(
-            PixivError::Status(429)
-        )));
-        assert!(fetch_error_is_retryable(&FetchError::Pixiv(
-            PixivError::Status(500)
-        )));
-        assert!(fetch_error_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!(!fetch_error_is_retryable(&FetchError::Pixiv(
-            PixivError::Status(400)
-        )));
-        assert!(!fetch_error_is_retryable(&FetchError::Pixiv(
-            PixivError::Status(401)
-        )));
-        assert!(!fetch_error_is_retryable(&FetchError::Pixiv(
-            PixivError::Status(403)
-        )));
-        assert!(!fetch_error_is_retryable(&FetchError::Pixiv(
-            PixivError::Status(404)
-        )));
-        assert!(!fetch_error_is_retryable(&FetchError::Pixiv(
-            PixivError::Api("invalid_grant".into())
-        )));
-        assert!(!fetch_error_is_retryable(&FetchError::Pixiv(
-            PixivError::NoAuth
-        )));
-        let json_err = serde_json::from_str::<serde_json::Value>("x").unwrap_err();
-        assert!(!fetch_error_is_retryable(&FetchError::Pixiv(
-            PixivError::Json(json_err)
-        )));
-        assert!(!fetch_error_is_retryable(&FetchError::NotFound));
-        assert!(!fetch_error_is_retryable(&FetchError::Blocked));
-        assert!(!fetch_error_is_retryable(&FetchError::Sensitive));
-        assert!(!fetch_error_is_retryable(&FetchError::TooLarge));
+    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]

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,53 +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::Status(code) => write!(f, "pixiv status {code}"),
-            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,

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()
 });
@@ -49,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 {
@@ -60,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/tombstoned
-/// tweets surface as `FetchError::NotFound`.
+/// Fetches a tweet from the syndication endpoint. Deleted/blocked tweets
+/// 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
@@ -85,9 +128,7 @@ pub async fn fetch(id: &str) -> Result<Tweet, FetchError> {
         };
     }
     let text = response.text().await?;
-    // Deleted/blocked tweets answer with an `errors` array or a
-    // TweetTombstone (HTTP 200, no `id_str`); NSFW withholding is an empty
-    // `{}`. Both classes are permanent — classify before parsing the tweet.
+    // Classify before parsing the tweet (see [`parse_syndication_body`]).
     parse_syndication_body(&text)?;
     Tweet::from_syndication_json(&text).map_err(FetchError::Json)
 }
@@ -95,17 +136,32 @@ pub async fn fetch(id: &str) -> Result<Tweet, FetchError> {
 /// 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`
-///   (deleted by the author / suspended — HTTP 200, no `errors`, no `id_str`).
-/// - `Sensitive`: an empty `{}` (NSFW / age-restricted withholding).
+///   **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.
-///
-/// The tombstone shape must NOT fall through to `Sensitive`: the bot would
-/// otherwise answer "No media found" for a deleted tweet instead of failing.
 fn parse_syndication_body(text: &str) -> Result<serde_json::Value, FetchError> {
     let body: serde_json::Value = serde_json::from_str(text)?;
-    let tombstoned = body.get("tombstone").is_some()
-        || body.get("__typename").and_then(|t| t.as_str()) == Some("TweetTombstone");
-    if body.get("errors").is_some() || tombstoned {
+    if body.get("errors").is_some() {
+        return Err(FetchError::NotFound);
+    }
+    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() {
@@ -301,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,
         }
@@ -352,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!([
@@ -590,9 +670,10 @@ mod tests {
 
     #[test]
     fn syndication_tombstone_maps_to_not_found() {
-        // Deleted tweets answer HTTP 200 with a TweetTombstone (no `errors`,
-        // no `id_str`); it must not fall through to Sensitive, which would
-        // make the bot reply "No media found" for a deleted tweet.
+        // 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": {
@@ -605,6 +686,34 @@ mod tests {
         ));
     }
 
+    #[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": [...]}.
@@ -667,4 +776,17 @@ mod tests {
             "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.1"
+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::{
@@ -79,12 +79,17 @@ pub async fn stop_url_workers() {
     }
 }
 
-pub static CHAT_STORE: LazyLock<ChatStore> =
-    LazyLock::new(|| ChatStore::open("data/task_queue.db").expect("failed to open chat store"));
+/// One shared SQLite pool for the three stores (chat state, task queue, link
+/// cache): a single pool bounds concurrent DB work on `data/task_queue.db`
+/// instead of three independent pools competing for the same file. The schema
+/// for all three tables is initialized once, here.
+static DB: LazyLock<Arc<db::DbPool>> =
+    LazyLock::new(|| db::open_store("data/task_queue.db").expect("failed to open database"));
+
+pub static CHAT_STORE: LazyLock<ChatStore> = LazyLock::new(|| ChatStore::new(Arc::clone(&DB)));
 pub static TASK_QUEUE: LazyLock<PersistentTaskQueue> =
-    LazyLock::new(|| PersistentTaskQueue::new("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)]
@@ -402,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(),
@@ -636,7 +641,9 @@ async fn url_media(bot: Bot, message: &Message, url: &str) {
     {
         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)

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;
             }
         }
     }

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()),
@@ -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

@@ -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,
             },
         }),
     }
@@ -958,24 +979,21 @@ pub async fn send_media_sequence(bot: &Bot, task: &Task) -> Result<Vec<i64>, Sen
                         .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) => {
@@ -1055,16 +1073,28 @@ pub async fn send_animation(bot: &Bot, task: &Task) -> Result<Vec<i64>, SendErro
                 "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
                     {
@@ -1076,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

@@ -1,8 +1,10 @@
 # 站点适配器重构方案：让新增站点变成"新模块 + 注册一行"
 
-> 状态：设计稿（未实施）。目标：把"加一个新站点"从改 8-9 处收敛到 3 处，
-> 并让站点身份、重试策略、下载 header 等站点能力归位到站点模块自身。
-> 本文只改文档，不动代码；每阶段均可独立合入、独立回滚。
+> 状态：**已实施**（阶段 1-5，提交 `7ca8fd1` / `5e23916` / `bf4e615` / `5679a8c` +
+> 本文档收尾）。目标：把"加一个新站点"从改 8-9 处收敛到 3 处，并让站点身份、
+> 重试策略、下载 header 等站点能力归位到站点模块自身。实施过程中的关键偏差
+> （async 形态）见 §3 的 "async 形态" 段——原生 AFIT 实测不可用于 dyn 分派，
+> 最终采用手写 `BoxFuture`（`SiteFuture` 别名）。
 
 ---
 
@@ -117,19 +119,23 @@ PATTERN（pixiv 的 PATTERN 只匹配 `pixiv.net/artworks/...`），所以 `medi
 
 **动机**：加站点时 bot crate 与中央分派零改动；站点列表成为唯一注册点。
 
-**新增**（`site/mod.rs`）：
+**新增**（`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;
-    fn cache_key(&self, url: &str) -> Option<String>;   // 默认: id + 捕获组1
-    fn fetch_from_url(&self, url: &str)
-        -> Pin<Box<dyn Future<Output = Result<Fetched, FetchError>> + Send>>;
-    fn is_retryable(&self, err: &FetchError) -> bool;   // 默认: Http|Transient
+    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) -> Option<BoxFuture<'static, Result<(), String>>>;      // 默认: None
+    fn validate(&self) -> SiteFuture<'static, (), String>;  // 默认: Ok(())
 }
 
 static SITES: LazyLock<Vec<Box<dyn Site>>> = LazyLock::new(|| vec![
@@ -137,38 +143,52 @@ static SITES: LazyLock<Vec<Box<dyn Site>>> = LazyLock::new(|| vec![
 ]);
 ```
 
-- `fetch_once` → `find_site(url)`（首个 PATTERN 命中且 `enabled()` 的站点）
-  → `site.fetch_from_url(url).await`；
-- `cache_key` / `site_ids()` / `media_headers` / `validate_all()` 全部遍历 `SITES`；
-- `fetch_error_is_retryable` 删除，重试判定走 `site.is_retryable`；
-- `main.rs:74-84` 的 pixiv 特判 → `site::validate_all()`（pixiv 的 `validate` 失败时
-  内部调用现有 `pixiv::disable()`，行为保持）；
-- 保留各站点的 `PATTERN`/`enabled()`/`fetch_from_url()` 顶层导出（兼容现有
-  `fetch_once` 及测试），trait 只是包一层薄壳。
+- `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 形态**：仓库没有 `async-trait` 依赖。两个选择：
-(a) 手写 `Pin<Box<dyn Future>>` 返回类型（零新依赖，契合仓库手写风格，签名略丑）；
-(b) 引入 `async-trait`（可读性好，新增一个依赖）。
-建议先 (a)，理由：仓库显式偏好手写错误/状态机，且 `BoxFuture` 已有先例
-（`queue.rs:38` 的 `BoxFuture`）。
+**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 泛化（可选，配合阶段 3）
+### 阶段 4：FetchError 泛化（已实施）
 
-**动机**：`FetchError::Pixiv(PixivError)`（`site/mod.rs:16,184,241-245`）是站点特有
-错误嵌进通用枚举；第 4 个站点要么再加变体，要么用泛化变体。
+**改动**：`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` 再返回。
 
-**改动**：`FetchError` 增加 `Site { site: &'static str, error: Box<dyn std::error::Error + Send + Sync> }`，
-`Pixiv(PixivError)` 变体保留但内部迁移到 `Site`（或直接替换并更新
-`is_retryable`/`Display`/`source()` 与测试）。重试判定在阶段 3 已归站点，
-中央枚举只剩通用类（Http/Json/NotFound/Blocked/Sensitive/TooLarge/Transient/Io）。
-
-**风险**：中。`Display`/`source()`/`From<PixivError>` 与 `fetch_error_is_retryable`
-测试（`site/mod.rs:480-522`）需同步。
-**回滚**：revert。
+**风险**：低（纯增量变体）。测试：`site_error_variant_displays_and_sources`。
 
 ### 阶段 5：收尾
 
@@ -198,18 +218,20 @@ static SITES: LazyLock<Vec<Box<dyn Site>>> = LazyLock::new(|| vec![
   模式是仓库惯例，与站点扩展无关）。
 - **不做**：schema 迁移——新站点只产生新的 cache key 前缀与 `message_format` JSON
   key，`link_cache`/`chat_state` 表结构均无需变化。
-- **代价**：阶段 3 引入 `dyn Site` 与（选择 (a) 时）手写 `BoxFuture` 签名；若站点
-  数量长期 ≤5 且无新增迹象，阶段 2 的折中方案已够用，阶段 3/4 可无限期推迟。
+- **代价**：阶段 3 引入 `dyn Site` 与 boxed future 签名（`SiteFuture`，见 §3）；
+  `Send` 约束前移到 trait 边界，站点 impl 的 future 必须 Send（现仅在各
+  `tokio::spawn` 点检查，重构后在 impl 处即报错，提前暴露问题）。
+  若站点数量长期 ≤5 且无新增迹象，阶段 2 的折中方案已够用；本次已按完整方案
+  实施到阶段 4。
 
-## 6. 建议的提交序列
+## 6. 提交序列（已按此实施）
 
-| 阶段 | 提交消息（建议） |
-|---|---|
-| 1 | `refactor(site): carry site_id on Fetched; unify cache-key site lookup` |
-| 2 | `refactor(site): move cache_key/is_retryable/media_headers into site modules` |
-| 3 | `refactor(site): introduce Site trait and SITES registry` |
-| 4 | `refactor(site): genericize FetchError::Site` |
-| 5 | `docs: update site adapter convention in AGENTS.md` |
+| 阶段 | 提交 | 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 为可选深化。
+每阶段独立合入、独立回滚；阶段 2 完成后"加站点"摩擦已收敛，3/4 为深化。