From ae69d729300f9e3b1a6cacb5fb5d05fa6e0c8df8 Mon Sep 17 00:00:00 2001 From: YoursFunny Date: Fri, 14 Aug 2026 23:38:41 +0800 Subject: [PATCH] refactor(handlers): inject AppContext into url_media; cover the full URL pipeline --- crates/xmedia-bot/src/handlers/callback.rs | 4 +- crates/xmedia-bot/src/handlers/commands.rs | 39 ++-- crates/xmedia-bot/src/handlers/mod.rs | 20 +- crates/xmedia-bot/src/handlers/urls.rs | 248 ++++++++++++++++++--- crates/xmedia-bot/src/media_sender.rs | 44 +++- crates/xmedia-bot/src/send.rs | 9 +- 6 files changed, 297 insertions(+), 67 deletions(-) diff --git a/crates/xmedia-bot/src/handlers/callback.rs b/crates/xmedia-bot/src/handlers/callback.rs index c8c6fbd..0bbda7c 100644 --- a/crates/xmedia-bot/src/handlers/callback.rs +++ b/crates/xmedia-bot/src/handlers/callback.rs @@ -2,7 +2,7 @@ //! "template|" buttons. use super::urls::enqueue_retry; -use super::{CHAT_STORE, CONFIG}; +use super::{CHAT_STORE, CONFIG, TASK_QUEUE}; use crate::send::{self, Task}; use crate::state::unix_now; use teloxide::RequestError; @@ -83,7 +83,7 @@ pub async fn callback_query_handler(bot: Bot, query: CallbackQuery) -> Result<() task, }) => { log::info!("forward queued for retry in {delay_seconds:.1}s"); - enqueue_retry(task, delay_seconds).await; + enqueue_retry(&TASK_QUEUE, task, delay_seconds).await; bot.answer_callback_query(callback_query_id) .text("Forward queued for retry.") .await?; diff --git a/crates/xmedia-bot/src/handlers/commands.rs b/crates/xmedia-bot/src/handlers/commands.rs index 947f955..377d14c 100644 --- a/crates/xmedia-bot/src/handlers/commands.rs +++ b/crates/xmedia-bot/src/handlers/commands.rs @@ -153,7 +153,7 @@ pub(crate) async fn execute_command( "Bot can't post messages to the channel".to_string() } }; - reply(bot.clone(), message.clone(), result).await?; + reply(bot, message.chat.id.0, message.id, result).await?; } Command::RemoveForwardChannel => { let chat_id = message.chat.id.0; @@ -167,7 +167,7 @@ pub(crate) async fn execute_command( } }) .await; - reply(bot.clone(), message.clone(), text).await?; + reply(bot, message.chat.id.0, message.id, text).await?; } Command::EditBeforeForward => { let chat_id = message.chat.id.0; @@ -185,7 +185,7 @@ pub(crate) async fn execute_command( } }) .await; - reply(bot.clone(), message.clone(), text).await?; + reply(bot, message.chat.id.0, message.id, text).await?; } Command::SetTemplate(name) => { let chat_id = message.chat.id.0; @@ -210,13 +210,13 @@ pub(crate) async fn execute_command( } } }; - reply(bot.clone(), message.clone(), text).await?; + reply(bot, message.chat.id.0, message.id, text).await?; } Command::BotDict => { let chat_data = CHAT_STORE.get(message.chat.id.0).await; let debug = format!("{chat_data:?}"); let text = html_escape::encode_text(&debug).into_owned(); - reply(bot.clone(), message.clone(), text).await?; + reply(bot, message.chat.id.0, message.id, text).await?; } Command::SetFormat(arg) => { let chat_id = message.chat.id.0; @@ -226,8 +226,9 @@ pub(crate) async fn execute_command( } _ => { reply( - bot.clone(), - message.clone(), + bot, + message.chat.id.0, + message.id, "Usage: /set_format ", ) .await?; @@ -236,8 +237,9 @@ pub(crate) async fn execute_command( }; if !x_media::site::site_ids().contains(&site) { reply( - bot.clone(), - message.clone(), + bot, + message.chat.id.0, + message.id, "Unknown site. Use twitter, bsky or pixiv.", ) .await?; @@ -248,7 +250,7 @@ pub(crate) async fn execute_command( data.message_format.insert(site.to_string(), format); }) .await; - reply(bot.clone(), message.clone(), "Format set.").await?; + reply(bot, message.chat.id.0, message.id, "Format set.").await?; } Command::ClearCache(arg) => { let sender_id = message @@ -257,7 +259,7 @@ pub(crate) async fn execute_command( .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?; + reply(bot, message.chat.id.0, message.id, "Admin only.").await?; return Ok(()); } let arg = arg.trim(); @@ -265,8 +267,9 @@ pub(crate) async fn execute_command( let removed = LINK_CACHE.clear(None).await; log::info!("cache cleared by {sender_id}: {removed} entries"); reply( - bot.clone(), - message.clone(), + bot, + message.chat.id.0, + message.id, format!("Cleared {removed} cached entr{}.", plural(removed)), ) .await?; @@ -275,8 +278,9 @@ pub(crate) async fn execute_command( Some(key) => key, None => { reply( - bot.clone(), - message.clone(), + bot, + message.chat.id.0, + message.id, "Unrecognized link. Use a twitter/x, pixiv or bsky post URL.", ) .await?; @@ -286,8 +290,9 @@ pub(crate) async fn execute_command( let removed = LINK_CACHE.clear(Some(&key)).await; log::info!("cache entry cleared by {sender_id}: {key} ({removed} rows)"); reply( - bot.clone(), - message.clone(), + bot, + message.chat.id.0, + message.id, format!( "Cleared cache for {arg} ({} entr{}).", removed, diff --git a/crates/xmedia-bot/src/handlers/mod.rs b/crates/xmedia-bot/src/handlers/mod.rs index 60fcff2..30038d9 100644 --- a/crates/xmedia-bot/src/handlers/mod.rs +++ b/crates/xmedia-bot/src/handlers/mod.rs @@ -18,21 +18,27 @@ pub use inline::inline_query_handler; pub use statics::{CHAT_STORE, CONFIG, LINK_CACHE, TASK_QUEUE}; pub use urls::{start_url_workers, stop_url_workers}; +use crate::media_sender::MediaSender; use commands::{Command, execute_command}; use teloxide::RequestError; use teloxide::prelude::*; -use teloxide::types::{ChatId, ChatKind, Message, MessageId, ParseMode, ReplyParameters}; +use teloxide::types::{ChatId, ChatKind, Message, MessageId, ParseMode}; use teloxide::utils::command::BotCommands; use urls::{URL_JOBS, extract_urls}; -/// Reply to a message, keeping the reply decoration even if the original was -/// already deleted. -pub(crate) async fn reply(bot: Bot, message: Message, text: T) -> Result +/// Reply to a message by id, keeping the reply decoration even if the +/// original was already deleted. +pub(crate) async fn reply( + sender: &dyn MediaSender, + chat_id: i64, + reply_to: MessageId, + text: T, +) -> Result where T: Into, { - bot.send_message(message.chat.id, text) - .reply_parameters(ReplyParameters::new(message.id).allow_sending_without_reply()) + sender + .send_message(ChatId(chat_id), text.into(), Some(reply_to), None) .await } @@ -134,7 +140,7 @@ pub async fn message_handler(bot: Bot, message: Message) -> Result<(), RequestEr log::warn!("url workers not started; dropping link"); break; }; - let _ = tx.send((bot.clone(), message.clone(), url)).await; + let _ = tx.send((message.clone(), url)).await; } } respond(()) diff --git a/crates/xmedia-bot/src/handlers/urls.rs b/crates/xmedia-bot/src/handlers/urls.rs index 269f7b1..5a06062 100644 --- a/crates/xmedia-bot/src/handlers/urls.rs +++ b/crates/xmedia-bot/src/handlers/urls.rs @@ -2,18 +2,21 @@ //! worker pool, link-cache fast path, fetch, task build and send dispatch. use super::{CHAT_STORE, CONFIG, LINK_CACHE, TASK_QUEUE, log_key, reply}; +use crate::config::Config; use crate::db::now_f64; -use crate::link_cache::{CachedMediaKind, CachedPost}; +use crate::link_cache::{CachedMediaKind, CachedPost, LinkCache}; +use crate::media_sender::MediaSender; +use crate::queue::PersistentTaskQueue; use crate::send::{self, MediaItemPayload, Task}; -use crate::state::ChatData; +use crate::state::{ChatData, ChatStore}; use std::collections::HashSet; use std::sync::LazyLock; -use teloxide::prelude::*; -use teloxide::types::{ChatAction, ChatId, Message, MessageEntityKind}; +use teloxide::types::{ChatAction, ChatId, Message, MessageEntityKind, MessageId}; use x_media::media::Media; -/// One URL job: bot handle + the message + the extracted URL. -type UrlJob = (Bot, Message, String); +/// One URL job: the message + the extracted URL (the sender and stores come +/// from the shared [`AppContext`], assembled from statics inside the worker). +type UrlJob = (Message, String); /// Bounded channel of URL jobs drained by [`start_url_workers`]. The bound /// caps both queued memory and shutdown backlog; a full channel applies /// backpressure to the per-chat handler instead of spawning unbounded tasks. @@ -32,6 +35,27 @@ static URL_WORKER_HANDLES: LazyLock { + pub sender: &'a dyn MediaSender, + pub chat_store: &'a ChatStore, + pub task_queue: &'a PersistentTaskQueue, + pub link_cache: &'a LinkCache, + pub config: &'a Config, +} + +/// Assembles the production context from the process-wide statics. +fn app_context() -> AppContext<'static> { + AppContext { + sender: &*crate::send::BOT, + chat_store: &CHAT_STORE, + task_queue: &TASK_QUEUE, + link_cache: &LINK_CACHE, + config: &CONFIG, + } +} + /// Starts the URL job workers (called once from main after the queue starts). /// teloxide dispatches updates to a per-chat worker that handles them /// sequentially, so a batch-forward of many messages would otherwise be @@ -46,10 +70,13 @@ pub async fn start_url_workers() { for _ in 0..URL_WORKERS { let rx = std::sync::Arc::clone(&rx); handles.push(tokio::spawn(async move { + let ctx = app_context(); while !URL_STOP.load(std::sync::atomic::Ordering::Relaxed) { let job = rx.lock().await.recv().await; match job { - Some((bot, message, url)) => url_media(bot, &message, &url).await, + Some((message, url)) => { + url_media(&ctx, message.chat.id.0, message.id.0 as i64, &url).await + } None => break, } } @@ -142,10 +169,10 @@ fn media_to_payload(media: &Media, sensitive: bool) -> MediaItemPayload { } } -pub(crate) async fn enqueue_retry(task: Task, delay_seconds: f64) { +pub(crate) async fn enqueue_retry(queue: &PersistentTaskQueue, task: Task, delay_seconds: f64) { let payload = serde_json::to_value(task).expect("task serializes"); let run_after = now_f64() + delay_seconds; - if let Err(e) = TASK_QUEUE.enqueue(payload, run_after).await { + if let Err(e) = queue.enqueue(payload, run_after).await { log::error!("failed to enqueue retry: {e}"); } } @@ -153,10 +180,16 @@ pub(crate) async fn enqueue_retry(task: Task, delay_seconds: f64) { /// Sends a task and handles the outcome: post-send actions on success, retry /// enqueue on retryable failure, reply + link-cache invalidation on /// permanent failure (a stale cached file id must not repeat forever). -async fn dispatch_send(bot: Bot, message: &Message, task: &Task, url: &str) { +async fn dispatch_send( + ctx: &AppContext<'_>, + chat_id: i64, + reply_to: MessageId, + task: &Task, + url: &str, +) { let result = match task { - Task::SendAnimation { .. } => send::send_animation(&bot, task).await, - Task::SendMediaSequence { .. } => send::send_media_sequence(&bot, task).await, + Task::SendAnimation { .. } => send::send_animation(ctx.sender, task).await, + Task::SendMediaSequence { .. } => send::send_media_sequence(ctx.sender, task).await, Task::ForwardMessages { .. } => unreachable!(), }; match result { @@ -166,7 +199,7 @@ async fn dispatch_send(bot: Bot, message: &Message, task: &Task, url: &str) { message_ids.len(), log_key(url) ); - send::post_send_actions(&bot, task, message_ids).await; + send::post_send_actions(ctx.sender, task, message_ids).await; // The task settled: drop any keep-alive temp media. send::release_keep_alive(task); } @@ -178,17 +211,29 @@ async fn dispatch_send(bot: Bot, message: &Message, task: &Task, url: &str) { "send for [key={}] failed, queued for retry in {delay_seconds:.1}s", log_key(url) ); - enqueue_retry(task, delay_seconds).await; - let _ = reply(bot, message.clone(), "Send failed. Task queued for retry.").await; + enqueue_retry(ctx.task_queue, task, delay_seconds).await; + let _ = reply( + ctx.sender, + chat_id, + reply_to, + "Send failed. Task queued for retry.", + ) + .await; } Err(send::SendError::Permanent { message: err_message, task, }) => { - send::invalidate_cache(&task).await; + send::invalidate_cache_with(ctx.link_cache, &task).await; send::release_keep_alive(&task); log::error!("send for {url} failed permanently: {err_message}"); - let _ = reply(bot, message.clone(), format!("Send failed: {err_message}")).await; + let _ = reply( + ctx.sender, + chat_id, + reply_to, + format!("Send failed: {err_message}"), + ) + .await; } } } @@ -198,30 +243,30 @@ async fn dispatch_send(bot: Bot, message: &Message, task: &Task, url: &str) { #[allow(clippy::too_many_arguments)] fn build_send_task( chat_data: &ChatData, - message: &Message, + chat_id: i64, + reply_to_message_id: i64, source_url: String, caption: String, items: Vec, cache_data: Option, ) -> Task { - let chat_id = message.chat.id.0; if items.len() == 1 && matches!(items[0], MediaItemPayload::Animation { .. }) { Task::SendAnimation { chat_id, - reply_to_message_id: message.id.0 as i64, + reply_to_message_id, caption, animation: items.into_iter().next().unwrap(), source_url, edit_before_forward: chat_data.edit_before_forward, forward_channel_id: chat_data.forward_channel_id, notify_chat_id: Some(chat_id), - notify_message_id: Some(message.id.0 as i64), + notify_message_id: Some(reply_to_message_id), cache_data, } } else { Task::SendMediaSequence { chat_id, - reply_to_message_id: message.id.0 as i64, + reply_to_message_id, caption, // Photos first so a mixed photo+video group starts with a photo // (Telegram's sendMediaGroup rule); order within each kind is kept. @@ -232,15 +277,16 @@ fn build_send_task( edit_before_forward: chat_data.edit_before_forward, forward_channel_id: chat_data.forward_channel_id, notify_chat_id: Some(chat_id), - notify_message_id: Some(message.id.0 as i64), + notify_message_id: Some(reply_to_message_id), cache_data, } } } -async fn url_media(bot: Bot, message: &Message, url: &str) { - let chat_id = message.chat.id.0; - if let Err(e) = bot +async fn url_media(ctx: &AppContext<'_>, chat_id: i64, reply_to_message_id: i64, url: &str) { + let reply_to = MessageId(reply_to_message_id as i32); + if let Err(e) = ctx + .sender .send_chat_action(ChatId(chat_id), ChatAction::Typing) .await { @@ -251,10 +297,10 @@ async fn url_media(bot: Bot, message: &Message, url: &str) { // no source-site request, no download, no upload. Keyed by the // normalized post id so x.com / fxtwitter / /photo/N variants collide. if let Some(key) = x_media::site::cache_key(url) - && let Some(cached) = LINK_CACHE.get(&key, CONFIG.link_cache_ttl).await + && let Some(cached) = ctx.link_cache.get(&key, ctx.config.link_cache_ttl).await { log::debug!("link cache hit for {key}"); - let chat_data = CHAT_STORE.get(chat_id).await; + let chat_data = ctx.chat_store.get(chat_id).await; // Cache keys are prefixed with the site id ("twitter:…"), matching // the value a fresh fetch would read from Fetched::site_id. let site = x_media::site::site_id_from_key(&key); @@ -302,13 +348,14 @@ async fn url_media(bot: Bot, message: &Message, url: &str) { .collect(); let task = build_send_task( &chat_data, - message, + chat_id, + reply_to_message_id, cached.url.clone(), caption, items, Some(cached), ); - dispatch_send(bot, message, &task, url).await; + dispatch_send(ctx, chat_id, reply_to, &task, url).await; return; } @@ -322,8 +369,9 @@ async fn url_media(bot: Bot, message: &Message, url: &str) { Err(e) => { log::error!("fetch {url}: {e}"); let _ = reply( - bot, - message.clone(), + ctx.sender, + chat_id, + reply_to, "Failed to fetch media from this link.", ) .await; @@ -331,14 +379,15 @@ async fn url_media(bot: Bot, message: &Message, url: &str) { Ok(Some(mut fetched)) => { if fetched.media.is_empty() { let _ = reply( - bot, - message.clone(), + ctx.sender, + chat_id, + reply_to, "No media found or media type is not supported.", ) .await; return; } - let chat_data = CHAT_STORE.get(chat_id).await; + let chat_data = ctx.chat_store.get(chat_id).await; // Per-site caption format override (empty -> built-in caption). let format = chat_data .message_format @@ -367,7 +416,8 @@ async fn url_media(bot: Bot, message: &Message, url: &str) { .collect(); let task = build_send_task( &chat_data, - message, + chat_id, + reply_to_message_id, fetched.source_url.clone(), caption, items, @@ -380,7 +430,131 @@ async fn url_media(bot: Bot, message: &Message, url: &str) { if let Some(dir) = fetched.take_keep_alive() { send::KEEP_ALIVE.lock().push(dir); } - dispatch_send(bot, message, &task, url).await; + dispatch_send(ctx, chat_id, reply_to, &task, url).await; } } } + +#[cfg(test)] +mod tests { + use super::*; + use crate::db; + use crate::link_cache::CachedMedia; + use crate::media_sender::test_support::{MockSender, Outcome}; + use std::sync::Arc; + use std::time::Duration; + use teloxide::{ApiError, RequestError}; + + fn permanent_error() -> RequestError { + RequestError::Api(ApiError::Unknown( + "Bad Request: message is not modified".into(), + )) + } + + fn cached_photo_entry() -> CachedPost { + CachedPost { + url: "https://x.com/u/status/1".into(), + caption: "cap".into(), + title: "t".into(), + author: "a".into(), + author_url: "au".into(), + tags: "".into(), + sensitive: false, + media: vec![CachedMedia { + kind: CachedMediaKind::Photo, + file_id: "file-1".into(), + }], + } + } + + #[tokio::test] + async fn cache_hit_sends_file_ids_and_invalidates_on_permanent_failure() { + let dir = tempfile::tempdir().unwrap(); + let pool = db::open_store(dir.path().join("t.db").to_str().unwrap()).unwrap(); + let chat_store = ChatStore::new(Arc::clone(&pool)); + let task_queue = PersistentTaskQueue::new(Arc::clone(&pool)); + let link_cache = LinkCache::new(Arc::clone(&pool)); + let config = Config::load(); + let sender = MockSender::scripted( + vec![Outcome::GroupErr, Outcome::MessageErr], + permanent_error, + ); + let ctx = AppContext { + sender: &sender, + chat_store: &chat_store, + task_queue: &task_queue, + link_cache: &link_cache, + config: &config, + }; + link_cache.put("twitter:1", &cached_photo_entry()).await; + + url_media(&ctx, 1, 2, "https://x.com/u/status/1").await; + + // The cached file id went out as a group send; the permanent failure + // then triggered the fire-and-forget reply (its mock error is fine). + assert_eq!( + sender.calls(), + vec!["send_chat_action", "send_media_group", "send_message"] + ); + // The stale cache entry was invalidated so the next request re-fetches. + assert!( + link_cache + .get("twitter:1", Duration::from_secs(3600)) + .await + .is_none() + ); + } + + #[tokio::test] + async fn cache_hit_success_keeps_the_cache_entry() { + let dir = tempfile::tempdir().unwrap(); + let pool = db::open_store(dir.path().join("t.db").to_str().unwrap()).unwrap(); + let chat_store = ChatStore::new(Arc::clone(&pool)); + let task_queue = PersistentTaskQueue::new(Arc::clone(&pool)); + let link_cache = LinkCache::new(Arc::clone(&pool)); + let config = Config::load(); + let sender = MockSender::scripted(vec![Outcome::GroupOk], permanent_error); + let ctx = AppContext { + sender: &sender, + chat_store: &chat_store, + task_queue: &task_queue, + link_cache: &link_cache, + config: &config, + }; + link_cache.put("twitter:1", &cached_photo_entry()).await; + + url_media(&ctx, 1, 2, "https://x.com/u/status/1").await; + + assert_eq!(sender.calls(), vec!["send_chat_action", "send_media_group"]); + // Success must not evict the entry. + assert!( + link_cache + .get("twitter:1", Duration::from_secs(3600)) + .await + .is_some() + ); + } + + #[tokio::test] + async fn unsupported_url_is_ignored_silently() { + let dir = tempfile::tempdir().unwrap(); + let pool = db::open_store(dir.path().join("t.db").to_str().unwrap()).unwrap(); + let chat_store = ChatStore::new(Arc::clone(&pool)); + let task_queue = PersistentTaskQueue::new(Arc::clone(&pool)); + let link_cache = LinkCache::new(Arc::clone(&pool)); + let config = Config::load(); + let sender = MockSender::scripted(vec![], permanent_error); + let ctx = AppContext { + sender: &sender, + chat_store: &chat_store, + task_queue: &task_queue, + link_cache: &link_cache, + config: &config, + }; + + // No cache key → the fetch dispatcher returns Ok(None) without any + // network; nothing is sent or replied. + url_media(&ctx, 1, 2, "https://example.com/not-a-post").await; + assert_eq!(sender.calls(), vec!["send_chat_action"]); + } +} diff --git a/crates/xmedia-bot/src/media_sender.rs b/crates/xmedia-bot/src/media_sender.rs index c95d3e2..d0b5725 100644 --- a/crates/xmedia-bot/src/media_sender.rs +++ b/crates/xmedia-bot/src/media_sender.rs @@ -8,7 +8,7 @@ use teloxide::RequestError; use teloxide::prelude::Requester; use teloxide::prelude::*; use teloxide::types::{ - ChatId, InlineKeyboardMarkup, InputFile, InputMedia, Message, MessageId, ParseMode, + ChatAction, ChatId, InlineKeyboardMarkup, InputFile, InputMedia, Message, MessageId, ParseMode, ReplyParameters, }; @@ -56,6 +56,13 @@ pub trait MediaSender: Send + Sync { reply_to: Option, reply_markup: Option, ) -> BoxFuture<'_, Result>; + + /// Sets the chat's "typing / uploading …" indicator (cosmetic). + fn send_chat_action( + &self, + chat_id: ChatId, + action: ChatAction, + ) -> BoxFuture<'_, Result<(), RequestError>>; } impl MediaSender for Bot { @@ -122,6 +129,20 @@ impl MediaSender for Bot { request.await }) } + + fn send_chat_action( + &self, + chat_id: ChatId, + action: ChatAction, + ) -> BoxFuture<'_, Result<(), RequestError>> { + Box::pin(async move { + // teloxide's `send_chat_action` returns `Result` (its + // unit marker type); map the success to `()`. + ::send_chat_action(self, chat_id, action) + .await + .map(|_| ()) + }) + } } /// Test support: a scripted [`MediaSender`] mock (no Telegram API involved). @@ -139,6 +160,9 @@ pub(crate) mod test_support { AnimationErr, CopyOk, CopyErr, + /// An error from `send_message` (replies are fire-and-forget, so an + /// error is fine for tests). + MessageErr, } /// Replays a script and records the method names that were called. @@ -241,7 +265,23 @@ pub(crate) mod test_support { _reply_to: Option, _reply_markup: Option, ) -> BoxFuture<'_, Result> { - Box::pin(async move { panic!("send_message should not be called in these tests") }) + Box::pin(async move { + match self.next("send_message") { + Outcome::MessageErr => Err(self.error()), + other => panic!("unexpected outcome {other:?} for send_message"), + } + }) + } + + fn send_chat_action( + &self, + _chat_id: ChatId, + _action: ChatAction, + ) -> BoxFuture<'_, Result<(), RequestError>> { + Box::pin(async move { + self.calls.lock().unwrap().push("send_chat_action"); + Ok(()) + }) } } } diff --git a/crates/xmedia-bot/src/send.rs b/crates/xmedia-bot/src/send.rs index 6cc3d47..9a1d7da 100644 --- a/crates/xmedia-bot/src/send.rs +++ b/crates/xmedia-bot/src/send.rs @@ -4,7 +4,7 @@ //! and uploads it via multipart). use crate::handlers::{CHAT_STORE, LINK_CACHE, TASK_QUEUE, log_key}; -use crate::link_cache::{CachedMedia, CachedMediaKind, CachedPost}; +use crate::link_cache::{CachedMedia, CachedMediaKind, CachedPost, LinkCache}; use crate::media_sender::MediaSender; use crate::photo::{self, MAX_UPLOAD_BYTES, PhotoPrep}; use crate::queue::QueueError; @@ -254,12 +254,17 @@ async fn cache_animation_send(task: &Task, message: &Message) { /// A cached Telegram file id failed permanently (stale/expired); drop the /// cache entry so the next request re-fetches instead of repeating it. pub async fn invalidate_cache(task: &Task) { + invalidate_cache_with(&LINK_CACHE, task).await; +} + +/// [`invalidate_cache`] against an injected cache (tests pass a tempdir one). +pub async fn invalidate_cache_with(cache: &LinkCache, task: &Task) { if task.is_cached_send() && let Some(url) = task.source_url() && let Some(key) = x_media::site::cache_key(url) { log::debug!("removing stale link cache entry for [key={}]", log_key(url)); - LINK_CACHE.remove(&key).await; + cache.remove(&key).await; } }