chore: bump version to 1.0.8

Apply rustfmt to the 11 files that had drifted (86 hunks): x-media
site modules (bsky/pixiv/twitter) and xmedia-bot (config/main/
photo/send). Formatting only - no semantic changes; full test suite
still green.

Cargo.lock

@@ -3351,7 +3351,7 @@ checksum = "1ffae5123b2d3fc086436f8834ae3ab053a283cfac8fe0a0b8eaae044768a4c4"
 
 [[package]]
 name = "x-media"
-version = "1.0.7"
+version = "1.0.8"
 dependencies = [
  "bytes",
  "dotenv",
@@ -3370,7 +3370,7 @@ dependencies = [
 
 [[package]]
 name = "xmedia-bot"
-version = "1.0.7"
+version = "1.0.8"
 dependencies = [
  "dotenv",
  "fast_image_resize",

README.md

@@ -101,12 +101,14 @@ Telegram 只接受 443/80/88/8443 端口。
 
 | 命令 | 说明 |
 |---|---|
-| `/set_forward_channel <频道>` | 设置转发频道 |
+| `/start` | 欢迎语 |
+| `/help` | 查看全部命令及用法（即本文档的命令表） |
+| `/set_forward_channel <频道>` | 设置转发频道，参数为 `@频道名` 或频道 ID；设置后发送的媒体消息会自动转发到该频道 |
 | `/remove_forward_channel` | 取消转发频道 |
-| `/edit_before_forward` | 开关转发前编辑 |
-| `/set_template <名称>` | 将回复的消息（含 `[]`）保存为模板 |
-| `/set_format <站点> <格式>` | 自定义 caption 格式（占位符 `{url}` `{title}` `{tags}` 等） |
-| `/bot_dict` | 查看聊天状态 |
+| `/edit_before_forward` | 开关「转发前编辑」：开启后，转发成功后 bot 会发一条提示消息，回复它可修改第一条转发消息的 caption（或点击模板按钮套用模板） |
+| `/set_template <名称>` | 回复一条含 `[]` 的消息，将其保存为命名模板；转发时 `[]` 会被替换为原帖链接（配合「转发前编辑」使用） |
+| `/set_format <站点> <格式>` | 自定义某站点的 caption 格式。站点：`twitter` / `bsky` / `pixiv`。占位符：`{url}` `{author}` `{author_url}` `{title}` `{tags}` |
+| `/bot_dict` | 查看当前聊天状态（调试用） |
 
 链接处理仅限私聊；命令在任意聊天可用。
 

crates/x-media/Cargo.toml

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

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

@@ -5,9 +5,8 @@ use html_escape::encode_text;
 use regex::Regex;
 use std::sync::LazyLock;
 
-pub static PATTERN: LazyLock<Regex> = LazyLock::new(|| {
-    Regex::new(r"bsky\.app/profile/([\w.\-:]+)/post/([\w.\-~]+)").unwrap()
-});
+pub static PATTERN: LazyLock<Regex> =
+    LazyLock::new(|| Regex::new(r"bsky\.app/profile/([\w.\-:]+)/post/([\w.\-~]+)").unwrap());
 
 pub fn enabled() -> bool {
     true
@@ -15,8 +14,14 @@ pub fn enabled() -> bool {
 
 pub async fn fetch_from_url(url: &str) -> Result<Fetched, FetchError> {
     let caps = PATTERN.captures(url).ok_or(FetchError::NotFound)?;
-    let handle = caps.get(1).map(|m| m.as_str()).ok_or(FetchError::NotFound)?;
-    let rkey = caps.get(2).map(|m| m.as_str()).ok_or(FetchError::NotFound)?;
+    let handle = caps
+        .get(1)
+        .map(|m| m.as_str())
+        .ok_or(FetchError::NotFound)?;
+    let rkey = caps
+        .get(2)
+        .map(|m| m.as_str())
+        .ok_or(FetchError::NotFound)?;
     Ok(fetch(handle, rkey).await?.into())
 }
 
@@ -197,7 +202,10 @@ mod tests {
         }));
         let post = Post::from_json(&raw.to_string(), "3xxxx".into()).unwrap();
         let fetched: Fetched = post.into();
-        assert_eq!(fetched.source_url, "https://bsky.app/profile/user.bsky.social/post/3xxxx");
+        assert_eq!(
+            fetched.source_url,
+            "https://bsky.app/profile/user.bsky.social/post/3xxxx"
+        );
         assert_eq!(fetched.title, "hello <world>");
         assert_eq!(fetched.media.len(), 1);
         assert!(!fetched.sensitive);
@@ -246,11 +254,10 @@ mod tests {
 
     #[tokio::test]
     async fn live_fetch_with_photos() {
-        let fetched = fetch_from_url(
-            "https://bsky.app/profile/asagi0398.bsky.social/post/3mqkhrq5w6k2m",
-        )
-        .await
-        .unwrap();
+        let fetched =
+            fetch_from_url("https://bsky.app/profile/asagi0398.bsky.social/post/3mqkhrq5w6k2m")
+                .await
+                .unwrap();
         assert_eq!(
             fetched.source_url,
             "https://bsky.app/profile/asagi0398.bsky.social/post/3mqkhrq5w6k2m"
@@ -260,9 +267,10 @@ mod tests {
 
     #[tokio::test]
     async fn live_fetch_smoke() {
-        let fetched = fetch_from_url("https://bsky.app/profile/fu-futa.bsky.social/post/3laoveufjv224")
-            .await
-            .unwrap();
+        let fetched =
+            fetch_from_url("https://bsky.app/profile/fu-futa.bsky.social/post/3laoveufjv224")
+                .await
+                .unwrap();
         assert_eq!(
             fetched.source_url,
             "https://bsky.app/profile/fu-futa.bsky.social/post/3laoveufjv224"

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

@@ -179,7 +179,6 @@ impl From<reqwest::Error> for FetchError {
     }
 }
 
-
 impl From<serde_json::Error> for FetchError {
     fn from(e: serde_json::Error) -> Self {
         FetchError::Json(e)

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

@@ -10,8 +10,8 @@ use crate::site::FetchError;
 use std::env;
 use std::fmt;
 use std::io::{Cursor, Read};
-use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::LazyLock;
+use std::sync::atomic::{AtomicBool, Ordering};
 use std::time::{Duration, SystemTime};
 
 const AUTH_TOKEN_URL: &str = "https://oauth.secure.pixiv.net/auth/token";
@@ -127,7 +127,9 @@ impl PixivAPI {
     pub async fn illust_detail(&self, illust_id: u64) -> Result<IllustrationModel, PixivError> {
         let access_token = self.get_access_token().await?;
         let response = crate::site::CLIENT
-            .get(format!("{APP_API_URL}/v1/illust/detail?illust_id={illust_id}"))
+            .get(format!(
+                "{APP_API_URL}/v1/illust/detail?illust_id={illust_id}"
+            ))
             .header("app-os", "ios")
             .header("app-os-version", "14.6")
             .header("User-Agent", APP_USER_AGENT)
@@ -175,7 +177,9 @@ impl PixivAPI {
     pub async fn ugoira_metadata(&self, illust_id: u64) -> Result<UgoiraMetadataModel, PixivError> {
         let access_token = self.get_access_token().await?;
         let response = crate::site::CLIENT
-            .get(format!("{APP_API_URL}/v1/ugoira/metadata?illust_id={illust_id}"))
+            .get(format!(
+                "{APP_API_URL}/v1/ugoira/metadata?illust_id={illust_id}"
+            ))
             .header("app-os", "ios")
             .header("app-os-version", "14.6")
             .header("User-Agent", APP_USER_AGENT)
@@ -218,87 +222,89 @@ impl PixivAPI {
         let Some(zip_url) = zip_url else {
             return Ok(None);
         };
-        let zip_bytes = crate::site::download_media(&zip_url).await.map_err(|e| match e {
-            FetchError::Http(e) => PixivError::Http(e),
-            other => PixivError::Api(format!("frame zip download failed: {other}")),
-        })?;
+        let zip_bytes = crate::site::download_media(&zip_url)
+            .await
+            .map_err(|e| match e {
+                FetchError::Http(e) => PixivError::Http(e),
+                other => PixivError::Api(format!("frame zip download failed: {other}")),
+            })?;
         let frame_delays = metadata.frames.iter().map(|f| f.delay).collect::<Vec<_>>();
-        let result = tokio::task::spawn_blocking(
-            move || -> Result<(String, tempfile::TempDir), String> {
-            let frames_dir = tempfile::tempdir().map_err(|e| e.to_string())?;
-            let out_dir = tempfile::tempdir().map_err(|e| e.to_string())?;
+        let result =
+            tokio::task::spawn_blocking(move || -> Result<(String, tempfile::TempDir), String> {
+                let frames_dir = tempfile::tempdir().map_err(|e| e.to_string())?;
+                let out_dir = tempfile::tempdir().map_err(|e| e.to_string())?;
 
-            // Extract frames to canonical zero-padded names; pixiv ugoira
-            // frames are uniformly jpg or png per artwork.
-            let mut archive = zip::ZipArchive::new(Cursor::new(zip_bytes))
-                .map_err(|e| format!("unzip: {e}"))?;
-            // pixiv ugoira frames are uniformly jpg or png per artwork; take
-            // the extension from the first entry.
-            let extension = if archive.len() > 0 {
-                let first_name = archive
-                    .by_index(0)
-                    .map_err(|e| e.to_string())?
-                    .name()
-                    .to_string();
-                first_name
-                    .rsplit('.')
-                    .next()
-                    .unwrap_or("jpg")
-                    .to_string()
-            } else {
-                "jpg".to_string()
-            };
-            let mut count = 0usize;
-            for i in 0..archive.len() {
-                let mut entry = archive.by_index(i).map_err(|e| e.to_string())?;
-                let mut bytes = Vec::new();
-                entry.read_to_end(&mut bytes).map_err(|e| e.to_string())?;
-                let path = frames_dir.path().join(format!("img_{count:05}.{extension}"));
-                std::fs::write(&path, bytes).map_err(|e| e.to_string())?;
-                count += 1;
-            }
-            if count == 0 {
-                return Err("empty frame zip".to_string());
-            }
+                // Extract frames to canonical zero-padded names; pixiv ugoira
+                // frames are uniformly jpg or png per artwork.
+                let mut archive = zip::ZipArchive::new(Cursor::new(zip_bytes))
+                    .map_err(|e| format!("unzip: {e}"))?;
+                // pixiv ugoira frames are uniformly jpg or png per artwork; take
+                // the extension from the first entry.
+                let extension = if archive.len() > 0 {
+                    let first_name = archive
+                        .by_index(0)
+                        .map_err(|e| e.to_string())?
+                        .name()
+                        .to_string();
+                    first_name.rsplit('.').next().unwrap_or("jpg").to_string()
+                } else {
+                    "jpg".to_string()
+                };
+                let mut count = 0usize;
+                for i in 0..archive.len() {
+                    let mut entry = archive.by_index(i).map_err(|e| e.to_string())?;
+                    let mut bytes = Vec::new();
+                    entry.read_to_end(&mut bytes).map_err(|e| e.to_string())?;
+                    let path = frames_dir
+                        .path()
+                        .join(format!("img_{count:05}.{extension}"));
+                    std::fs::write(&path, bytes).map_err(|e| e.to_string())?;
+                    count += 1;
+                }
+                if count == 0 {
+                    return Err("empty frame zip".to_string());
+                }
 
-            // Constant rate from the median frame delay (ms).
-            let mut delays = frame_delays;
-            delays.sort_unstable();
-            let median = delays[delays.len() / 2].max(1);
-            let framerate = 1000.0 / median as f64;
+                // Constant rate from the median frame delay (ms).
+                let mut delays = frame_delays;
+                delays.sort_unstable();
+                let median = delays[delays.len() / 2].max(1);
+                let framerate = 1000.0 / median as f64;
 
-            let output = out_dir.path().join("ugoira.mp4");
-            let status = std::process::Command::new("ffmpeg")
-                .args([
-                    "-y",
-                    "-framerate",
-                    &framerate.to_string(),
-                    "-i",
-                    &frames_dir.path().join(format!("img_%05d.{extension}")).to_string_lossy(),
-                    // libx264 needs even dimensions; pixiv ugoira frames can
-                    // be odd-sized (e.g. 277x405).
-                    "-vf",
-                    "scale=trunc(iw/2)*2:trunc(ih/2)*2",
-                    "-c:v",
-                    "libx264",
-                    "-pix_fmt",
-                    "yuv420p",
-                    "-movflags",
-                    "+faststart",
-                    &output.to_string_lossy(),
-                ])
-                .stdout(std::process::Stdio::null())
-                .stderr(std::process::Stdio::null())
-                .status()
-                .map_err(|e| format!("ffmpeg spawn failed: {e}"))?;
-            if !status.success() {
-                return Err(format!("ffmpeg exited with {status}"));
-            }
-            Ok((output.to_string_lossy().into_owned(), out_dir))
-        },
-        )
-        .await
-        .expect("ugoira encode worker panicked");
+                let output = out_dir.path().join("ugoira.mp4");
+                let status = std::process::Command::new("ffmpeg")
+                    .args([
+                        "-y",
+                        "-framerate",
+                        &framerate.to_string(),
+                        "-i",
+                        &frames_dir
+                            .path()
+                            .join(format!("img_%05d.{extension}"))
+                            .to_string_lossy(),
+                        // libx264 needs even dimensions; pixiv ugoira frames can
+                        // be odd-sized (e.g. 277x405).
+                        "-vf",
+                        "scale=trunc(iw/2)*2:trunc(ih/2)*2",
+                        "-c:v",
+                        "libx264",
+                        "-pix_fmt",
+                        "yuv420p",
+                        "-movflags",
+                        "+faststart",
+                        &output.to_string_lossy(),
+                    ])
+                    .stdout(std::process::Stdio::null())
+                    .stderr(std::process::Stdio::null())
+                    .status()
+                    .map_err(|e| format!("ffmpeg spawn failed: {e}"))?;
+                if !status.success() {
+                    return Err(format!("ffmpeg exited with {status}"));
+                }
+                Ok((output.to_string_lossy().into_owned(), out_dir))
+            })
+            .await
+            .expect("ugoira encode worker panicked");
         match result {
             Ok(pair) => Ok(Some(pair)),
             Err(message) => {
@@ -332,9 +338,8 @@ fn log_once_ffmpeg_missing() {
 }
 
 /// pixiv3-rs replacement: `None` when `PIXIV_REFRESH_TOKEN` is unset.
-static PIXIV_CLIENT: LazyLock<Option<PixivAPI>> = LazyLock::new(|| {
-    env::var("PIXIV_REFRESH_TOKEN").ok().map(PixivAPI::new)
-});
+static PIXIV_CLIENT: LazyLock<Option<PixivAPI>> =
+    LazyLock::new(|| env::var("PIXIV_REFRESH_TOKEN").ok().map(PixivAPI::new));
 
 /// Set at startup when the login validation fails; pixiv stays disabled until
 /// the next process start.

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

@@ -82,12 +82,15 @@ impl Illustration {
             // keeps media empty when encoding fails or ffmpeg is missing.
         } else if model.page_count > 1 {
             media.extend(model.meta_pages.iter().filter_map(|page| {
-                page.image_urls.original.clone().map(|original| Media::Illustration {
-                    title: None,
-                    url: original,
-                    thumbnail_url: Some(page.image_urls.medium.clone()),
-                    fallback_url: Some(page.image_urls.large.clone()),
-                })
+                page.image_urls
+                    .original
+                    .clone()
+                    .map(|original| Media::Illustration {
+                        title: None,
+                        url: original,
+                        thumbnail_url: Some(page.image_urls.medium.clone()),
+                        fallback_url: Some(page.image_urls.large.clone()),
+                    })
             }));
         } else if let Some(original) = model
             .meta_single_page
@@ -147,8 +150,8 @@ impl From<Illustration> for Fetched {
 
 #[cfg(test)]
 mod tests {
-    use super::*;
     use super::super::model::IllustrationModel;
+    use super::*;
 
     fn illust_json(
         type_: &str,
@@ -203,8 +206,14 @@ mod tests {
             ("https://pixiv.net/artworks/123456", "123456"),
             ("https://www.pixiv.net/en/artworks/123456", "123456"),
             ("https://www.pixiv.net/i/123456", "123456"),
-            ("https://www.pixiv.net/member_illust.php?mode=medium&illust_id=123456", "123456"),
-            ("https://www.pixiv.net/en/member_illust.php?illust_id=123456", "123456"),
+            (
+                "https://www.pixiv.net/member_illust.php?mode=medium&illust_id=123456",
+                "123456",
+            ),
+            (
+                "https://www.pixiv.net/en/member_illust.php?illust_id=123456",
+                "123456",
+            ),
         ];
         for (url, id) in cases {
             let caps = PATTERN.captures(url).unwrap_or_else(|| panic!("{url}"));
@@ -225,7 +234,14 @@ mod tests {
 
     #[test]
     fn ugoira_yields_empty_media() {
-        let v = illust_json("ugoira", 1, Some("https://i.pximg.net/orig.jpg"), None, vec![], 0);
+        let v = illust_json(
+            "ugoira",
+            1,
+            Some("https://i.pximg.net/orig.jpg"),
+            None,
+            vec![],
+            0,
+        );
         let illustration = parse(v);
         let fetched: Fetched = illustration.into();
         assert!(fetched.media.is_empty());
@@ -296,7 +312,12 @@ mod tests {
         let fetched: Fetched = parse(v).into();
         assert_eq!(fetched.media.len(), 1);
         match &fetched.media[0] {
-            Media::Illustration { url, thumbnail_url, fallback_url, .. } => {
+            Media::Illustration {
+                url,
+                thumbnail_url,
+                fallback_url,
+                ..
+            } => {
                 assert_eq!(url, "https://i.pximg.net/p2.jpg");
                 assert_eq!(thumbnail_url.as_deref(), Some("m2.jpg"));
                 assert_eq!(fallback_url.as_deref(), Some("l2.jpg"));
@@ -307,7 +328,14 @@ mod tests {
 
     #[test]
     fn caption_with_escapes_format_and_substitutes() {
-        let v = illust_json("illust", 1, Some("https://i.pximg.net/o.jpg"), None, vec![], 0);
+        let v = illust_json(
+            "illust",
+            1,
+            Some("https://i.pximg.net/o.jpg"),
+            None,
+            vec![],
+            0,
+        );
         let fetched: Fetched = parse(v).into();
         // Format string is escaped in full, then placeholders substituted.
         let out = fetched.caption_with("{title} by {author} <script> {tags}");
@@ -330,7 +358,14 @@ mod tests {
     #[test]
     fn ai_work_gets_leading_ai_tag() {
         // illust_ai_type == 2 is the only AI marker.
-        let v = illust_json("illust", 1, Some("https://i.pximg.net/o.jpg"), None, vec![], 2);
+        let v = illust_json(
+            "illust",
+            1,
+            Some("https://i.pximg.net/o.jpg"),
+            None,
+            vec![],
+            2,
+        );
         let fetched: Fetched = parse(v).into();
         assert!(
             fetched.caption.contains("#AI #tag1 #tag2"),
@@ -338,14 +373,25 @@ mod tests {
             fetched.caption
         );
         // The {tags} placeholder reflects the tag array too.
-        assert!(fetched.caption_with("{tags}").starts_with("#AI "), "got: {}", fetched.caption_with("{tags}"));
+        assert!(
+            fetched.caption_with("{tags}").starts_with("#AI "),
+            "got: {}",
+            fetched.caption_with("{tags}")
+        );
     }
 
     #[test]
     fn non_ai_work_has_no_ai_tag() {
         // 1 = explicitly not AI, 0 = undefined: neither gets the #AI tag.
         for ai_type in [0, 1] {
-            let v = illust_json("illust", 1, Some("https://i.pximg.net/o.jpg"), None, vec![], ai_type);
+            let v = illust_json(
+                "illust",
+                1,
+                Some("https://i.pximg.net/o.jpg"),
+                None,
+                vec![],
+                ai_type,
+            );
             let fetched: Fetched = parse(v).into();
             assert!(
                 !fetched.caption.contains("#AI"),
@@ -357,7 +403,14 @@ mod tests {
 
     #[test]
     fn caption_escapes_and_links() {
-        let v = illust_json("illust", 1, Some("https://i.pximg.net/o.jpg"), None, vec![], 0);
+        let v = illust_json(
+            "illust",
+            1,
+            Some("https://i.pximg.net/o.jpg"),
+            None,
+            vec![],
+            0,
+        );
         let fetched: Fetched = parse(v).into();
         assert!(
             fetched
@@ -367,9 +420,6 @@ mod tests {
             fetched.caption
         );
         assert!(fetched.caption.contains("#tag1 #tag2"));
-        assert_eq!(
-            fetched.source_url,
-            "https://www.pixiv.net/artworks/123"
-        );
+        assert_eq!(fetched.source_url, "https://www.pixiv.net/artworks/123");
     }
 }

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

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

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

@@ -23,7 +23,7 @@
 
 use std::sync::LazyLock;
 
-use serde_json::{json, Value};
+use serde_json::{Value, json};
 
 use crate::site::FetchError;
 
@@ -40,8 +40,7 @@ static AUTH_TOKEN: LazyLock<Option<String>> = LazyLock::new(|| {
 });
 
 /// Public "logged in" client token used by the x.com web app.
-const LOGGED_IN_BEARER: &str =
-    "Bearer AAAAAAAAAAAAAAAAAAAAANRILgAAAAAAnNwIzUejRCOuH5E6I8xnZz4puTs%3D1Zv7ttfk8LF81IUq16cHjhLTvJu4FA33AGWWjCpTnA";
+const LOGGED_IN_BEARER: &str = "Bearer AAAAAAAAAAAAAAAAAAAAANRILgAAAAAAnNwIzUejRCOuH5E6I8xnZz4puTs%3D1Zv7ttfk8LF81IUq16cHjhLTvJu4FA33AGWWjCpTnA";
 
 /// `TweetDetail` query id (from nazurin; still valid as of 2026-08,
 /// corroborated by the current FxEmbed build — see module caveats).
@@ -103,9 +102,7 @@ pub fn enabled() -> bool {
 /// Fetches a tweet as the logged-in user via the private GraphQL API.
 /// Returns the syndication-shaped [`Tweet`] (media included for NSFW posts).
 pub async fn fetch(id: &str) -> Result<Tweet, FetchError> {
-    let token = AUTH_TOKEN
-        .as_deref()
-        .ok_or(FetchError::Sensitive)?;
+    let token = AUTH_TOKEN.as_deref().ok_or(FetchError::Sensitive)?;
     // 16 random bytes as 32 hex chars: X rejects ct0 values of any other
     // length with 403 code 353 ("matching csrf cookie and header").
     let ct0: String = (0..16)
@@ -136,11 +133,12 @@ pub async fn fetch(id: &str) -> Result<Tweet, FetchError> {
     let text = response.text().await?;
     let json: Value = serde_json::from_str(&text)?;
     let result = parse_tweet_result(&json, id)?;
-    let syndication_shape = to_syndication_shape(&result)
-        .ok_or_else(|| FetchError::Json(serde_json::Error::io(std::io::Error::new(
+    let syndication_shape = to_syndication_shape(&result).ok_or_else(|| {
+        FetchError::Json(serde_json::Error::io(std::io::Error::new(
             std::io::ErrorKind::InvalidData,
             "missing tweet fields in GraphQL response",
-        ))))?;
+        )))
+    })?;
     Tweet::from_syndication_json(&syndication_shape.to_string()).map_err(FetchError::Json)
 }
 
@@ -228,7 +226,6 @@ fn to_syndication_shape(tweet: &Value) -> Option<Value> {
             "screen_name": user.get("screen_name"),
         },
         "possibly_sensitive": legacy.get("possibly_sensitive"),
-        "display_text_range": legacy.get("display_text_range"),
         "entities": legacy.get("entities"),
         "mediaDetails": legacy.pointer("/extended_entities/media"),
     }))
@@ -251,12 +248,12 @@ mod tests {
             "legacy": {
                 "id_str": "2083868672721039569",
                 "full_text": "nsfw content https://t.co/abc123",
-                "display_text_range": [0, 12],
                 "possibly_sensitive": true,
                 "entities": {
-                    "urls": [
-                        { "url": "https://t.co/abc123", "expanded_url": "https://example.com/x" }
-                    ]
+                    // The appended media link lives in extended_entities.media,
+                    // not entities.urls, so it has no expansion mapping and the
+                    // content-based strip removes it.
+                    "urls": []
                 },
                 "extended_entities": {
                     "media": [
@@ -318,8 +315,11 @@ mod tests {
             }
             other => panic!("expected video, got {other:?}"),
         }
-        assert_eq!(fetched.source_url, "https://x.com/nsfw_author/status/2083868672721039569");
-        // display_text_range cuts the trailing t.co link.
+        assert_eq!(
+            fetched.source_url,
+            "https://x.com/nsfw_author/status/2083868672721039569"
+        );
+        // The appended media short link (no URL-entity mapping) is stripped.
         assert_eq!(fetched.title, "nsfw content");
     }
 
@@ -331,7 +331,10 @@ mod tests {
         let json = conversation(rt);
         let result = parse_tweet_result(&json, "2083868672721039569").unwrap();
         assert!(result.pointer("/legacy/retweeted_status_result").is_none());
-        assert_eq!(result.pointer("/legacy/id_str").unwrap(), "2083868672721039569");
+        assert_eq!(
+            result.pointer("/legacy/id_str").unwrap(),
+            "2083868672721039569"
+        );
     }
 
     #[test]

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

@@ -35,9 +35,7 @@ pub async fn fetch_from_url(url: &str) -> Result<Fetched, FetchError> {
                     }
                 }
             } else {
-                log::info!(
-                    "tweet {id} is sensitive; set TWITTER_AUTH_TOKEN to fetch NSFW media"
-                );
+                log::info!("tweet {id} is sensitive; set TWITTER_AUTH_TOKEN to fetch NSFW media");
                 Ok(empty_fetched(url))
             }
         }
@@ -158,12 +156,10 @@ impl Tweet {
     pub fn from_syndication_json(raw_json: &str) -> Result<Self, serde_json::Error> {
         let json: model::SyndicationTweet = serde_json::from_str(raw_json)?;
         let id = json.id_str;
-        // Strip the appended media short link first, then expand the remaining
-        // t.co short links (the user's own URLs) to their real destinations.
-        let text = expand_links(
-            &strip_trailing_short_links(&json.text, json.display_text_range),
-            &json.entities.urls,
-        );
+        // Expand the user's t.co short links to their real destinations and
+        // strip the appended media short link, mirroring FxEmbed's linkFixer
+        // (no display_text_range arithmetic — see expand_links).
+        let text = expand_links(&json.text, &json.entities.urls);
         // `name` is the display name, `screen_name` the handle (Python's
         // vxtwitter mapping: author = display name, author_id = handle).
         let author = json.user.name;
@@ -204,51 +200,48 @@ impl Tweet {
     }
 }
 
-/// The raw syndication `text` ends with the appended media short link
-/// (" https://t.co/wmI8McgXul"). `display_text_range` marks the visible text;
-/// a regex strips any remaining trailing t.co link when the range is absent
-/// or a tweet ends in a URL short link.
-///
-/// X reports these indices in Unicode **code points**, not UTF-16 units
-/// (verified against GraphQL responses containing emoji: cutting an emoji
-/// tweet by UTF-16 units silently drops the character after the emoji).
-fn strip_trailing_short_links(text: &str, display_text_range: Option<[usize; 2]>) -> String {
-    let mut out = match display_text_range {
-        Some([start, end]) if start < end => {
-            text.chars().skip(start).take(end - start).collect()
-        }
-        _ => text.to_string(),
-    };
-    while TRAILING_TCO.is_match(&out) {
-        out = TRAILING_TCO.replace(&out, "").into_owned();
-    }
-    out
-}
-
-/// Trailing Twitter short link, optionally preceded by whitespace.
-static TRAILING_TCO: LazyLock<Regex> = LazyLock::new(|| {
-    Regex::new(r"\s*https?://t\.co/[A-Za-z0-9]+$").unwrap()
-});
-
-/// Replaces every t.co short link that has an entity mapping with its
-/// expanded URL. Short links without a mapping stay untouched.
+/// Mirrors FxEmbed's `linkFixer` (link-fixer.ts): expand every t.co short
+/// link that has an entity mapping to its real destination, drop internal
+/// `x.com/i/web/status/…` plumbing links, then strip any remaining t.co
+/// short link (the appended media link and other unmapped short links).
+/// Pure content matching — no `display_text_range` arithmetic, so the
+/// endpoint's inconsistent index units (UTF-16 vs code points, see the
+/// deleted `strip_trailing_short_links`) never matter.
 fn expand_links(text: &str, urls: &[model::SyndicationEntityUrl]) -> String {
     let mut out = text.to_string();
     for entity in urls {
-        if let Some(expanded) = &entity.expanded_url {
-            out = out.replace(&entity.url, expanded);
-        }
+        let Some(expanded) = &entity.expanded_url else {
+            continue;
+        };
+        let replacement = if WEB_STATUS_URL.is_match(expanded) {
+            ""
+        } else {
+            expanded
+        };
+        out = out.replace(&entity.url, replacement);
     }
-    out
+    TCO_LINK.replace_all(&out, "").into_owned()
 }
 
+/// Internal x.com page links (reply / quote plumbing) expand to
+/// `x.com/i/web/status/<id>`; FxEmbed drops them — the tweet's own content
+/// already carries the information.
+static WEB_STATUS_URL: LazyLock<Regex> =
+    LazyLock::new(|| Regex::new(r"^https://(?:x\.com|twitter\.com)/i/web/status/\w+").unwrap());
+
+/// A t.co short link, optionally preceded by a space. Any leftover
+/// occurrence (unmapped — e.g. the appended media link) is removed,
+/// mirroring FxEmbed. Real short-link codes are 10 alphanumerics; the
+/// length-agnostic class keeps fixtures and hypothetical odd lengths safe.
+static TCO_LINK: LazyLock<Regex> =
+    LazyLock::new(|| Regex::new(r" ?https?://t\.co/[A-Za-z0-9]+").unwrap());
+
 /// pbs.twimg.com serves a reduced default size without size params; `name=orig`
 /// returns the original file (fxtwitter used to hand out the original
 /// directly, the syndication API does not). Non-twimg URLs pass through
 /// unchanged.
 fn original_twimg_url(url: &str) -> String {
-    if url.starts_with("https://pbs.twimg.com/")
-        && (url.ends_with(".jpg") || url.ends_with(".png"))
+    if url.starts_with("https://pbs.twimg.com/") && (url.ends_with(".jpg") || url.ends_with(".png"))
     {
         format!("{url}?name=orig")
     } else {
@@ -363,24 +356,23 @@ mod tests {
         match &fetched.media[0] {
             Media::Illustration { url, .. } => {
                 // Photo URL is rewritten to request the original file.
-                assert_eq!(
-                    url,
-                    "https://pbs.twimg.com/media/photo.jpg?name=orig"
-                );
+                assert_eq!(url, "https://pbs.twimg.com/media/photo.jpg?name=orig");
             }
             other => panic!("expected illustration, got {other:?}"),
         }
         match &fetched.media[1] {
-            Media::Video { url, thumbnail_url, .. } => {
+            Media::Video {
+                url, thumbnail_url, ..
+            } => {
                 assert_eq!(url, "https://video.twimg.com/v.mp4");
                 assert_eq!(thumbnail_url, "https://pbs.twimg.com/thumb.jpg");
             }
             other => panic!("expected video, got {other:?}"),
         }
         assert!(
-            fetched
-                .caption
-                .contains("<a href=\"https://x.com/author_handle\">Display Name</a>: a &amp; b &lt;c&gt;"),
+            fetched.caption.contains(
+                "<a href=\"https://x.com/author_handle\">Display Name</a>: a &amp; b &lt;c&gt;"
+            ),
             "caption: {}",
             fetched.caption
         );
@@ -411,13 +403,12 @@ mod tests {
 
     #[test]
     fn syndication_text_strips_trailing_media_short_link() {
-        // Real syndication shape: the media short link sits after the visible
-        // text, and display_text_range marks where it begins.
+        // Real syndication shape: the appended media short link sits after the
+        // visible text; the unmapped t.co link is stripped by content.
         let raw = serde_json::json!({
             "__typename": "Tweet",
             "id_str": "1",
             "text": "hello world https://t.co/abc123",
-            "display_text_range": [0, 11],
             "user": { "name": "N", "screen_name": "h" },
             "mediaDetails": []
         });
@@ -427,8 +418,31 @@ mod tests {
     }
 
     #[test]
-    fn syndication_text_strips_trailing_short_link_without_range() {
-        // No display_text_range: the regex fallback removes the trailing link.
+    fn syndication_text_strips_trailing_link_regardless_of_index_units() {
+        // Real tweet 2084567054481571919: the visible text is 30 code points
+        // but 41 UTF-16 units, and the two endpoints historically reported
+        // display_text_range in different units (UTF-16 on syndication, code
+        // points on GraphQL). The FxEmbed-style content-based strip ignores
+        // the range entirely, so the appended media link is removed for any
+        // response shape.
+        let text = "妄想𝑨𝒅𝒅𝒊𝒄𝒕𝒊𝒐𝒏…🩷💚❤️\n#ゼンゼロ　#zzzero https://t.co/XnIi83EkEB";
+        let visible = "妄想𝑨𝒅𝒅𝒊𝒄𝒕𝒊𝒐𝒏…🩷💚❤️\n#ゼンゼロ　#zzzero";
+        let raw = serde_json::json!({
+            "__typename": "Tweet",
+            "id_str": "2084567054481571919",
+            "text": text,
+            "user": { "name": "N", "screen_name": "h" },
+            "mediaDetails": []
+        });
+        let tweet = Tweet::from_syndication_json(&raw.to_string()).unwrap();
+        assert_eq!(tweet.text, visible, "left a partial link");
+        assert!(!tweet.caption().contains("t.co"));
+    }
+
+    #[test]
+    fn syndication_text_strips_trailing_short_link_without_entities() {
+        // No URL entities at all: the leftover t.co link is stripped by the
+        // content regex.
         let raw = serde_json::json!({
             "__typename": "Tweet",
             "id_str": "1",
@@ -449,7 +463,6 @@ mod tests {
             "__typename": "Tweet",
             "id_str": "1",
             "text": "Test Tweet with @mentionThis $twtr https://t.co/RzmrQ6wAzD #hashtag https://t.co/9r69akA484",
-            "display_text_range": [0, 67],
             "user": { "name": "N", "screen_name": "h" },
             "entities": {
                 "urls": [{
@@ -469,25 +482,47 @@ mod tests {
     }
 
     #[test]
-    fn syndication_text_keeps_unmapped_short_links() {
-        // No entity mapping for the embedded link: it stays as-is. Only the
-        // trailing media link is stripped.
+    fn syndication_text_strips_unmapped_short_links() {
+        // FxEmbed parity: short links without an entity mapping (appended
+        // media link, embedded unmapped links) are stripped, not kept.
         let raw = serde_json::json!({
             "__typename": "Tweet",
             "id_str": "1",
             "text": "check https://t.co/abc123 #tag https://t.co/def456",
-            "display_text_range": [0, 30],
             "user": { "name": "N", "screen_name": "h" },
             "mediaDetails": []
         });
         let tweet = Tweet::from_syndication_json(&raw.to_string()).unwrap();
-        assert_eq!(tweet.text, "check https://t.co/abc123 #tag");
+        assert_eq!(tweet.text, "check #tag");
     }
 
     #[test]
-    fn syndication_text_utf16_display_range_keeps_multibyte() {
-        // display_text_range is in UTF-16 units; a Japanese text must not be
-        // sliced by UTF-8 bytes.
+    fn syndication_text_drops_internal_web_status_links() {
+        // FxEmbed parity: a mapped link expanding to an internal
+        // x.com/i/web/status/... page (reply/quote plumbing) is removed
+        // instead of being shown.
+        let raw = serde_json::json!({
+            "__typename": "Tweet",
+            "id_str": "1",
+            "text": "see https://t.co/xyz1234567 for context",
+            "user": { "name": "N", "screen_name": "h" },
+            "entities": {
+                "urls": [{
+                    "url": "https://t.co/xyz1234567",
+                    "expanded_url": "https://x.com/i/web/status/9876543210",
+                    "display_url": "x.com/i/web/status/9876543210"
+                }]
+            },
+            "mediaDetails": []
+        });
+        let tweet = Tweet::from_syndication_json(&raw.to_string()).unwrap();
+        assert_eq!(tweet.text, "see  for context");
+        assert!(!tweet.caption().contains("t.co"));
+    }
+
+    #[test]
+    fn syndication_text_keeps_multibyte_text() {
+        // Text-only tweet: no short links, the multibyte text is untouched.
         let text = "コミティア落ちたので、明日は行きません。🙏ごめんなさい";
         let units: Vec<u16> = text.encode_utf16().collect();
         assert_eq!(units.len(), 28);
@@ -495,7 +530,6 @@ mod tests {
             "__typename": "Tweet",
             "id_str": "1",
             "text": text,
-            "display_text_range": [0, 28],
             "user": { "name": "N", "screen_name": "h" },
             "mediaDetails": []
         });
@@ -547,6 +581,9 @@ mod tests {
     async fn live_fetch_deleted_tweet_is_not_found() {
         // Deleted tweet: the syndication endpoint answers with errors.
         let result = fetch("0").await;
-        assert!(matches!(result, Err(FetchError::NotFound)), "got {result:?}");
+        assert!(
+            matches!(result, Err(FetchError::NotFound)),
+            "got {result:?}"
+        );
     }
 }

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

@@ -9,10 +9,6 @@ pub struct SyndicationTweet {
     pub user: SyndicationUser,
     #[serde(default)]
     pub possibly_sensitive: Option<bool>,
-    /// Visible-text span; the raw `text` field has the appended media short
-    /// link after it. Indices are Unicode code points (not UTF-16 units).
-    #[serde(default, rename = "display_text_range")]
-    pub display_text_range: Option<[usize; 2]>,
     #[serde(default)]
     pub entities: SyndicationEntities,
     #[serde(default, rename = "mediaDetails")]

crates/xmedia-bot/Cargo.toml

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

crates/xmedia-bot/src/config.rs

@@ -24,39 +24,37 @@ pub struct Config {
 impl Config {
     pub fn load() -> Config {
         let admin_ids = env::var("BOT_ADMIN")
-        .ok()
-        .map(|s| {
-            s.split(',')
-                .filter_map(|part| part.trim().parse::<i64>().ok())
-                .collect()
-        })
-        .unwrap_or_default();
+            .ok()
+            .map(|s| {
+                s.split(',')
+                    .filter_map(|part| part.trim().parse::<i64>().ok())
+                    .collect()
+            })
+            .unwrap_or_default();
 
-    let edit_message_ttl = env::var("EDIT_MESSAGE_TTL_SECONDS")
-        .ok()
-        .and_then(|s| s.parse::<u64>().ok())
-        .map(Duration::from_secs)
-        .unwrap_or(Duration::from_secs(86400));
+        let edit_message_ttl = env::var("EDIT_MESSAGE_TTL_SECONDS")
+            .ok()
+            .and_then(|s| s.parse::<u64>().ok())
+            .map(Duration::from_secs)
+            .unwrap_or(Duration::from_secs(86400));
 
-    let link_cache_ttl = env::var("LINK_CACHE_TTL_SECONDS")
-        .ok()
-        .and_then(|s| s.parse::<u64>().ok())
-        .map(Duration::from_secs)
-        .unwrap_or(Duration::from_secs(7 * 24 * 3600));
+        let link_cache_ttl = env::var("LINK_CACHE_TTL_SECONDS")
+            .ok()
+            .and_then(|s| s.parse::<u64>().ok())
+            .map(Duration::from_secs)
+            .unwrap_or(Duration::from_secs(7 * 24 * 3600));
 
-    let webhook_enabled = env::var("WEBHOOK")
-        .is_ok_and(|v| matches!(v.to_lowercase().as_str(), "true" | "yes" | "1"));
-    let webhook_url = env::var("WEBHOOK_URL").ok().and_then(|s| s.parse().ok());
-    let webhook_listen = env::var("WEBHOOK_LISTEN").ok().and_then(|s| s.parse().ok());
-    let webhook_port = env::var("WEBHOOK_PORT").ok().and_then(|s| s.parse().ok());
-    // Empty strings count as unset (e.g. `-e WEBHOOK_CERT=` to disable a
-    // value that would otherwise come from `.env`).
-    let webhook_cert = env::var("WEBHOOK_CERT")
-        .ok()
-        .filter(|s| !s.is_empty());
-    let webhook_secret_token = env::var("WEBHOOK_SECRET_TOKEN")
-        .ok()
-        .filter(|s| !s.is_empty());
+        let webhook_enabled = env::var("WEBHOOK")
+            .is_ok_and(|v| matches!(v.to_lowercase().as_str(), "true" | "yes" | "1"));
+        let webhook_url = env::var("WEBHOOK_URL").ok().and_then(|s| s.parse().ok());
+        let webhook_listen = env::var("WEBHOOK_LISTEN").ok().and_then(|s| s.parse().ok());
+        let webhook_port = env::var("WEBHOOK_PORT").ok().and_then(|s| s.parse().ok());
+        // Empty strings count as unset (e.g. `-e WEBHOOK_CERT=` to disable a
+        // value that would otherwise come from `.env`).
+        let webhook_cert = env::var("WEBHOOK_CERT").ok().filter(|s| !s.is_empty());
+        let webhook_secret_token = env::var("WEBHOOK_SECRET_TOKEN")
+            .ok()
+            .filter(|s| !s.is_empty());
 
         Config {
             admin_ids,

crates/xmedia-bot/src/db.rs

@@ -0,0 +1,37 @@
+//! Shared SQLite plumbing for the three tables in `data/task_queue.db`
+//! (`tasks` in queue.rs, `chat_state` in state.rs, `link_cache` in
+//! link_cache.rs).
+//!
+//! Every operation opens its own short-lived connection with a busy timeout:
+//! handler tasks enqueue while workers lease/update rows concurrently, and
+//! without the timeout a concurrent write fails immediately with SQLITE_BUSY
+//! and the operation is lost. All I/O runs inside `spawn_blocking` via
+//! [`with_conn`] — rusqlite connections are not Send-friendly to hold across
+//! an await point, and blocking the async executor stalls every handler.
+
+use rusqlite::Connection;
+use std::time::Duration;
+
+/// Opens the shared DB with a busy timeout.
+pub fn open_db(path: &str) -> rusqlite::Result<Connection> {
+    let conn = Connection::open(path)?;
+    conn.busy_timeout(Duration::from_secs(5))?;
+    Ok(conn)
+}
+
+/// Runs `f` against a fresh connection on a blocking thread, returning the
+/// closure's result. Owns the `spawn_blocking` + `expect` ceremony shared by
+/// every table access; the caller maps errors to its own log line.
+pub async fn with_conn<T, F>(path: &str, f: F) -> rusqlite::Result<T>
+where
+    T: Send + 'static,
+    F: FnOnce(&mut Connection) -> rusqlite::Result<T> + Send + 'static,
+{
+    let path = path.to_string();
+    tokio::task::spawn_blocking(move || {
+        let mut conn = open_db(&path)?;
+        f(&mut conn)
+    })
+    .await
+    .expect("db worker panicked")
+}

crates/xmedia-bot/src/handlers.rs

@@ -5,20 +5,19 @@ use crate::send::{self, MediaItemPayload, Task};
 use crate::state::{ChatData, ChatStore, unix_now};
 use std::collections::HashSet;
 use std::sync::LazyLock;
+use teloxide::RequestError;
 use teloxide::prelude::*;
-use tokio::sync::Semaphore;
 use teloxide::types::{
     CallbackQuery, ChatAction, ChatId, ChatKind, InlineQuery, InlineQueryResult,
     InlineQueryResultMpeg4Gif, InlineQueryResultPhoto, InlineQueryResultVideo, Message,
     MessageEntityKind, MessageId, ParseMode, Recipient, ReplyParameters,
 };
 use teloxide::utils::command::BotCommands;
-use teloxide::RequestError;
+use tokio::sync::Semaphore;
 use x_media::media::Media;
 
-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::open("data/task_queue.db").expect("failed to open chat store"));
 pub static TASK_QUEUE: LazyLock<PersistentTaskQueue> =
     LazyLock::new(|| PersistentTaskQueue::new("data/task_queue.db"));
 pub static LINK_CACHE: LazyLock<LinkCache> =
@@ -34,24 +33,38 @@ pub static CONFIG: LazyLock<Config> = LazyLock::new(Config::load);
 static URL_TASKS: LazyLock<Semaphore> = LazyLock::new(|| Semaphore::new(8));
 
 #[derive(BotCommands, Clone)]
-#[command(rename_rule = "snake_case", description = "")]
+#[command(
+    rename_rule = "snake_case",
+    description = "Turn X/Pixiv/Bluesky links into media messages"
+)]
 enum Command {
-    #[command(description = "")]
+    #[command(description = "Get started")]
     Start,
-    #[command(description = "")]
+    #[command(description = "Show command help")]
     Help,
-    #[command(description = "", parse_with = "split")]
+    #[command(
+        description = "Set forward channel (@channel or ID)",
+        parse_with = "split"
+    )]
     SetForwardChannel(String),
-    #[command(description = "")]
+    #[command(description = "Remove forward channel")]
     RemoveForwardChannel,
-    #[command(description = "")]
+    #[command(description = "Toggle edit-before-forward")]
     EditBeforeForward,
-    #[command(description = "", parse_with = "split")]
+    #[command(
+        description = "Reply with [] to save as template",
+        parse_with = "split"
+    )]
     SetTemplate(String),
-    #[command(description = "")]
+    #[command(description = "Show chat state (debug)")]
     BotDict,
-    #[command(description = "", parse_with = "split")]
+    #[command(description = "Set site caption format", parse_with = "split")]
     SetFormat(String),
+    #[command(
+        description = "Clear link cache (admin; optional URL, else all)",
+        parse_with = "split"
+    )]
+    ClearCache(String),
 }
 
 async fn reply<T>(bot: Bot, message: Message, text: T) -> Result<Message, RequestError>
@@ -197,7 +210,11 @@ async fn set_forward_channel_handler(
     Ok(channel_id)
 }
 
-async fn execute_command(bot: &Bot, message: &Message, command: Command) -> Result<(), RequestError> {
+async fn execute_command(
+    bot: &Bot,
+    message: &Message,
+    command: Command,
+) -> Result<(), RequestError> {
     match command {
         Command::Start => {
             bot.send_message(message.chat.id, "Hello!").await?;
@@ -215,7 +232,8 @@ async fn execute_command(bot: &Bot, message: &Message, command: Command) -> Resu
                     "Add successfully.".to_string()
                 }
                 Err(SetForwardChannelError::EmptyParameter) => {
-                    "Receive empty parameter.\nYou should enter a channel id or username".to_string()
+                    "Receive empty parameter.\nYou should enter a channel id or username"
+                        .to_string()
                 }
                 Err(SetForwardChannelError::NotChannel) => {
                     "Given id / username is not a channel".to_string()
@@ -292,7 +310,9 @@ async fn execute_command(bot: &Bot, message: &Message, command: Command) -> Resu
         Command::SetFormat(arg) => {
             let chat_id = message.chat.id.0;
             let (site, format) = match arg.split_once(char::is_whitespace) {
-                Some((site, format)) if !format.trim().is_empty() => (site.trim(), format.trim().to_string()),
+                Some((site, format)) if !format.trim().is_empty() => {
+                    (site.trim(), format.trim().to_string())
+                }
                 _ => {
                     reply(
                         bot.clone(),
@@ -317,10 +337,62 @@ async fn execute_command(bot: &Bot, message: &Message, command: Command) -> Resu
             CHAT_STORE.set(chat_id, &chat_data).await;
             reply(bot.clone(), message.clone(), "Format set.").await?;
         }
+        Command::ClearCache(arg) => {
+            let sender_id = message
+                .from
+                .as_ref()
+                .map(|user| user.id.0 as i64)
+                .unwrap_or(-1);
+            if !CONFIG.admin_ids.contains(&sender_id) {
+                reply(bot.clone(), message.clone(), "Admin only.").await?;
+                return Ok(());
+            }
+            let arg = arg.trim();
+            if arg.is_empty() {
+                let removed = LINK_CACHE.clear(None).await;
+                log::info!("cache cleared by {sender_id}: {removed} entries");
+                reply(
+                    bot.clone(),
+                    message.clone(),
+                    format!("Cleared {removed} cached entr{}.", plural(removed)),
+                )
+                .await?;
+            } else {
+                let key = match x_media::site::cache_key(arg) {
+                    Some(key) => key,
+                    None => {
+                        reply(
+                            bot.clone(),
+                            message.clone(),
+                            "Unrecognized link. Use a twitter/x, pixiv or bsky post URL.",
+                        )
+                        .await?;
+                        return Ok(());
+                    }
+                };
+                let removed = LINK_CACHE.clear(Some(&key)).await;
+                log::info!("cache entry cleared by {sender_id}: {key} ({removed} rows)");
+                reply(
+                    bot.clone(),
+                    message.clone(),
+                    format!(
+                        "Cleared cache for {arg} ({} entr{}).",
+                        removed,
+                        plural(removed)
+                    ),
+                )
+                .await?;
+            }
+        }
     }
     Ok(())
 }
 
+/// `""` for one, `"ies"` for anything else — "1 entry" / "2 entries".
+fn plural(n: usize) -> &'static str {
+    if n == 1 { "" } else { "ies" }
+}
+
 /// For locally produced media (encoded ugoira MP4) the thumbnail URL is a
 /// hotlink-protected remote URL Telegram may not fetch; let Telegram generate
 /// its own thumbnail instead.
@@ -383,7 +455,10 @@ async fn dispatch_send(bot: Bot, message: &Message, task: &Task, url: &str) {
             log::info!("sent {} message(s) for {url}", message_ids.len());
             send::post_send_actions(&bot, task, message_ids).await;
         }
-        Err(send::SendError::Retryable { delay_seconds, task }) => {
+        Err(send::SendError::Retryable {
+            delay_seconds,
+            task,
+        }) => {
             log::info!("send for {url} failed, queued for retry in {delay_seconds:.1}s");
             enqueue_retry(task, delay_seconds).await;
             let _ = reply(bot, message.clone(), "Send failed. Task queued for retry.").await;
@@ -444,7 +519,10 @@ fn build_send_task(
 
 async fn url_media(bot: Bot, message: &Message, url: &str) {
     let chat_id = message.chat.id.0;
-    if let Err(e) = bot.send_chat_action(ChatId(chat_id), ChatAction::Typing).await {
+    if let Err(e) = bot
+        .send_chat_action(ChatId(chat_id), ChatAction::Typing)
+        .await
+    {
         log::error!("send_chat_action failed: {e}");
     }
 
@@ -520,7 +598,12 @@ async fn url_media(bot: Bot, message: &Message, url: &str) {
         // Retries exhausted: notify the user (Rust-only requirement 3).
         Err(e) => {
             log::error!("fetch {url}: {e}");
-            let _ = reply(bot, message.clone(), "Failed to fetch media from this link.").await;
+            let _ = reply(
+                bot,
+                message.clone(),
+                "Failed to fetch media from this link.",
+            )
+            .await;
         }
         Ok(Some(fetched)) => {
             if fetched.media.is_empty() {
@@ -542,8 +625,9 @@ async fn url_media(bot: Bot, message: &Message, url: &str) {
             let caption = fetched.caption_with(&format);
             // Raw render data for the link cache; the send fills in the
             // Telegram file ids and persists the entry.
-            let cache_data = fetched.render_fields().map(|(author, author_url, title, tags)| {
-                CachedPost {
+            let cache_data = fetched
+                .render_fields()
+                .map(|(author, author_url, title, tags)| CachedPost {
                     url: fetched.source_url.clone(),
                     caption: fetched.caption.clone(),
                     title: title.to_string(),
@@ -552,8 +636,7 @@ async fn url_media(bot: Bot, message: &Message, url: &str) {
                     tags: tags.to_string(),
                     sensitive: fetched.sensitive,
                     media: vec![],
-                }
-            });
+                });
             let items: Vec<MediaItemPayload> = fetched
                 .media
                 .iter()
@@ -583,7 +666,10 @@ pub async fn message_handler(bot: Bot, message: Message) -> Result<(), RequestEr
         .text()
         .map(|t| if t.len() > 120 { &t[..120] } else { t })
         .unwrap_or("<no text>");
-    log::info!("message from {sender} in {} (private={is_private}): {text_preview}", message.chat.id);
+    log::info!(
+        "message from {sender} in {} (private={is_private}): {text_preview}",
+        message.chat.id
+    );
     // URL/edit flows only run in private chats; commands run in any chat.
     if is_private && edit_message_handler(&bot, &message).await {
         return respond(());
@@ -653,8 +739,8 @@ pub async fn inline_query_handler(bot: Bot, query: InlineQuery) -> Result<(), Re
                             thumbnail,
                             fetched.title.clone(),
                         )
-                            .caption(caption)
-                            .parse_mode(ParseMode::Html),
+                        .caption(caption)
+                        .parse_mode(ParseMode::Html),
                     ),
                     Media::Animated { .. } => InlineQueryResult::Mpeg4Gif(
                         InlineQueryResultMpeg4Gif::new(id, url, thumbnail)
@@ -686,7 +772,10 @@ pub async fn callback_query_handler(bot: Bot, query: CallbackQuery) -> Result<()
     let mut 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!("callback from {}: no edit record for prompt {prompt_message_id}", chat_id);
+        log::info!(
+            "callback from {}: no edit record for prompt {prompt_message_id}",
+            chat_id
+        );
         bot.answer_callback_query(callback_query_id)
             .text("Expired")
             .await?;
@@ -705,7 +794,10 @@ pub async fn callback_query_handler(bot: Bot, query: CallbackQuery) -> Result<()
     let Some(data) = data else {
         return respond(());
     };
-    log::info!("callback from {} on prompt {prompt_message_id}: {data}", chat_id);
+    log::info!(
+        "callback from {} on prompt {prompt_message_id}: {data}",
+        chat_id
+    );
     if data == "forward" {
         match chat_data.forward_channel_id {
             Some(channel_id) => {
@@ -731,7 +823,10 @@ pub async fn callback_query_handler(bot: Bot, query: CallbackQuery) -> Result<()
                         chat_data.edit_message.remove(&prompt_message_id);
                         CHAT_STORE.set(chat_id, &chat_data).await;
                     }
-                    Err(send::SendError::Retryable { delay_seconds, task }) => {
+                    Err(send::SendError::Retryable {
+                        delay_seconds,
+                        task,
+                    }) => {
                         log::info!("forward queued for retry in {delay_seconds:.1}s");
                         enqueue_retry(task, delay_seconds).await;
                         bot.answer_callback_query(callback_query_id)

crates/xmedia-bot/src/link_cache.rs

@@ -8,7 +8,7 @@
 //! [`Config::link_cache_ttl`]; a stale entry is dropped lazily on read and
 //! by the periodic prune in `main`.
 
-use rusqlite::{params, Connection};
+use rusqlite::{Connection, params};
 use serde::{Deserialize, Serialize};
 use std::time::Duration;
 
@@ -49,12 +49,6 @@ pub struct LinkCache {
     db_path: String,
 }
 
-fn open_db(path: &str) -> rusqlite::Result<Connection> {
-    let conn = Connection::open(path)?;
-    conn.busy_timeout(Duration::from_secs(5))?;
-    Ok(conn)
-}
-
 impl LinkCache {
     pub fn open(db_path: &str) -> Self {
         if let Ok(conn) = Connection::open(db_path)
@@ -73,11 +67,9 @@ impl LinkCache {
     /// Returns the cached post if present and not expired; a stale entry is
     /// removed on the spot.
     pub async fn get(&self, key: &str, ttl: Duration) -> Option<CachedPost> {
-        let db_path = self.db_path.clone();
         let key = key.to_string();
         let ttl = ttl.as_secs_f64();
-        tokio::task::spawn_blocking(move || -> rusqlite::Result<Option<CachedPost>> {
-            let conn = open_db(&db_path)?;
+        let result = crate::db::with_conn(&self.db_path, move |conn| {
             let mut stmt =
                 conn.prepare("SELECT payload, created_at FROM link_cache WHERE url = ?1")?;
             let mut rows = stmt.query(params![key])?;
@@ -90,66 +82,84 @@ impl LinkCache {
                 conn.execute("DELETE FROM link_cache WHERE url = ?1", params![key])?;
                 return Ok(None);
             }
-            serde_json::from_str(&payload).map(Some).map_err(|e| {
-                rusqlite::Error::ToSqlConversionFailure(Box::new(e))
-            })
-        })
-        .await
-        .expect("link cache read worker panicked")
-        .unwrap_or_else(|e| {
-            log::error!("link cache read failed: {e}");
-            None
+            Ok(Some(serde_json::from_str::<CachedPost>(&payload).map_err(
+                |e| rusqlite::Error::ToSqlConversionFailure(Box::new(e)),
+            )?))
         })
+        .await;
+        match result {
+            Ok(v) => v,
+            Err(e) => {
+                log::error!("link cache read failed: {e}");
+                None
+            }
+        }
     }
 
     pub async fn put(&self, key: &str, post: &CachedPost) {
-        let db_path = self.db_path.clone();
         let key = key.to_string();
         let payload = serde_json::to_string(post).expect("cached post serializes");
-        tokio::task::spawn_blocking(move || -> rusqlite::Result<()> {
-            let conn = open_db(&db_path)?;
+        let result = crate::db::with_conn(&self.db_path, move |conn| {
             conn.execute(
                 "INSERT OR REPLACE INTO link_cache (url, payload, created_at) VALUES (?1, ?2, ?3)",
                 params![key, payload, now_f64()],
             )?;
             Ok(())
         })
-        .await
-        .expect("link cache write worker panicked")
-        .unwrap_or_else(|e| log::error!("link cache write failed: {e}"));
+        .await;
+        if let Err(e) = result {
+            log::error!("link cache write failed: {e}");
+        }
     }
 
     /// Drops an entry (e.g. a cached file id that turned out invalid).
     pub async fn remove(&self, key: &str) {
-        let db_path = self.db_path.clone();
         let key = key.to_string();
-        tokio::task::spawn_blocking(move || -> rusqlite::Result<()> {
-            let conn = open_db(&db_path)?;
+        let result = crate::db::with_conn(&self.db_path, move |conn| {
             conn.execute("DELETE FROM link_cache WHERE url = ?1", params![key])?;
             Ok(())
         })
-        .await
-        .expect("link cache delete worker panicked")
-        .unwrap_or_else(|e| log::error!("link cache delete failed: {e}"));
+        .await;
+        if let Err(e) = result {
+            log::error!("link cache delete failed: {e}");
+        }
     }
 
     /// Removes expired entries; returns how many were deleted.
     pub async fn prune(&self, ttl: Duration) -> usize {
-        let db_path = self.db_path.clone();
         let cutoff = now_f64() - ttl.as_secs_f64();
-        tokio::task::spawn_blocking(move || -> rusqlite::Result<usize> {
-            let conn = open_db(&db_path)?;
+        let result = crate::db::with_conn(&self.db_path, move |conn| {
             conn.execute(
                 "DELETE FROM link_cache WHERE created_at < ?1",
                 params![cutoff],
             )
         })
-        .await
-        .expect("link cache prune worker panicked")
-        .unwrap_or_else(|e| {
-            log::error!("link cache prune failed: {e}");
-            0
+        .await;
+        match result {
+            Ok(n) => n,
+            Err(e) => {
+                log::error!("link cache prune failed: {e}");
+                0
+            }
+        }
+    }
+
+    /// Deletes one entry (by normalized cache key) or the whole cache when
+    /// `key` is `None`. Returns how many rows were removed.
+    pub async fn clear(&self, key: Option<&str>) -> usize {
+        let key = key.map(str::to_string);
+        let result = crate::db::with_conn(&self.db_path, move |conn| match &key {
+            Some(key) => conn.execute("DELETE FROM link_cache WHERE url = ?1", params![key]),
+            None => conn.execute("DELETE FROM link_cache", []),
         })
+        .await;
+        match result {
+            Ok(n) => n,
+            Err(e) => {
+                log::error!("link cache clear failed: {e}");
+                0
+            }
+        }
     }
 }
 
@@ -200,14 +210,21 @@ mod tests {
         // Force the row into the past so a 1s TTL expires it.
         {
             let conn = Connection::open(dir.path().join("c.db")).unwrap();
-            conn.execute(
-                "UPDATE link_cache SET created_at = created_at - 100",
-                [],
-            )
-            .unwrap();
+            conn.execute("UPDATE link_cache SET created_at = created_at - 100", [])
+                .unwrap();
         }
-        assert!(cache.get("twitter:1", Duration::from_secs(1)).await.is_none());
-        assert!(cache.get("twitter:1", Duration::from_secs(3600)).await.is_none());
+        assert!(
+            cache
+                .get("twitter:1", Duration::from_secs(1))
+                .await
+                .is_none()
+        );
+        assert!(
+            cache
+                .get("twitter:1", Duration::from_secs(3600))
+                .await
+                .is_none()
+        );
     }
 
     #[tokio::test]
@@ -217,14 +234,60 @@ mod tests {
         cache.put("twitter:1", &entry()).await;
         cache.put("pixiv:2", &entry()).await;
         cache.remove("twitter:1").await;
-        assert!(cache.get("twitter:1", Duration::from_secs(3600)).await.is_none());
-        assert!(cache.get("pixiv:2", Duration::from_secs(3600)).await.is_some());
+        assert!(
+            cache
+                .get("twitter:1", Duration::from_secs(3600))
+                .await
+                .is_none()
+        );
+        assert!(
+            cache
+                .get("pixiv:2", Duration::from_secs(3600))
+                .await
+                .is_some()
+        );
         {
             let conn = Connection::open(dir.path().join("c.db")).unwrap();
             conn.execute("UPDATE link_cache SET created_at = created_at - 100", [])
                 .unwrap();
         }
         assert_eq!(cache.prune(Duration::from_secs(1)).await, 1);
-        assert!(cache.get("pixiv:2", Duration::from_secs(3600)).await.is_none());
+        assert!(
+            cache
+                .get("pixiv:2", Duration::from_secs(3600))
+                .await
+                .is_none()
+        );
+    }
+
+    #[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());
+        cache.put("twitter:1", &entry()).await;
+        cache.put("pixiv:2", &entry()).await;
+        // By key: only the matching row is removed.
+        assert_eq!(cache.clear(Some("twitter:1")).await, 1);
+        assert!(
+            cache
+                .get("twitter:1", Duration::from_secs(3600))
+                .await
+                .is_none()
+        );
+        assert!(
+            cache
+                .get("pixiv:2", Duration::from_secs(3600))
+                .await
+                .is_some()
+        );
+        // Whole cache: nothing left; removing an absent key deletes 0 rows.
+        assert_eq!(cache.clear(None).await, 1);
+        assert!(
+            cache
+                .get("pixiv:2", Duration::from_secs(3600))
+                .await
+                .is_none()
+        );
+        assert_eq!(cache.clear(None).await, 0);
     }
 }

crates/xmedia-bot/src/main.rs

@@ -1,13 +1,14 @@
 use dotenv::dotenv;
 use teloxide::dptree::endpoint;
+use teloxide::prelude::*;
 use teloxide::stop::StopToken;
 use teloxide::types::{ChatId, InputFile, MessageId};
-use teloxide::update_listeners::{self, webhooks, UpdateListener};
-use teloxide::prelude::*;
+use teloxide::update_listeners::{self, UpdateListener, webhooks};
 use tokio::sync::watch;
 use x_media::site;
 
 mod config;
+mod db;
 mod handlers;
 mod link_cache;
 mod photo;
@@ -74,7 +75,10 @@ async fn main() {
     }
 
     // Edit-expiry sweep: clears the prompt's buttons once the record expires.
-    log::info!("edit-expiry sweep: every 300s, ttl {}", CONFIG.edit_message_ttl.as_secs());
+    log::info!(
+        "edit-expiry sweep: every 300s, ttl {}",
+        CONFIG.edit_message_ttl.as_secs()
+    );
     let (stop_tx, stop_rx) = watch::channel(false);
     {
         let bot = bot.clone();
@@ -95,7 +99,10 @@ async fn main() {
                     // If the prompt was already deleted, this fails with a
                     // 400 "message to edit not found" — log and ignore.
                     if let Err(e) = bot
-                        .edit_message_reply_markup(ChatId(chat_id), MessageId(prompt_message_id as i32))
+                        .edit_message_reply_markup(
+                            ChatId(chat_id),
+                            MessageId(prompt_message_id as i32),
+                        )
                         .await
                     {
                         log::info!("edit-expiry sweep: prompt message gone: {e}");
@@ -117,10 +124,7 @@ async fn main() {
 
     if CONFIG.webhook_enabled {
         log::info!("running in webhook mode");
-        let url = CONFIG
-            .webhook_url
-            .clone()
-            .expect("WEBHOOK_URL is not set");
+        let url = CONFIG.webhook_url.clone().expect("WEBHOOK_URL is not set");
         // `webhooks::axum` calls set_webhook itself (with the full options,
         // secret token included) — no explicit registration here.
         let listen = CONFIG.webhook_listen.expect("WEBHOOK_LISTEN is not set");

crates/xmedia-bot/src/photo.rs

@@ -216,13 +216,15 @@ fn target_dims(w: u32, h: u32) -> (u32, u32) {
 /// RGB, Lanczos-downscale beyond the dimension cap, encode PNG — a PNG still
 /// over the upload cap afterwards becomes JPEG.
 fn prepare_png(file: NamedTempFile, bytes: Vec<u8>) -> Result<PhotoPrep, String> {
-    let (w, h, _bit_depth, color_type) =
-        parse_png_header(&bytes).ok_or("invalid PNG header")?;
+    let (w, h, _bit_depth, color_type) = parse_png_header(&bytes).ok_or("invalid PNG header")?;
     let size_over = bytes.len() as u64 > MAX_UPLOAD_BYTES;
     if w + h <= PHOTO_MAX_DIMENSION_SUM && !size_over {
         return Ok(PhotoPrep::Upload(file));
     }
-    log::info!("photo {w}x{h} ({_bit_depth:?} {color_type:?}, {} bytes) needs processing", bytes.len());
+    log::info!(
+        "photo {w}x{h} ({_bit_depth:?} {color_type:?}, {} bytes) needs processing",
+        bytes.len()
+    );
 
     let channels = output_channels(color_type);
     if (w as u64) * (h as u64) * channels as u64 > MAX_DECODE_BYTES {
@@ -238,7 +240,9 @@ fn prepare_png(file: NamedTempFile, bytes: Vec<u8>) -> Result<PhotoPrep, String>
     };
     let mut decoder = png::Decoder::new(std::io::Cursor::new(&bytes));
     decoder.set_transformations(transforms);
-    let mut reader = decoder.read_info().map_err(|e| format!("png decode: {e}"))?;
+    let mut reader = decoder
+        .read_info()
+        .map_err(|e| format!("png decode: {e}"))?;
     let out_w = reader.info().width;
     let out_h = reader.info().height;
     let mut buf = vec![
@@ -457,7 +461,9 @@ mod tests {
         let mut bytes = Vec::new();
         {
             let encoder = jpeg_encoder::Encoder::new(&mut bytes, 90);
-            encoder.encode(&rgb, w, h, jpeg_encoder::ColorType::Rgb).unwrap();
+            encoder
+                .encode(&rgb, w, h, jpeg_encoder::ColorType::Rgb)
+                .unwrap();
         }
         let mut file = tempfile::Builder::new().suffix(".jpg").tempfile().unwrap();
         std::io::Write::write_all(file.as_file_mut(), &bytes).unwrap();
@@ -485,7 +491,9 @@ mod tests {
         for y in 0..h {
             for x in 0..w {
                 let base = (x + y) * 255 / (w + h);
-                rng = rng.wrapping_mul(6364136223846793005).wrapping_add(1442695040888963407);
+                rng = rng
+                    .wrapping_mul(6364136223846793005)
+                    .wrapping_add(1442695040888963407);
                 let n = ((rng >> 33) % 11) as i32 - 5; // noise in [-5, 5]
                 let v = (base as i32 + n).clamp(0, 255) as u8;
                 data.extend_from_slice(&[v, v, v]);
@@ -499,7 +507,11 @@ mod tests {
             let mut writer = encoder.write_header().unwrap();
             writer.write_image_data(&data).unwrap();
         }
-        assert!(bytes.len() as u64 > MAX_UPLOAD_BYTES, "test needs a >10MiB PNG, got {}", bytes.len());
+        assert!(
+            bytes.len() as u64 > MAX_UPLOAD_BYTES,
+            "test needs a >10MiB PNG, got {}",
+            bytes.len()
+        );
 
         let mut file = tempfile::Builder::new().suffix(".png").tempfile().unwrap();
         std::io::Write::write_all(file.as_file_mut(), &bytes).unwrap();

crates/xmedia-bot/src/queue.rs

@@ -6,11 +6,11 @@
 //! replaced by dedicated columns.
 
 use parking_lot::Mutex;
-use rusqlite::{params, Connection, TransactionBehavior};
+use rusqlite::{Connection, TransactionBehavior, params};
 use serde_json::Value;
 use std::pin::Pin;
-use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
 use std::sync::Arc;
+use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
 use std::time::{Duration, SystemTime, UNIX_EPOCH};
 use tokio::sync::Notify;
 use tokio::task::JoinHandle;
@@ -29,15 +29,9 @@ const QUEUE_WORKERS: usize = 4;
 pub enum QueueError {
     /// Reschedule with the given delay; after `MAX_RETRIES` attempts the task
     /// is dead-lettered instead.
-    Retryable {
-        delay_seconds: f64,
-        payload: Value,
-    },
+    Retryable { delay_seconds: f64, payload: Value },
     /// Give up now.
-    Permanent {
-        message: String,
-        payload: Value,
-    },
+    Permanent { message: String, payload: Value },
 }
 
 type BoxFuture<'a, T> = Pin<Box<dyn Future<Output = T> + Send + 'a>>;
@@ -74,16 +68,7 @@ fn now_f64() -> f64 {
         .unwrap_or(0.0)
 }
 
-/// Opens the queue DB with a busy timeout. Handler tasks enqueue while
-/// workers lease/update rows concurrently; without the timeout a concurrent
-/// write fails immediately with SQLITE_BUSY and the operation is lost.
-fn open_db(path: &str) -> rusqlite::Result<Connection> {
-    let conn = Connection::open(path)?;
-    conn.busy_timeout(Duration::from_secs(5))?;
-    Ok(conn)
-}
-
-fn ensure_schema(conn: &Connection) -> rusqlite::Result<()> {
+fn ensure_schema(conn: &rusqlite::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, \
@@ -162,10 +147,8 @@ impl PersistentTaskQueue {
             self.counter.fetch_add(1, Ordering::Relaxed)
         );
         let payload = payload.to_string();
-        let db_path = self.db_path.clone();
         log::info!("enqueued {id} (run_after {run_after:.1})");
-        tokio::task::spawn_blocking(move || -> rusqlite::Result<()> {
-            let conn = open_db(&db_path)?;
+        crate::db::with_conn(&self.db_path, move |conn| {
             conn.execute(
                 "INSERT OR REPLACE INTO tasks (id, payload, run_after, attempts, status, locked_until, created_at) \
                  VALUES (?1, ?2, ?3, 0, 'pending', 0, ?4)",
@@ -173,8 +156,7 @@ impl PersistentTaskQueue {
             )?;
             Ok(())
         })
-        .await
-        .expect("queue insert worker panicked")?;
+        .await?;
         // Wake every sleeping worker: with several workers the one that finds
         // nothing due must not starve the newly inserted row.
         self.notify.notify_waiters();
@@ -182,18 +164,17 @@ impl PersistentTaskQueue {
     }
 
     async fn recover_stale(&self) {
-        let db_path = self.db_path.clone();
-        tokio::task::spawn_blocking(move || -> rusqlite::Result<()> {
-            let conn = open_db(&db_path)?;
+        let result = crate::db::with_conn(&self.db_path, move |conn| {
             conn.execute(
                 "UPDATE tasks SET status='pending', locked_until=0 WHERE status='in_progress' AND locked_until < ?1",
                 params![now_f64()],
             )?;
             Ok(())
         })
-        .await
-        .expect("queue recovery worker panicked")
-        .unwrap_or_else(|e| log::error!("queue recovery failed: {e}"));
+        .await;
+        if let Err(e) = result {
+            log::error!("queue recovery failed: {e}");
+        }
     }
 }
 
@@ -225,9 +206,7 @@ impl QueueWorker {
 
     /// Leases the oldest due row (sets it `in_progress` with a lock TTL).
     async fn lease_next(&self) -> Option<LeasedRow> {
-        let db_path = self.db_path.clone();
-        tokio::task::spawn_blocking(move || -> rusqlite::Result<Option<LeasedRow>> {
-            let mut conn = open_db(&db_path)?;
+        let result = crate::db::with_conn(&self.db_path, |conn| {
             // BEGIN IMMEDIATE: with several workers, a deferred transaction
             // that read before another worker's lease commit would fail with
             // SQLITE_BUSY_SNAPSHOT. Taking the write lock up front serializes
@@ -265,31 +244,34 @@ impl QueueWorker {
                 attempts,
             }))
         })
-        .await
-        .expect("queue lease worker panicked")
-        .unwrap_or_else(|e| {
-            log::error!("queue lease failed: {e}");
-            None
-        })
+        .await;
+        match result {
+            Ok(row) => row,
+            Err(e) => {
+                log::error!("queue lease failed: {e}");
+                None
+            }
+        }
     }
 
     async fn earliest_run_after(&self) -> Option<f64> {
-        let db_path = self.db_path.clone();
-        tokio::task::spawn_blocking(move || -> rusqlite::Result<Option<f64>> {
-            let conn = open_db(&db_path)?;
-            let mut stmt = conn.prepare("SELECT MIN(run_after) FROM tasks WHERE status='pending'")?;
+        let result = crate::db::with_conn(&self.db_path, |conn| {
+            let mut stmt =
+                conn.prepare("SELECT MIN(run_after) FROM tasks WHERE status='pending'")?;
             let mut rows = stmt.query([])?;
             match rows.next()? {
                 Some(row) => Ok(row.get::<_, Option<f64>>(0)?),
                 None => Ok(None),
             }
         })
-        .await
-        .expect("queue timing worker panicked")
-        .unwrap_or_else(|e| {
-            log::error!("queue timing query failed: {e}");
-            None
-        })
+        .await;
+        match result {
+            Ok(v) => v,
+            Err(e) => {
+                log::error!("queue timing query failed: {e}");
+                None
+            }
+        }
     }
 
     async fn process(&self, row: LeasedRow) {
@@ -336,33 +318,31 @@ impl QueueWorker {
     }
 
     async fn delete_row(&self, id: &str) {
-        let db_path = self.db_path.clone();
         let id = id.to_string();
-        tokio::task::spawn_blocking(move || -> rusqlite::Result<()> {
-            let conn = open_db(&db_path)?;
+        let result = crate::db::with_conn(&self.db_path, move |conn| {
             conn.execute("DELETE FROM tasks WHERE id = ?1", params![id])?;
             Ok(())
         })
-        .await
-        .expect("queue delete worker panicked")
-        .unwrap_or_else(|e| log::error!("queue delete failed: {e}"));
+        .await;
+        if let Err(e) = result {
+            log::error!("queue delete failed: {e}");
+        }
     }
 
     async fn reschedule(&self, id: &str, payload: Value, delay_seconds: f64, attempts: i32) {
-        let db_path = self.db_path.clone();
         let id = id.to_string();
         let payload = payload.to_string();
-        tokio::task::spawn_blocking(move || -> rusqlite::Result<()> {
-            let conn = open_db(&db_path)?;
+        let result = crate::db::with_conn(&self.db_path, move |conn| {
             conn.execute(
                 "UPDATE tasks SET payload=?1, run_after=?2, attempts=?3, status='pending', locked_until=0 WHERE id=?4",
                 params![payload, now_f64() + delay_seconds, attempts, id],
             )?;
             Ok(())
         })
-        .await
-        .expect("queue reschedule worker panicked")
-        .unwrap_or_else(|e| log::error!("queue reschedule failed: {e}"));
+        .await;
+        if let Err(e) = result {
+            log::error!("queue reschedule failed: {e}");
+        }
         self.notify.notify_waiters();
     }
 }

crates/xmedia-bot/src/send.rs

@@ -5,20 +5,19 @@
 
 use crate::handlers::{CHAT_STORE, LINK_CACHE, TASK_QUEUE};
 use crate::link_cache::{CachedMedia, CachedMediaKind, CachedPost};
-use crate::photo::{self, PhotoPrep, MAX_UPLOAD_BYTES};
+use crate::photo::{self, MAX_UPLOAD_BYTES, PhotoPrep};
 use crate::queue::QueueError;
 use crate::state::{EditMessage, unix_now};
 use rand::Rng;
 use serde::{Deserialize, Serialize};
 use std::collections::HashMap;
-use tempfile::NamedTempFile;
 use teloxide::prelude::*;
 use teloxide::types::{
-    ChatId, InlineKeyboardButton, InlineKeyboardMarkup, InputFile, InputMedia,
-    InputMediaAnimation, InputMediaPhoto, InputMediaVideo, Message, MessageId, ParseMode,
-    ReplyParameters,
+    ChatId, InlineKeyboardButton, InlineKeyboardMarkup, InputFile, InputMedia, InputMediaAnimation,
+    InputMediaPhoto, InputMediaVideo, Message, MessageId, ParseMode, ReplyParameters,
 };
 use teloxide::{ApiError, RequestError};
+use tempfile::NamedTempFile;
 use x_media::site::FetchError;
 
 #[derive(Serialize, Deserialize, Clone, Debug)]
@@ -111,16 +110,18 @@ pub enum Task {
 impl Task {
     fn cache_data(&self) -> Option<&CachedPost> {
         match self {
-            Task::SendMediaSequence { cache_data, .. }
-            | Task::SendAnimation { cache_data, .. } => cache_data.as_ref(),
+            Task::SendMediaSequence { cache_data, .. } | Task::SendAnimation { cache_data, .. } => {
+                cache_data.as_ref()
+            }
             Task::ForwardMessages { .. } => None,
         }
     }
 
     fn source_url(&self) -> Option<&str> {
         match self {
-            Task::SendMediaSequence { source_url, .. }
-            | Task::SendAnimation { source_url, .. } => Some(source_url),
+            Task::SendMediaSequence { source_url, .. } | Task::SendAnimation { source_url, .. } => {
+                Some(source_url)
+            }
             Task::ForwardMessages { .. } => None,
         }
     }
@@ -138,9 +139,10 @@ fn file_id_of_message(message: &Message, item: &MediaItemPayload) -> Option<Stri
     match item {
         // `photo()` returns all sizes, smallest first — the largest carries
         // the file id of the sent media.
-        MediaItemPayload::Photo { .. } => {
-            message.photo().and_then(|sizes| sizes.last()).map(|p| p.file.id.to_string())
-        }
+        MediaItemPayload::Photo { .. } => message
+            .photo()
+            .and_then(|sizes| sizes.last())
+            .map(|p| p.file.id.to_string()),
         MediaItemPayload::Video { .. } => message.video().map(|v| v.file.id.to_string()),
         MediaItemPayload::Animation { .. } => message.animation().map(|a| a.file.id.to_string()),
     }
@@ -214,7 +216,10 @@ pub const MAX_MEDIA_GROUP: usize = 9;
 
 /// Splits media into batches of at most [`MAX_MEDIA_GROUP`] items.
 pub fn chunk_media_items<T: Clone>(items: Vec<T>) -> Vec<Vec<T>> {
-    items.chunks(MAX_MEDIA_GROUP).map(|chunk| chunk.to_vec()).collect()
+    items
+        .chunks(MAX_MEDIA_GROUP)
+        .map(|chunk| chunk.to_vec())
+        .collect()
 }
 
 /// Exponential backoff with jitter, capped at 30s.
@@ -250,31 +255,41 @@ pub fn is_size_error(e: &ApiError) -> bool {
         return true;
     }
     let description = e.to_string().to_lowercase();
-    ["too large", "too big"].iter().any(|marker| description.contains(marker))
+    ["too large", "too big"]
+        .iter()
+        .any(|marker| description.contains(marker))
 }
 
 /// Task-free classification of a Telegram request error. The callers attach
 /// the (updated) task when building a [`SendError`].
 pub enum Classification {
-    Retryable { delay_seconds: f64 },
-    Permanent { message: String },
+    Retryable {
+        delay_seconds: f64,
+    },
+    Permanent {
+        message: String,
+    },
     /// Handled by the download fallback, not a queue retry.
     MediaFetchFailure,
 }
 
 pub fn classify_request_error(e: &RequestError) -> Classification {
     match e {
-        RequestError::RetryAfter(seconds) => {
-            Classification::Retryable { delay_seconds: seconds.seconds() as f64 }
-        }
+        RequestError::RetryAfter(seconds) => Classification::Retryable {
+            delay_seconds: seconds.seconds() as f64,
+        },
         RequestError::Network(_) => Classification::Retryable {
             delay_seconds: retry_delay_seconds(0),
         },
         RequestError::Api(api) if is_media_fetch_failure(api) => Classification::MediaFetchFailure,
-        RequestError::Api(api) => Classification::Permanent { message: api.to_string() },
+        RequestError::Api(api) => Classification::Permanent {
+            message: api.to_string(),
+        },
         RequestError::MigrateToChatId(_)
         | RequestError::InvalidJson { .. }
-        | RequestError::Io(_) => Classification::Permanent { message: e.to_string() },
+        | RequestError::Io(_) => Classification::Permanent {
+            message: e.to_string(),
+        },
     }
 }
 
@@ -390,9 +405,9 @@ fn build_media_group(
         .map(|(i, item)| {
             let item_caption = if i == 0 { caption } else { None };
             Ok(match item {
-                MediaItemPayload::Photo {
-                    has_spoiler, ..
-                } => photo_media(item.input_file()?, item_caption, *has_spoiler),
+                MediaItemPayload::Photo { has_spoiler, .. } => {
+                    photo_media(item.input_file()?, item_caption, *has_spoiler)
+                }
                 MediaItemPayload::Video {
                     has_spoiler,
                     thumbnail,
@@ -404,9 +419,9 @@ fn build_media_group(
                     }
                     video
                 }
-                MediaItemPayload::Animation {
-                    has_spoiler, ..
-                } => animation_media(item.input_file()?, item_caption, *has_spoiler),
+                MediaItemPayload::Animation { has_spoiler, .. } => {
+                    animation_media(item.input_file()?, item_caption, *has_spoiler)
+                }
             })
         })
         .collect()
@@ -431,8 +446,12 @@ fn sniff_ext(bytes: &[u8]) -> &'static str {
 }
 
 enum FallbackError {
-    Retryable { delay_seconds: f64 },
-    Permanent { message: String },
+    Retryable {
+        delay_seconds: f64,
+    },
+    Permanent {
+        message: String,
+    },
     /// The downloaded file exceeds the upload cap; the caller falls back to
     /// the item's smaller URL.
     MediaTooLarge,
@@ -468,9 +487,7 @@ async fn download_to_temp(item: &MediaItemPayload) -> Result<NamedTempFile, Fall
     };
     // Photos are downloaded even over the cap so `prepare_photo` can
     // downscale / transcode them; only videos/animations short-circuit.
-    if !matches!(item, MediaItemPayload::Photo { .. })
-        && bytes.len() as u64 > MAX_UPLOAD_BYTES
-    {
+    if !matches!(item, MediaItemPayload::Photo { .. }) && bytes.len() as u64 > MAX_UPLOAD_BYTES {
         return Err(FallbackError::MediaTooLarge);
     }
     let ext = sniff_ext(&bytes);
@@ -628,14 +645,16 @@ async fn send_batch_via_upload(
     }
     let result = bot
         .send_media_group(ChatId(chat_id), items)
-        .reply_parameters(ReplyParameters::new(MessageId(reply_to as i32)).allow_sending_without_reply())
+        .reply_parameters(
+            ReplyParameters::new(MessageId(reply_to as i32)).allow_sending_without_reply(),
+        )
         .await;
     match result {
         Ok(messages) => Ok(messages),
         Err(e) => Err(match classify_request_error(&e) {
-            Classification::Retryable { delay_seconds } => FallbackError::Retryable {
-                delay_seconds,
-            },
+            Classification::Retryable { delay_seconds } => {
+                FallbackError::Retryable { delay_seconds }
+            }
             Classification::Permanent { message } => FallbackError::Permanent { message },
             Classification::MediaFetchFailure => FallbackError::Permanent {
                 message: "upload failed".into(),
@@ -702,7 +721,11 @@ pub async fn send_media_sequence(bot: &Bot, task: &Task) -> Result<Vec<i64>, Sen
     let fresh_send = *batch_index == 0 && sent.is_empty();
     for idx in *batch_index..media_batches.len() {
         let batch = &media_batches[idx];
-        let caption = if idx == 0 { Some(caption.as_str()) } else { None };
+        let caption = if idx == 0 {
+            Some(caption.as_str())
+        } else {
+            None
+        };
         let items = match build_media_group(batch, caption) {
             Ok(items) => items,
             Err(message) => {
@@ -714,7 +737,9 @@ pub async fn send_media_sequence(bot: &Bot, task: &Task) -> Result<Vec<i64>, Sen
         };
         match bot
             .send_media_group(ChatId(chat_id), items)
-            .reply_parameters(ReplyParameters::new(MessageId(reply_to as i32)).allow_sending_without_reply())
+            .reply_parameters(
+                ReplyParameters::new(MessageId(reply_to as i32)).allow_sending_without_reply(),
+            )
             .await
         {
             Ok(messages) => {
@@ -726,9 +751,7 @@ pub async fn send_media_sequence(bot: &Bot, task: &Task) -> Result<Vec<i64>, Sen
                 collect_file_ids(&messages, batch, &mut cached_media);
                 sent.extend(messages.into_iter().map(|m| m.id.0 as i64));
             }
-            Err(RequestError::Api(api))
-                if is_media_fetch_failure(&api) || is_size_error(&api) =>
-            {
+            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("?")
@@ -779,7 +802,9 @@ async fn send_animation_inner(
         .send_animation(ChatId(chat_id), file)
         .caption(caption)
         .parse_mode(ParseMode::Html)
-        .reply_parameters(ReplyParameters::new(MessageId(reply_to as i32)).allow_sending_without_reply());
+        .reply_parameters(
+            ReplyParameters::new(MessageId(reply_to as i32)).allow_sending_without_reply(),
+        );
     if spoiler {
         request = request.has_spoiler(true);
     }
@@ -802,9 +827,7 @@ pub async fn send_animation(bot: &Bot, task: &Task) -> Result<Vec<i64>, SendErro
     let reply_to = *reply_to_message_id;
     let (media_url, has_spoiler) = match animation {
         MediaItemPayload::Animation {
-            media,
-            has_spoiler,
-            ..
+            media, has_spoiler, ..
         } => (media, *has_spoiler),
         MediaItemPayload::Photo { .. } | MediaItemPayload::Video { .. } => {
             unreachable!("SendAnimation carries an Animation payload")
@@ -812,19 +835,20 @@ pub async fn send_animation(bot: &Bot, task: &Task) -> Result<Vec<i64>, SendErro
     };
     let url_file = match input_file_for(media_url) {
         Ok(file) => file,
-        Err(message) => return Err(SendError::Permanent { message, task: task.clone() }),
+        Err(message) => {
+            return Err(SendError::Permanent {
+                message,
+                task: task.clone(),
+            });
+        }
     };
-    match send_animation_inner(bot, chat_id, reply_to, caption, has_spoiler, url_file)
-        .await
-    {
+    match send_animation_inner(bot, chat_id, reply_to, caption, has_spoiler, url_file).await {
         Ok(message) => {
             let id = message.id.0 as i64;
             cache_animation_send(task, &message).await;
             Ok(vec![id])
         }
-        Err(RequestError::Api(api))
-            if is_media_fetch_failure(&api) || is_size_error(&api) =>
-        {
+        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
@@ -872,21 +896,24 @@ pub async fn send_animation(bot: &Bot, task: &Task) -> Result<Vec<i64>, SendErro
                                 Err(e) => Err(classify_to_send_error(&e, task.clone())),
                             }
                         }
-                        Err(message) => {
-                            Err(SendError::Permanent { message, task: 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(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(classify_to_send_error(&e, task.clone())),
@@ -910,7 +937,11 @@ pub async fn forward_messages(bot: &Bot, task: &Task) -> Result<(), SendError> {
         .map(|id| MessageId(*id as i32))
         .collect::<Vec<_>>();
     match bot
-        .copy_messages(ChatId(*to_chat_id), ChatId(*from_chat_id), message_ids.clone())
+        .copy_messages(
+            ChatId(*to_chat_id),
+            ChatId(*from_chat_id),
+            message_ids.clone(),
+        )
         .await
     {
         Ok(_) => {
@@ -944,12 +975,18 @@ pub fn build_edit_markup(templates: &HashMap<String, String>) -> InlineKeyboardM
 
 /// Notifies a chat about a dead-lettered task (skips when `notify_chat_id` is
 /// absent).
-pub async fn notify_failure(bot: &Bot, chat_id: Option<i64>, message_id: Option<i64>, message: &str) {
+pub async fn notify_failure(
+    bot: &Bot,
+    chat_id: Option<i64>,
+    message_id: Option<i64>,
+    message: &str,
+) {
     let Some(chat_id) = chat_id else { return };
     let mut request = bot.send_message(ChatId(chat_id), message);
     if let Some(message_id) = message_id {
-        request = request
-            .reply_parameters(ReplyParameters::new(MessageId(message_id as i32)).allow_sending_without_reply());
+        request = request.reply_parameters(
+            ReplyParameters::new(MessageId(message_id as i32)).allow_sending_without_reply(),
+        );
     }
     if let Err(e) = request.await {
         log::error!("failed to notify about failed task: {e}");
@@ -959,38 +996,45 @@ pub async fn notify_failure(bot: &Bot, chat_id: Option<i64>, message_id: Option<
 /// After a successful send: either open the edit-before-forward prompt or
 /// forward to the configured channel (with retry/queue handling).
 pub async fn post_send_actions(bot: &Bot, task: &Task, message_ids: Vec<i64>) {
-    let (chat_id, reply_to, source_url, edit_before_forward, forward_channel_id, notify_chat_id, notify_message_id) =
-        match task {
-            Task::SendMediaSequence {
-                chat_id,
-                reply_to_message_id,
-                source_url,
-                edit_before_forward,
-                forward_channel_id,
-                notify_chat_id,
-                notify_message_id,
-                ..
-            }
-            | Task::SendAnimation {
-                chat_id,
-                reply_to_message_id,
-                source_url,
-                edit_before_forward,
-                forward_channel_id,
-                notify_chat_id,
-                notify_message_id,
-                ..
-            } => (
-                *chat_id,
-                *reply_to_message_id,
-                source_url.clone(),
-                *edit_before_forward,
-                *forward_channel_id,
-                *notify_chat_id,
-                *notify_message_id,
-            ),
-            Task::ForwardMessages { .. } => return,
-        };
+    let (
+        chat_id,
+        reply_to,
+        source_url,
+        edit_before_forward,
+        forward_channel_id,
+        notify_chat_id,
+        notify_message_id,
+    ) = match task {
+        Task::SendMediaSequence {
+            chat_id,
+            reply_to_message_id,
+            source_url,
+            edit_before_forward,
+            forward_channel_id,
+            notify_chat_id,
+            notify_message_id,
+            ..
+        }
+        | Task::SendAnimation {
+            chat_id,
+            reply_to_message_id,
+            source_url,
+            edit_before_forward,
+            forward_channel_id,
+            notify_chat_id,
+            notify_message_id,
+            ..
+        } => (
+            *chat_id,
+            *reply_to_message_id,
+            source_url.clone(),
+            *edit_before_forward,
+            *forward_channel_id,
+            *notify_chat_id,
+            *notify_message_id,
+        ),
+        Task::ForwardMessages { .. } => return,
+    };
 
     if edit_before_forward {
         let mut chat_data = CHAT_STORE.get(chat_id).await;
@@ -1027,7 +1071,10 @@ pub async fn post_send_actions(bot: &Bot, task: &Task, message_ids: Vec<i64>) {
     }
 
     if let Some(channel_id) = forward_channel_id {
-        log::info!("forwarding {} message(s) to channel {channel_id}", message_ids.len());
+        log::info!(
+            "forwarding {} message(s) to channel {channel_id}",
+            message_ids.len()
+        );
         let forward_task = Task::ForwardMessages {
             from_chat_id: chat_id,
             to_chat_id: channel_id,
@@ -1037,7 +1084,10 @@ pub async fn post_send_actions(bot: &Bot, task: &Task, message_ids: Vec<i64>) {
         };
         match forward_messages(bot, &forward_task).await {
             Ok(()) => {}
-            Err(SendError::Retryable { delay_seconds, task }) => {
+            Err(SendError::Retryable {
+                delay_seconds,
+                task,
+            }) => {
                 let payload = serde_json::to_value(task).expect("task serializes");
                 let run_after = std::time::SystemTime::now()
                     .duration_since(std::time::UNIX_EPOCH)
@@ -1077,7 +1127,10 @@ pub async fn handle_task(payload: serde_json::Value) -> Result<(), QueueError> {
         Task::SendMediaSequence { .. } | Task::SendAnimation { .. } => {
             let message_ids = match send_media_or_animation(&bot, &task).await {
                 Ok(ids) => ids,
-                Err(SendError::Retryable { delay_seconds, task }) => {
+                Err(SendError::Retryable {
+                    delay_seconds,
+                    task,
+                }) => {
                     return Err(QueueError::Retryable {
                         delay_seconds,
                         payload: serde_json::to_value(task).expect("task serializes"),
@@ -1096,7 +1149,10 @@ pub async fn handle_task(payload: serde_json::Value) -> Result<(), QueueError> {
         }
         Task::ForwardMessages { .. } => match forward_messages(&bot, &task).await {
             Ok(()) => Ok(()),
-            Err(SendError::Retryable { delay_seconds, task }) => Err(QueueError::Retryable {
+            Err(SendError::Retryable {
+                delay_seconds,
+                task,
+            }) => Err(QueueError::Retryable {
                 delay_seconds,
                 payload: serde_json::to_value(task).expect("task serializes"),
             }),
@@ -1151,7 +1207,11 @@ mod tests {
         assert_eq!(chunk_media_items((0..10).collect()).len(), 2);
         assert_eq!(chunk_media_items((0..25).collect()).len(), 3);
         assert_eq!(chunk_media_items((0..25).collect())[2].len(), 7);
-        assert!(chunk_media_items((0..25).collect()).iter().all(|c| c.len() <= 9));
+        assert!(
+            chunk_media_items((0..25).collect())
+                .iter()
+                .all(|c| c.len() <= 9)
+        );
     }
 
     #[test]
@@ -1175,7 +1235,10 @@ mod tests {
             let api = ApiError::Unknown(description.to_string());
             assert!(is_media_fetch_failure(&api), "{description}");
         }
-        for description in ["Bad Request: message is not modified", "Forbidden: bot was blocked by the user"] {
+        for description in [
+            "Bad Request: message is not modified",
+            "Forbidden: bot was blocked by the user",
+        ] {
             let api = ApiError::Unknown(description.to_string());
             assert!(!is_media_fetch_failure(&api), "{description}");
         }
@@ -1196,7 +1259,10 @@ mod tests {
             assert!(is_size_error(&api), "{description}");
         }
         // Unrelated errors must not match.
-        for description in ["Bad Request: WEBPAGE_MEDIA_EMPTY", "Bad Request: message is not modified"] {
+        for description in [
+            "Bad Request: WEBPAGE_MEDIA_EMPTY",
+            "Bad Request: message is not modified",
+        ] {
             let api = ApiError::Unknown(description.to_string());
             assert!(!is_size_error(&api), "{description}");
         }
@@ -1205,9 +1271,16 @@ mod tests {
     #[test]
     fn media_item_payload_fallback_url_serde_default() {
         // Old queued payloads without the field deserialize with None.
-        let json = serde_json::json!({"kind": "photo", "media": "https://a/b.jpg", "has_spoiler": false});
+        let json =
+            serde_json::json!({"kind": "photo", "media": "https://a/b.jpg", "has_spoiler": false});
         let photo: MediaItemPayload = serde_json::from_value(json).unwrap();
-        assert!(matches!(photo, MediaItemPayload::Photo { fallback_url: None, .. }));
+        assert!(matches!(
+            photo,
+            MediaItemPayload::Photo {
+                fallback_url: None,
+                ..
+            }
+        ));
         assert_eq!(photo.fallback_url(), None);
     }
 
@@ -1286,7 +1359,13 @@ mod tests {
                 assert_eq!(sent_message_ids, vec![11, 12]);
                 assert_eq!(forward_channel_id, Some(333));
                 assert_eq!(media_batches.len(), 2);
-                assert!(matches!(media_batches[0][0], MediaItemPayload::Photo { has_spoiler: true, .. }));
+                assert!(matches!(
+                    media_batches[0][0],
+                    MediaItemPayload::Photo {
+                        has_spoiler: true,
+                        ..
+                    }
+                ));
             }
             other => panic!("expected SendMediaSequence, got {other:?}"),
         }

crates/xmedia-bot/src/state.rs

@@ -2,7 +2,7 @@
 //! `data/task_queue.db`, shared with the task queue).
 
 use parking_lot::Mutex;
-use rusqlite::{params, Connection};
+use rusqlite::params;
 use serde::{Deserialize, Serialize};
 use std::collections::HashMap;
 use std::path::Path;
@@ -45,7 +45,9 @@ pub fn unix_now() -> i64 {
 }
 
 impl ChatStore {
-    /// Creates the parent directory and both tables (idempotent).
+    /// 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()
@@ -53,12 +55,9 @@ impl ChatStore {
             std::fs::create_dir_all(parent)
                 .map_err(|e| rusqlite::Error::ToSqlConversionFailure(Box::new(e)))?;
         }
-        let conn = Connection::open(path)?;
+        let conn = crate::db::open_db(path)?;
         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 TABLE IF NOT EXISTS chat_state (chat_id TEXT PRIMARY KEY, payload TEXT NOT NULL);",
+            "CREATE TABLE IF NOT EXISTS chat_state (chat_id TEXT PRIMARY KEY, payload TEXT NOT NULL);",
         )?;
         drop(conn);
         Ok(ChatStore {
@@ -71,22 +70,19 @@ impl ChatStore {
         if let Some(data) = self.cache.lock().get(&chat_id) {
             return data.clone();
         }
-        let db_path = self.db_path.clone();
-        let payload = tokio::task::spawn_blocking(move || -> rusqlite::Result<Option<String>> {
-            let conn = Connection::open(&db_path)?;
+        let chat_key = chat_id.to_string();
+        let payload = crate::db::with_conn(&self.db_path, move |conn| {
             // Concurrent handler tasks (batch-forwards) may write chat_state
-            // while this read runs; without a busy timeout a write lock
-            // collision fails the query immediately.
-            conn.busy_timeout(std::time::Duration::from_secs(5))?;
+            // while this read runs; the shared busy timeout handles the
+            // write-lock collision instead of failing the query.
             let mut stmt = conn.prepare("SELECT payload FROM chat_state WHERE chat_id = ?1")?;
-            let mut rows = stmt.query(params![chat_id.to_string()])?;
+            let mut rows = stmt.query(params![chat_key])?;
             match rows.next()? {
-                Some(row) => Ok(Some(row.get(0)?)),
+                Some(row) => Ok(Some(row.get::<_, String>(0)?)),
                 None => Ok(None),
             }
         })
         .await
-        .expect("chat_state worker panicked")
         .unwrap_or_else(|e| {
             log::error!("chat_state read failed: {e}");
             None
@@ -101,19 +97,18 @@ impl ChatStore {
     pub async fn set(&self, chat_id: i64, data: &ChatData) {
         self.cache.lock().insert(chat_id, data.clone());
         let payload = serde_json::to_string(data).expect("chat state serializes");
-        let db_path = self.db_path.clone();
-        tokio::task::spawn_blocking(move || -> rusqlite::Result<()> {
-            let conn = Connection::open(&db_path)?;
-            conn.busy_timeout(std::time::Duration::from_secs(5))?;
+        let chat_id = chat_id.to_string();
+        let result = crate::db::with_conn(&self.db_path, move |conn| {
             conn.execute(
                 "INSERT OR REPLACE INTO chat_state (chat_id, payload) VALUES (?1, ?2)",
-                params![chat_id.to_string(), payload],
+                params![chat_id, payload],
             )?;
             Ok(())
         })
-        .await
-        .expect("chat_state worker panicked")
-        .unwrap_or_else(|e| log::error!("chat_state write failed: {e}"));
+        .await;
+        if let Err(e) = result {
+            log::error!("chat_state write failed: {e}");
+        }
     }
 
     /// Removes edit-before-forward records whose `created_at + ttl` is in the
@@ -149,7 +144,10 @@ impl ChatStore {
             self.set(chat_id, &data).await;
         }
         if !removed.is_empty() {
-            log::info!("pruned {} expired edit-before-forward record(s)", removed.len());
+            log::info!(
+                "pruned {} expired edit-before-forward record(s)",
+                removed.len()
+            );
         }
         removed
     }

docker-compose.yml.example

@@ -11,14 +11,14 @@ services:
       - html:/usr/share/nginx/html:ro
     networks: [proxy]
     labels:
-      - 'com.github.nginx-proxy.nginx=true'
+      - 'com.github.nginx-proxy.nginx'
     container_name: nginx-proxy
 
   acme-companion:
     image: nginxproxy/acme-companion
     restart: always
     environment:
-      DEFAULT_EMAIL: 'admin@yoursfunny.top'
+      DEFAULT_EMAIL: ''
     volumes:
       - /var/run/docker.sock:/var/run/docker.sock:ro
       - certs:/etc/nginx/certs:rw